Jyotisha/scripts/consult_full_data_cache.py

"""Cache the full-data report packet for the consult card.

The key is chart identity + ayanamsa + node mode + engine version.
The file name is a hash. Birth data stays inside the scratch file.
"""

from __future__ import annotations

import hashlib
import json
import time
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Callable

ROOT = Path(__file__).resolve().parents[1]
CACHE_SCHEMA = "consult-full-data-cache-v1"
SLOW_MISS_SECONDS = 20.0

PacketBuilder = Callable[[dict[str, Any]], dict[str, Any]]


def cache_dir() -> Path:
    path = ROOT / "scratch" / "local" / "consult_full_data_cache"
    path.mkdir(parents=True, exist_ok=True)
    return path


def engine_version() -> str:
    try:
        from jyotish_vedic import __version__ as version
    except Exception:
        return "unknown"
    return str(version)


def _number(value: Any) -> int | float:
    number = float(value)
    if number == int(number):
        return int(number)
    return number


def identity_from_body(body: dict[str, Any]) -> dict[str, Any]:
    """Birth fields are required. Missing keys raise KeyError so the caller can gap."""
    node = body.get("node_mode", body.get("nodeMode", "mean"))
    ayanamsa = body.get("ayanamsa", "lahiri")
    second = body.get("second", 0)
    return {
        "schema": CACHE_SCHEMA,
        "year": int(body["year"]),
        "month": int(body["month"]),
        "day": int(body["day"]),
        "hour": _number(body["hour"]),
        "minute": _number(body["minute"]),
        "second": _number(second if second is not None else 0),
        "lat": _number(body["lat"]),
        "lon": _number(body["lon"]),
        "tz": _number(body["tz"]),
        "ayanamsa": str(ayanamsa),
        "node_mode": str(node or "mean"),
        "engine_version": engine_version(),
    }


def cache_key(identity: dict[str, Any]) -> str:
    canonical = json.dumps(identity, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _cache_path(identity: dict[str, Any]) -> Path:
    return cache_dir() / f"{cache_key(identity)}.json"


def get_or_build(
    identity: dict[str, Any],
    builder: PacketBuilder | None = None,
    *,
    clock: Callable[[], float] | None = None,
) -> dict[str, Any]:
    """Return the packet. A miss calls builder once and stores the packet."""
    now = clock or time.perf_counter
    started = now()
    path = _cache_path(identity)
    if path.is_file():
        packet = json.loads(path.read_text(encoding="utf-8"))
        elapsed = now() - started
        return {"packet": packet, "hit": True, "elapsed_s": elapsed, "slow": False}
    if builder is None:
        builder = build_full_data_packet
    packet = builder(identity)
    temporary = path.with_suffix(".json.tmp")
    temporary.write_text(json.dumps(packet, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
    temporary.replace(path)
    elapsed = now() - started
    return {
        "packet": packet,
        "hit": False,
        "elapsed_s": elapsed,
        "slow": elapsed > SLOW_MISS_SECONDS,
    }


def build_full_data_packet(identity: dict[str, Any]) -> dict[str, Any]:
    """Packet-only build. No markdown render and no sanitized rewrite."""
    from datetime import date

    from calculation_profile_contract import attach_calculation_profile
    from jyotish_engine import build_pl9_style_export_packet, cmd_full_reading

    today = date.today()
    args = SimpleNamespace(
        year=int(identity["year"]),
        month=int(identity["month"]),
        day=int(identity["day"]),
        hour=float(identity["hour"]),
        minute=float(identity["minute"]),
        second=float(identity["second"]),
        lat=float(identity["lat"]),
        lon=float(identity["lon"]),
        tz=float(identity["tz"]),
        ayanamsa=identity["ayanamsa"],
        node_mode=identity["node_mode"],
        today=today.isoformat(),
        target_year=today.year,
        age=today.year - int(identity["year"]),
        transit_date=today.isoformat(),
        birth_time_accuracy="confirmed",
    )
    reading = cmd_full_reading(args)
    if not isinstance(reading, dict) or reading.get("error"):
        raise RuntimeError("full_reading_unavailable")
    packet = build_pl9_style_export_packet(reading)
    return attach_calculation_profile(packet, args)


def _requested_domains(body: dict[str, Any]) -> list[str]:
    # Card domains (parents, children) are not the engine theme (family).
    raw = body.get("consult_card_domains")
    if raw is None:
        raw = body.get("themes", body.get("theme"))
    if raw is None:
        return []
    if isinstance(raw, str):
        return [raw]
    if isinstance(raw, list):
        return [str(item) for item in raw if isinstance(item, str)]
    return []


def attach_consult_full_data(result: dict[str, Any], body: dict[str, Any] | None) -> dict[str, Any]:
    """Thin hook for the consultation response. Failures do not break the chat."""
    payload = dict(body or {})
    if payload.get("include_consult_card_facts") is not True:
        return result
    try:
        from consult_card_domain_facts import build_consult_card_facts
    except ImportError:
        from scripts.consult_card_domain_facts import build_consult_card_facts
    try:
        identity = identity_from_body(payload)
        cached = get_or_build(identity)
        facts = build_consult_card_facts(
            cached["packet"],
            _requested_domains(payload),
            minute_confirmed=None,
            body=payload,
        )
        facts["timing"] = {
            "hit": bool(cached["hit"]),
            "elapsed_s": cached["elapsed_s"],
            "slow": bool(cached["slow"]),
        }
    except Exception as exc:
        facts = {
            "schema": "consult-card-facts-v1",
            "gaps": ["consult_full_data_unavailable"],
            "error_type": type(exc).__name__,
        }
    attached = dict(result)
    attached["consult_card_facts"] = facts
    return attached