Jyotisha/scripts/report_render/reading_facts.py

"""Facts for the model-written reading at the front of the report (issue #113 R1).

The reading is written by a model, but only from what the engine already
computed. This module copies the consult-card leaves (the same facts the chat
uses) from the report's own full-data packet, drops paths and gaps, and adds
the current dasha dates. It computes nothing new.
"""

from __future__ import annotations

import copy
from typing import Any

try:
    from scripts.consult_card_domain_facts import assert_fact_leaves, build_consult_card_facts, dig
    from scripts.consult_full_data_cache import _attach_graha_drishti
except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution path
    from consult_card_domain_facts import assert_fact_leaves, build_consult_card_facts, dig
    from consult_full_data_cache import _attach_graha_drishti

READING_FACTS_SCHEMA = "report-reading-facts-v1"
READING_DOMAINS = ("general", "career", "wealth", "marriage", "health")
DASHA = "worksheets.timing_and_predictive_systems.dasha.current_dasha"
UPCOMING_SUB_PERIODS = 2


def _compact(node: Any) -> Any:
    """Leaf values only. Gaps, withheld and blocked leaves are left out."""
    if isinstance(node, dict):
        if "status" in node and "check" in node:
            return node.get("value") if node["status"] == "value" else None
        out = {}
        for key, value in node.items():
            kept = _compact(value)
            if kept is not None and kept != {} and kept != []:
                out[key] = kept
        return out
    if isinstance(node, list):
        return [item for item in (_compact(entry) for entry in node) if item is not None]
    return node


def _period(row: Any) -> dict[str, Any] | None:
    if not isinstance(row, dict) or not isinstance(row.get("lord"), str):
        return None
    out = {"lord": row["lord"]}
    for key in ("start", "end"):
        if isinstance(row.get(key), str):
            out[key] = row[key]
    return out


def _dasha_dates(packet: dict) -> dict[str, Any]:
    current = dig(packet, DASHA)
    if not isinstance(current, dict):
        return {}
    dates: dict[str, Any] = {}
    md = _period(current)
    if md:
        dates["mahadasha"] = md
    rows = current.get("antardasha_timeline")
    if isinstance(rows, list):
        index = next((i for i, row in enumerate(rows) if isinstance(row, dict) and row.get("is_current") is True), None)
        if index is not None:
            now = _period(rows[index])
            if now:
                dates["antardasha"] = now
            upcoming = [_period(row) for row in rows[index + 1:index + 1 + UPCOMING_SUB_PERIODS]]
            dates["next_antardashas"] = [row for row in upcoming if row]
    return dates


def _domain_factors(row: dict) -> list[dict[str, Any]]:
    factors = []
    for factor in row.get("factors") or []:
        if isinstance(factor, dict) and factor.get("kind") == "planet" and isinstance(factor.get("planet"), str):
            factors.append({"role": factor.get("role"), "planet": factor["planet"]})
    return factors


def build_report_reading_facts(
    packet: dict,
    full_reading: dict,
    *,
    minute_confirmed: bool,
    today: str | None = None,
) -> dict[str, Any]:
    """Compact, packet-sourced facts for the reading. Raises on any non-packet leaf."""
    working = _attach_graha_drishti(copy.deepcopy(packet), full_reading)
    raw = build_consult_card_facts(working, list(READING_DOMAINS), minute_confirmed=minute_confirmed)
    rows = {row["domain"]: row for row in raw["domains"]}
    for row in rows.values():
        assert_fact_leaves(working, row)
    general = rows["general"]
    planets = {}
    for name, record in general["planets"].items():
        kept = _compact({key: value for key, value in record.items() if key != "domain_varga_house"})
        if kept.get("present"):
            kept.pop("present")
            planets[name] = kept
    domains = {}
    for name, row in rows.items():
        domains[name] = {
            "varga": row["domain_varga"],
            "houses": _compact(row["houses"]),
            "planets_in_varga": {
                planet: house
                for planet, record in row["planets"].items()
                if (house := _compact(record.get("domain_varga_house"))) is not None
            },
            "significators": _domain_factors(row),
        }
    ascendant = dig(packet, "core_chart.ascendant")
    return {
        "schema": READING_FACTS_SCHEMA,
        "minute_confirmed": minute_confirmed,
        **({"today": today} if today else {}),
        "ascendant": _compact({
            key: {"status": "value", "check": "path", "value": ascendant.get(key)}
            for key in ("sign", "lord", "nakshatra")
            if isinstance(ascendant, dict) and ascendant.get(key) is not None
        }),
        "planets": planets,
        "domains": domains,
        "timing": {**_compact(general["timing"]), **_dasha_dates(packet)},
    }