Jyotisha/scripts/raman_support_observations.py

#!/usr/bin/env python3
"""Raman 2026-08-20 support observations for commercial consultation.

Loads governed_support_only_candidate_ready packets. This is not timing truth
and not a unique-minute or fate-outcome authority.
"""

from __future__ import annotations

import json
from functools import lru_cache
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
PACKET_DIR = ROOT / "references" / "oracle" / "raman_2026_08_20"

BLOCKED_PACKET_PREFIXES = (
    "raman_bvr_health_adversity",
    "raman_bvr_wealth_poverty",
    "raman_bvr_relationship_family",
    "raman_vol2_longevity",
    "raman_bvr_relationship_sibling_batch",  # wide sibling, not sibling_core
)

DOMAIN_PACKETS = {
    "career": ("raman_bvr_nabhasa_chandra_sunapha_batch_closure_packet_2026_08_20.json",),
    "wealth": ("raman_bvr_wealth_core_batch_closure_packet_2026_08_20.json",),
    "finance": ("raman_bvr_wealth_core_batch_closure_packet_2026_08_20.json",),
    "marriage": ("raman_bvr_relationship_spouse_batch_closure_packet_2026_08_20.json",),
    "relationship": ("raman_bvr_relationship_spouse_batch_closure_packet_2026_08_20.json",),
    "family": (
        "raman_bvr_relationship_children_batch_closure_packet_2026_08_20.json",
        "raman_bvr_relationship_mother_batch_closure_packet_2026_08_20.json",
        "raman_bvr_relationship_sibling_core_batch_closure_packet_2026_08_20.json",
    ),
}

DOMAIN_HOUSES = {
    "career": (10,),
    "wealth": (2, 11),
    "finance": (2, 11),
    "marriage": (7,),
    "relationship": (7,),
    "family": (3, 4, 5, 9),
}

NATURAL_KARAKA = {
    1: "Sun",
    2: "Jupiter",
    3: "Mars",
    4: "Moon",
    5: "Jupiter",
    6: "Mars",
    7: "Venus",
    8: "Saturn",
    9: "Jupiter",
    10: "Sun",
    11: "Jupiter",
    12: "Saturn",
}

EXECUTABLE_STEPS = (
    "house_itself_assessment",
    "house_lord_assessment",
    "karaka_assessment",
)
METHODOLOGY_STEPS = (
    "moon_sun_repeat_assessment",
    "navamsa_confirmation",
    "integrated_synthesis",
)

CLAIM_BOUNDARY = "governed_support_only_candidate_ready"


def _token(value: str) -> str:
    return "".join(ch for ch in value.lower() if ch.isalnum())


@lru_cache(maxsize=32)
def _load_packet(filename: str) -> dict[str, Any]:
    path = PACKET_DIR / filename
    if not path.is_file():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def packet_members(filename: str) -> list[dict[str, str]]:
    packet = _load_packet(filename)
    rows = []
    for member in packet.get("member_rules") or []:
        if not isinstance(member, dict):
            continue
        display = str(member.get("display_name") or "").strip()
        if not display:
            continue
        rows.append({
            "rule_id": str(member.get("rule_id") or ""),
            "display_name": display,
            "canonical_mapping": str(member.get("canonical_mapping") or ""),
            "claim_status": str(member.get("claim_status") or "support_only_candidate"),
        })
    return rows


def _yoga_aliases(entry: dict[str, Any]) -> set[str]:
    names = [
        str(entry.get("name") or ""),
        str(entry.get("display_name") or ""),
        str(entry.get("yoga") or ""),
        str(entry.get("rule_id") or ""),
        str(entry.get("canonical_mapping") or ""),
    ]
    tokens = {_token(item) for item in names if item}
    extra = set()
    for token in tokens:
        extra.add(token.replace("yoga", "").replace("yog", ""))
    return {item for item in tokens | extra if item}


