F-r-ldraledigplannerare/backend/scripts/fetch_swedish_calendar.py

#!/usr/bin/env python3
"""Fetch and cache Swedish calendar data (holidays, de facto days).

This script fetches Swedish holiday data from the Dagsmart API
(https://api.dagsmart.se/) and caches it locally as JSON.

The cache is used by the planning engine to determine working days
without making API calls at runtime, ensuring the planner is offline
and deterministic.

Rate limiting: 5 requests per 10 seconds (enforced via time.sleep).
"""

import datetime
import json
import sys
import time
from pathlib import Path
from typing import Any

import httpx


# Configuration
DAGSMART_API_BASE = "https://api.dagsmart.se/v3/holidays"
CURRENT_YEAR = datetime.date.today().year
YEARS_TO_FETCH = 10  # Fetch 10 years forward from current year (inclusive)
RATE_LIMIT_DELAY = 2.5  # 5 requests per 10 seconds; conservative margin
OUTPUT_DIR = Path(__file__).parent.parent / "app" / "data"
OUTPUT_FILE = OUTPUT_DIR / "swedish_calendar.json"


def validate_api_response(data: dict[str, Any]) -> None:
    """Validate that API response has expected structure."""
    if "year" not in data:
        raise ValueError("API response missing 'year' field")
    if "days" not in data:
        raise ValueError("API response missing 'days' field")
    if not isinstance(data["days"], list):
        raise ValueError("API 'days' field must be a list")

    # Validate each day record
    for day in data["days"]:
        if "date" not in day:
            raise ValueError("API day record missing 'date'")
        if "holidayType" not in day:
            raise ValueError("API day record missing 'holidayType'")
        if "workingDay" not in day:
            raise ValueError("API day record missing 'workingDay'")
        if "observances" not in day:
            raise ValueError("API day record missing 'observances'")
        if not isinstance(day["observances"], list):
            raise ValueError("API day 'observances' must be a list")


def fetch_year(year: int, max_retries: int = 5) -> dict[str, Any]:
    """Fetch calendar data for a single year from Dagsmart API.

    Implements exponential backoff retry for rate limiting (429 responses).

    Args:
        year: Calendar year to fetch
        max_retries: Maximum number of retry attempts

    Returns:
        Parsed JSON response from API

    Raises:
        httpx.HTTPError: On network or HTTP errors
        ValueError: On invalid response format
    """
    url = f"{DAGSMART_API_BASE}?year={year}"
    print(f"Fetching {year}...", file=sys.stderr)

    for attempt in range(max_retries):
        try:
            response = httpx.get(url, timeout=10.0)
            response.raise_for_status()
            data = response.json()
            validate_api_response(data)
            return data
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429 and attempt < max_retries - 1:
                # Rate limited; wait and retry with exponential backoff
                wait_time = 5 * (2**attempt)
                print(
                    f"Rate limited (429). Waiting {wait_time}s before retry "
                    f"({attempt + 1}/{max_retries})...",
                    file=sys.stderr,
                )
                time.sleep(wait_time)
            else:
                raise RuntimeError(
                    f"Failed to fetch calendar for year {year} from {url}: {e}"
                ) from e
        except httpx.HTTPError as e:
            raise RuntimeError(f"Failed to fetch calendar for year {year} from {url}: {e}") from e


def normalize_day(day: dict[str, Any]) -> dict[str, Any]:
    """Normalize a day from API response into our schema.

    Normalizes category to consistent "de_facto" spelling and lowercases names.

    Args:
        day: Day record from Dagsmart API response

    Returns:
        Normalized day record with our schema
    """
    # Extract name from the first observance
    observances = day.get("observances", [])
    name_sv = "Unknown"
    name_en = "Unknown"

    if observances:
        primary_observance = observances[0]
        name_data = primary_observance.get("name", {})
        name_sv = name_data.get("sv", "Unknown")
        name_en = name_data.get("en", "Unknown")

    # Normalize category: vendor uses "defacto", we use "de_facto"
    category = day["holidayType"]
    if category == "defacto":
        category = "de_facto"

    # Normalize names to lowercase for consistency
    name_sv = name_sv.lower() if name_sv != "Unknown" else name_sv
    name_en = name_en.lower() if name_en != "Unknown" else name_en

    return {
        "date": day["date"],
        "name_sv": name_sv,
        "name_en": name_en,
        "category": category,
        "working_day": day["workingDay"],
    }


