"""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}