State APItravel.state.gov ↗

"""
NVC Timeframes API Client

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from datetime import datetime
from typing import Optional


class ParseClient:
    """Client for interacting with the Parse API."""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API client.

        Args:
            api_key: API key for authentication. Defaults to PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "98633553-78ac-4f4b-83e9-e03dc403acca"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError(
                "API key not provided. Set PARSE_API_KEY environment variable or pass api_key parameter."
            )

    def _call(self, endpoint: str, method: str = "POST", **params) -> dict:
        """
        Make a request to the Parse API.

        Args:
            endpoint: The endpoint name (e.g., 'get_nvc_timeframes')
            method: HTTP method ('GET' or 'POST')
            **params: Additional parameters for the request

        Returns:
            Response JSON as dictionary

        Raises:
            requests.RequestException: If the API request fails
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {"X-API-Key": self.api_key, "Content-Type": "application/json"}

        try:
            if method.upper() == "GET":
                response = requests.get(url, headers=headers, params=params)
            elif method.upper() == "POST":
                payload = params if params else {}
                response = requests.post(url, headers=headers, json=payload)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")

            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"API Error: {e}")
            raise

    def get_nvc_timeframes(self) -> dict:
        """
        Retrieve current NVC processing timeframes.

        Returns:
            Dictionary containing:
            - case_file_creation_time: Case creation processing timeframe
            - case_review_time: Case review processing timeframe
            - public_inquiry_form_response_time: Inquiry response timeframe

        Each timeframe contains:
            - full_text: Human-readable description
            - as_of_date: Date the information was last updated
            - activity_date: Date of cases/documents currently being processed
        """
        return self._call("get_nvc_timeframes", method="GET")


def parse_date(date_str: str) -> datetime:
    """
    Parse date string in format 'D-Mon-YY' to datetime object.

    Args:
        date_str: Date string (e.g., '8-Jun-26')

    Returns:
        datetime object
    """
    return datetime.strptime(date_str, "%d-%b-%y")


def calculate_processing_days(as_of_date: str, activity_date: str) -> int:
    """
    Calculate the number of days between as_of_date and activity_date.

    Args:
        as_of_date: Current date (when data was updated)
        activity_date: Date of cases being processed

    Returns:
        Number of days difference
    """
    as_of = parse_date(as_of_date)
    activity = parse_date(activity_date)
    return (as_of - activity).days


def main():
    """Main workflow demonstrating practical usage of NVC Timeframes API."""
    # Initialize client
    client = ParseClient()

    print("=" * 70)
    print("NVC VISA CENTER - PROCESSING TIMEFRAMES MONITOR")
    print("=" * 70)

    # Fetch current NVC timeframes
    print("\nFetching current NVC processing timeframes...")
    timeframes = client.get_nvc_timeframes()

    # Extract individual timeframe data
    case_creation = timeframes.get("case_file_creation_time", {})
    case_review = timeframes.get("case_review_time", {})
    inquiry_response = timeframes.get("public_inquiry_form_response_time", {})

    # Process and display case file creation timeframe
    print("\n" + "-" * 70)
    print("📋 CASE FILE CREATION TIMEFRAME")
    print("-" * 70)
    print(f"Status: {case_creation.get('full_text', 'N/A')}")
    as_of_date = case_creation.get("as_of_date")
    activity_date = case_creation.get("activity_date")
    if as_of_date and activity_date:
        days = calculate_processing_days(as_of_date, activity_date)
        print(f"Last Updated: {as_of_date}")
        print(f"Currently Processing Cases From: {activity_date}")
        print(f"Approximate Processing Time: {days} days")

    # Process and display case review timeframe
    print("\n" + "-" * 70)
    print("📊 CASE REVIEW TIMEFRAME")
    print("-" * 70)
    print(f"Status: {case_review.get('full_text', 'N/A')}")
    as_of_date = case_review.get("as_of_date")
    activity_date = case_review.get("activity_date")
    if as_of_date and activity_date:
        days = calculate_processing_days(as_of_date, activity_date)
        print(f"Last Updated: {as_of_date}")
        print(f"Currently Processing Documents From: {activity_date}")
        print(f"Approximate Processing Time: {days} days")

    # Process and display inquiry response timeframe
    print("\n" + "-" * 70)
    print("❓ PUBLIC INQUIRY FORM RESPONSE TIMEFRAME")
    print("-" * 70)
    print(f"Status: {inquiry_response.get('full_text', 'N/A')}")
    as_of_date = inquiry_response.get("as_of_date")
    activity_date = inquiry_response.get("activity_date")
    if as_of_date and activity_date:
        days = calculate_processing_days(as_of_date, activity_date)
        print(f"Last Updated: {as_of_date}")
        print(f"Currently Responding to Inquiries From: {activity_date}")
        print(f"Approximate Response Time: {days} days")

    # Summary analysis
    print("\n" + "-" * 70)
    print("📈 PROCESSING TIME SUMMARY")
    print("-" * 70)

    timeframes_list = [
        ("Case Creation", case_creation),
        ("Case Review", case_review),
        ("Inquiry Response", inquiry_response),
    ]

    max_days = 0
    slowest_process = ""

    for process_name, process_data in timeframes_list:
        as_of = process_data.get("as_of_date")
        activity = process_data.get("activity_date")
        if as_of and activity:
            days = calculate_processing_days(as_of, activity)
            print(f"  • {process_name}: {days} days")
            if days > max_days:
                max_days = days
                slowest_process = process_name

    if slowest_process:
        print(f"\n⚠️  Slowest Process: {slowest_process} (~{max_days} days)")
        print("✅ Data updated weekly by the U.S. Department of State")

    print("\n" + "=" * 70)


if __name__ == "__main__":
    main()