def compute_bridge_days(year: int, non_working_dates: set[str]) -> list[dict[str, Any]]:
    """Compute bridge days (klämdagar) for a year.

    A bridge day is an ordinary working day (Mon-Fri) that:
    - Is not already a public or de facto day
    - Has a non-working day immediately before or after
    (weekends count as non-working)

    These are days where employers often grant leave because taking one
    vacation day bridges between two non-working days, creating a longer
    continuous break.

    Args:
        year: Calendar year
        non_working_dates: Set of date strings (YYYY-MM-DD) that are non-working

    Returns:
        List of bridge day records
    """
    bridge_days = []

    # Iterate through all days in the year
    start_date = datetime.date(year, 1, 1)
    end_date = datetime.date(year, 12, 31)
    current = start_date

    while current <= end_date:
        # Check if it's a working day (Mon-Fri)
        if current.weekday() < 5:  # 0-4 are Mon-Fri
            date_str = current.isoformat()

            # Not already in non-working set
            if date_str not in non_working_dates:
                # Check if day before and after are both non-working
                prev_day = current - datetime.timedelta(days=1)
                next_day = current + datetime.timedelta(days=1)

                prev_is_nonworking = (
                    prev_day.isoformat() in non_working_dates
                    or prev_day.weekday() >= 5  # Saturday or Sunday
                )
                next_is_nonworking = (
                    next_day.isoformat() in non_working_dates or next_day.weekday() >= 5
                )

                if prev_is_nonworking and next_is_nonworking:
                    bridge_days.append(
                        {
                            "date": date_str,
                            "name_sv": "klämdag",
                            "name_en": "bridge day",
                            "category": "bridge_day",
                            "working_day": True,
                        }
                    )

        current += datetime.timedelta(days=1)

    return bridge_days


def normalize_year(api_response: dict[str, Any]) -> dict[str, Any]:
    """Normalize API response for a year into our schema.

    Includes computing bridge days (klämdagar) from the calendar.

    Args:
        api_response: Full API response for a year

    Returns:
        Normalized year record with our schema
    """
    year = api_response["year"]
    normalized_days = [normalize_day(d) for d in api_response["days"]]

    # Sort by date
    normalized_days.sort(key=lambda d: d["date"])

    # Compute bridge days (klämdagar) from the calendar
    non_working_dates = {d["date"] for d in normalized_days if not d["working_day"]}
    bridge_days = compute_bridge_days(year, non_working_dates)

    # Add bridge days to the list
    normalized_days.extend(bridge_days)

    # Final sort
    normalized_days.sort(key=lambda d: d["date"])

    return {
        "year": year,
        "days": normalized_days,
    }


def fetch_all_years() -> dict[str, Any]:
    """Fetch and normalize calendar data for all required years.

    Returns:
        Complete calendar object with all years
    """
    all_years = []
    end_year = CURRENT_YEAR + YEARS_TO_FETCH

    for year in range(CURRENT_YEAR, end_year):
        try:
            api_response = fetch_year(year)
            normalized = normalize_year(api_response)
            all_years.append(normalized)
        except RuntimeError as e:
            print(f"ERROR: {e}", file=sys.stderr)
            sys.exit(1)

        # Rate limiting: 5 requests per 10 seconds = 2 seconds per request
        time.sleep(RATE_LIMIT_DELAY)

    return {
        "years": all_years,
    }


def write_calendar_atomic(calendar_data: dict[str, Any]) -> None:
    """Write calendar data to file atomically.

    This function writes to a temporary file first, then renames it
    to the target file. This ensures we never leave a partial/corrupt
    cache file behind if the script fails.

    Args:
        calendar_data: Complete calendar object to write
    """
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    # Write to a temporary file first
    temp_file = OUTPUT_FILE.with_suffix(".tmp")

    try:
        with open(temp_file, "w") as f:
            json.dump(calendar_data, f, indent=2, sort_keys=True, ensure_ascii=False)
            f.write("\n")  # Add trailing newline for clean git diffs
    except Exception as e:
        # Clean up temp file if write failed
        if temp_file.exists():
            temp_file.unlink()
        raise RuntimeError(f"Failed to write temporary calendar file: {e}") from e

    # Atomically rename to target
    try:
        temp_file.replace(OUTPUT_FILE)
        print(f"Wrote calendar to {OUTPUT_FILE}", file=sys.stderr)
    except Exception as e:
        if temp_file.exists():
            temp_file.unlink()
        raise RuntimeError(f"Failed to move calendar file to final location: {e}") from e


def main() -> None:
    """Main entry point."""
    print(
        f"Fetching Swedish calendar for years {CURRENT_YEAR}-{CURRENT_YEAR + YEARS_TO_FETCH - 1}",
        file=sys.stderr,
    )

    calendar_data = fetch_all_years()

    print(f"Fetched {len(calendar_data['years'])} years", file=sys.stderr)

    write_calendar_atomic(calendar_data)

    print("Done.", file=sys.stderr)


if __name__ == "__main__":
    main()