def sieve_yogas_for_routes(
    detected: list[dict[str, Any]],
    routes: set[str],
    unevaluated: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    filenames: list[str] = []
    selected_routes = {item for item in routes if item in DOMAIN_PACKETS}
    if not selected_routes:
        selected_routes = {"career", "wealth", "marriage", "family"}
    for route in sorted(selected_routes):
        filenames.extend(DOMAIN_PACKETS[route])
    members: list[dict[str, str]] = []
    seen_ids: set[str] = set()
    for filename in filenames:
        for member in packet_members(filename):
            if member["rule_id"] in seen_ids:
                continue
            seen_ids.add(member["rule_id"])
            members.append(member)

    detected_tokens: list[tuple[dict[str, Any], set[str]]] = [
        (entry, _yoga_aliases(entry)) for entry in detected if isinstance(entry, dict)
    ]
    # Rules the engine could not evaluate (占星师第三轮审查 2026-11-07): a packet
    # member matching one is `not_evaluated`, never a miss ("not formed").
    # Matched by rule id only: a same-named rule from another detector that
    # did evaluate is a real miss.
    unevaluated_by_id: dict[str, dict[str, Any]] = {
        str(entry.get("rule_id")): entry for entry in (unevaluated or [])
        if isinstance(entry, dict) and entry.get("rule_id")
    }
    hits: list[dict[str, Any]] = []
    misses: list[dict[str, Any]] = []
    not_evaluated: list[dict[str, Any]] = []
    used_names: set[str] = set()
    for member in members:
        aliases = _yoga_aliases(member)
        matched = None
        for entry, tokens in detected_tokens:
            if aliases & tokens:
                matched = entry
                break
        row = {
            "name": member["display_name"],
            "rule_id": member["rule_id"],
            "hit": matched is not None,
            "claim_status": member["claim_status"],
            "claim_boundary": CLAIM_BOUNDARY,
        }
        if matched:
            row["planets"] = matched.get("planets")
            row["category"] = matched.get("category")
            # 产品 2026-11-07 同步占星师 Yoga 规则批次: kind and formation checks travel with the hit.
            # 占星师 2026-11-08 第六轮 R3：自参照标记（formation_evidence）随命中一起带出。
            for key in ("claim_kind", "formation_checks", "formation", "formation_evidence"):
                if matched.get(key):
                    row[key] = matched[key]
            hits.append(row)
            used_names.add(str(matched.get("name") or ""))
            continue
        failed = unevaluated_by_id.get(member["rule_id"])
        if failed is not None:
            row["hit"] = None
            row["evaluation_status"] = failed.get("condition_status") or "evaluation_failed"
            row["error_class"] = failed.get("error_class")
            row["claim_boundary"] = "evaluation_failed_not_absent"
            not_evaluated.append(row)
        else:
            misses.append(row)
    leftover = [
        entry for entry, _tokens in detected_tokens
        if str(entry.get("name") or "") not in used_names
    ]
    return {
        "status": "executed",
        "claim_boundary": CLAIM_BOUNDARY,
        "routes": sorted(selected_routes),
        "hits": hits,
        "misses": misses,
        "not_evaluated": not_evaluated,
        "ungoverned_detected": leftover,
        "yogas": hits + misses,
        "count": len(hits) + len(misses),
    }


def _planet(chart: dict[str, Any], name: str) -> dict[str, Any]:
    planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
    row = planets.get(name)
    return row if isinstance(row, dict) else {}


def _house(chart: dict[str, Any], number: int) -> dict[str, Any]:
    houses = chart.get("houses") if isinstance(chart.get("houses"), dict) else {}
    row = houses.get(f"house_{number}")
    return row if isinstance(row, dict) else {}


def _occupants(chart: dict[str, Any], number: int) -> list[str]:
    planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
    found = []
    for name, payload in planets.items():
        if not isinstance(payload, dict):
            continue
        try:
            if int(payload.get("house") or 0) == number:
                found.append(name)
        except (TypeError, ValueError):
            continue
    return found


def _d9_lagna(chart: dict[str, Any]) -> str | None:
    modules = chart.get("modules") if isinstance(chart.get("modules"), dict) else {}
    varga = modules.get("varga_full") if isinstance(modules.get("varga_full"), dict) else {}
    d9 = varga.get("D9") or varga.get("d9")
    if isinstance(d9, dict):
        asc = d9.get("Ascendant") or d9.get("ascendant") or d9.get("lagna")
        if isinstance(asc, dict):
            sign = asc.get("sign") or asc.get("rashi")
            if sign:
                return str(sign)
        sign = d9.get("lagna_sign") or d9.get("asc_sign")
        if sign:
            return str(sign)
    return None


def house_six_step(chart: dict[str, Any], house_number: int) -> dict[str, Any]:
    house = _house(chart, house_number)
    cusp = house.get("cusp_sign") or house.get("sign")
    lord = house.get("lord")
    lord_planet = _planet(chart, str(lord)) if lord else {}
    karaka_name = NATURAL_KARAKA.get(house_number)
    karaka = _planet(chart, karaka_name) if karaka_name else {}
    sun = _planet(chart, "Sun")
    moon = _planet(chart, "Moon")
    occupants = _occupants(chart, house_number)
    executable = {
        EXECUTABLE_STEPS[0]: {
            "status": "executed" if cusp else "blocked",
            "layer": "executable_core",
            "cusp_sign": cusp,
            "occupants": occupants,
        },
        EXECUTABLE_STEPS[1]: {
            "status": "executed" if lord else "blocked",
            "layer": "executable_core",
            "lord": lord,
            "lord_sign": lord_planet.get("sign"),
            "lord_house": lord_planet.get("house"),
            "lord_status": lord_planet.get("status") or lord_planet.get("dignity"),
        },
        EXECUTABLE_STEPS[2]: {
            "status": "executed" if karaka else "blocked",
            "layer": "executable_core",
            "karaka": karaka_name,
            "karaka_sign": karaka.get("sign"),
            "karaka_house": karaka.get("house"),
        },
    }
    methodology = {
        METHODOLOGY_STEPS[0]: {
            "status": "methodology_only",
            "layer": "methodology",
            "sun_house": sun.get("house"),
            "moon_house": moon.get("house"),
            "repeats_house": house_number in {
                int(sun.get("house") or 0),
                int(moon.get("house") or 0),
            },
        },
        METHODOLOGY_STEPS[1]: {
            "status": "methodology_only",
            "layer": "methodology",
            "d9_lagna": _d9_lagna(chart),
        },
        METHODOLOGY_STEPS[2]: {
            "status": "methodology_only",
            "layer": "methodology",
            "note": "positive/negative majority is a reading guide, not an 8/20 executable contract",
        },
    }
    return {
        "house": house_number,
        "claim_boundary": CLAIM_BOUNDARY,
        "executable_core": executable,
        "methodology": methodology,
        "steps": {**executable, **methodology},
    }


def build_raman_support_observations(
    chart: dict[str, Any],
    detected_yogas: list[dict[str, Any]],
    routes: set[str],
    unevaluated: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    selected = {item for item in routes if item in DOMAIN_HOUSES}
    if not selected:
        selected = {"career", "wealth", "marriage", "family"}
    houses: list[dict[str, Any]] = []
    seen: set[int] = set()
    for route in sorted(selected):
        for number in DOMAIN_HOUSES[route]:
            if number in seen:
                continue
            seen.add(number)
            houses.append(house_six_step(chart, number))
    yogas = sieve_yogas_for_routes(detected_yogas, routes, unevaluated=unevaluated)
    return {
        "status": "executed",
        "claim_boundary": CLAIM_BOUNDARY,
        "promotion_gate_status": CLAIM_BOUNDARY,
        "houses": houses,
        "yogas": yogas,
        "blocked_packets": list(BLOCKED_PACKET_PREFIXES),
    }