Jyotisha/scripts/annual_service/api.py

"""HTTP-facing entry points for the annual package and report facts.

``scripts/jyotish_api_server.py`` registers these thinly (like
``chart_view_points.api``): it loads this module, calls a function with the
JSON body and maps ``AnnualServiceInputError`` to HTTP 400. Request parsing
lives here, not on the handler class.
"""

from __future__ import annotations

from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

try:  # scripts/ on sys.path (API server, tests)
    from annual_service import package as _package
    from annual_service import report_facts as _facts
except ImportError:  # pragma: no cover - repository-root import
    from scripts.annual_service import package as _package
    from scripts.annual_service import report_facts as _facts


class AnnualServiceInputError(ValueError):
    """Bad request input; the API maps it to HTTP 400."""


def _number(body: dict[str, Any], key: str, low: float, high: float, *, default: float | None = None,
            aliases: tuple[str, ...] = ()) -> float:
    value = body.get(key)
    for alias in aliases:
        if value is None:
            value = body.get(alias)
    if value is None or value == "":
        if default is None:
            raise AnnualServiceInputError(f"{key} is required")
        return float(default)
    if isinstance(value, bool):
        raise AnnualServiceInputError(f"{key} must be a number")
    try:
        number = float(value)
    except (TypeError, ValueError) as exc:
        raise AnnualServiceInputError(f"{key} must be a number") from exc
    if not low <= number <= high:
        raise AnnualServiceInputError(f"{key} must be between {low:g} and {high:g}")
    return number


def _ayanamsa(body: dict[str, Any]) -> str:
    try:
        from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, UnsupportedAyanamsaError, normalize_ayanamsa_name
    except ImportError:  # pragma: no cover
        from scripts.ayanamsa_utils import DEFAULT_AYANAMSA_NAME, UnsupportedAyanamsaError, normalize_ayanamsa_name
    raw = body.get("ayanamsa") or body.get("ayanamsa_name") or body.get("ayanamsa_policy")
    if raw in (None, ""):
        return DEFAULT_AYANAMSA_NAME
    try:
        return normalize_ayanamsa_name(raw)
    except UnsupportedAyanamsaError as exc:
        raise AnnualServiceInputError(str(exc)) from exc


def _node_mode(body: dict[str, Any]) -> str:
    raw = body.get("node_mode", body.get("nodeMode"))
    if raw in (None, ""):
        return "mean"
    if not isinstance(raw, str) or raw.strip().lower() not in {"mean", "true"}:
        raise AnnualServiceInputError("node_mode must be mean or true")
    return raw.strip().lower()


def _timezone_id(body: dict[str, Any]) -> str | None:
    for key in ("timezone_id", "timezone"):
        value = body.get(key)
        if isinstance(value, str) and "/" in value:
            try:
                ZoneInfo(value.strip())
            except Exception as exc:  # noqa: BLE001 - an unknown IANA name is a bad request
                raise AnnualServiceInputError(f"{key} is not a known IANA timezone") from exc
            return value.strip()
    return None


def _birth(body: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(body, dict):
        raise AnnualServiceInputError("JSON body must be an object")
    year = int(_number(body, "year", 1800, 2400))
    month = int(_number(body, "month", 1, 12))
    day = int(_number(body, "day", 1, 31))
    hour = int(_number(body, "hour", 0, 23))
    minute = int(_number(body, "minute", 0, 59))
    second = int(_number(body, "second", 0, 59, default=0, aliases=("birth_second",)))
    try:
        local = datetime(year, month, day, hour, minute, second)
    except ValueError as exc:
        raise AnnualServiceInputError("Invalid birth date") from exc
    lat = _number(body, "lat", -90, 90, aliases=("birth_lat",))
    lon = _number(body, "lon", -180, 180, aliases=("birth_lon",))
    if body.get("tz") not in (None, ""):
        tz = _number(body, "tz", -14, 14)
    else:
        try:
            from domain_calculation_service import infer_timezone_offset
        except ImportError:  # pragma: no cover
            from scripts.domain_calculation_service import infer_timezone_offset
        try:
            tz = float(infer_timezone_offset(lat=lat, lon=lon, local_datetime=local))
        except ValueError as exc:
            raise AnnualServiceInputError(str(exc)) from exc
    return {"year": year, "month": month, "day": day, "hour": hour, "minute": minute, "second": second,
            "lat": lat, "lon": lon, "tz": tz, "timezone_id": _timezone_id(body)}


def _request(body: dict[str, Any]) -> dict[str, Any]:
    birth = _birth(body)
    try:
        return _package.normalize_request(
            birth=birth,
            ayanamsa=_ayanamsa(body),
            node_mode=_node_mode(body),
            report_range=body.get("report_range"),
            observation_date=body.get("observation_date") or body.get("today"),
            return_location=body.get("return_location"),
        )
    except _package.AnnualPackageInputError as exc:
        raise AnnualServiceInputError(str(exc)) from exc


def compute_annual_package(body: dict[str, Any]) -> dict[str, Any]:
    """`/api/annual` with mode=package, and `/api/annual/package`."""
    request = _request(body)
    detail = body.get("detail") or "full"
    if detail not in ("full", "slim"):
        raise AnnualServiceInputError("detail must be full or slim")
    package = _package.build_annual_package(request, detail=detail)
    return {
        "success": True,
        "endpoint": "annual",
        "mode": "package",
        "annual_package_version": package["annual_package_version"],
        "package": package,
    }


def compute_annual_report_facts(body: dict[str, Any]) -> dict[str, Any]:
    """`/api/annual/report_facts`: the annual report writer's only input (#153)."""
    request = _request(body)
    confirmed = body.get("birth_time_confirmed")
    if not isinstance(confirmed, bool):
        raise AnnualServiceInputError("birth_time_confirmed must be true or false")
    facts = _facts.build_annual_report_facts(request, birth_time_confirmed=confirmed)
    return {"success": True, "endpoint": "annual_report_facts", "facts": facts}