Jyotisha/scripts/public_real_case_benchmark.py

#!/usr/bin/env python3
"""Replay research-grade public events through the local Jyotish evidence stack."""

from __future__ import annotations

import argparse
import copy
import json
import subprocess
import sys
from datetime import date
from pathlib import Path
from typing import Any

from scripts.functional_benefics import derive_functional_benefic_malefic
from scripts.narayana_dasha import narayana_dasha_full_report


ROOT = Path(__file__).resolve().parents[1]
ENGINE = ROOT / "scripts" / "jyotish_engine.py"
SIGNS = [
    "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
]
EVENT_HOUSES = {"career": [10, 6, 9, 11], "marriage": [7, 2, 11, 5]}
EVENT_KARAKAS = {"career": {"Sun", "Saturn", "Mercury"}, "marriage": {"Venus", "Jupiter"}}
PRIMARY_HOUSE = {"career": 10, "marriage": 7}
EXPECTED_LABEL = {"career": "career_status", "marriage": "legal_marriage"}
SIGN_LORDS = {
    "Aries": "Mars", "Taurus": "Venus", "Gemini": "Mercury", "Cancer": "Moon",
    "Leo": "Sun", "Virgo": "Mercury", "Libra": "Venus", "Scorpio": "Mars",
    "Sagittarius": "Jupiter", "Capricorn": "Saturn", "Aquarius": "Saturn", "Pisces": "Jupiter",
}
_ENGINE_JSON_CACHE: dict[str, dict[str, Any]] = {}


def clear_engine_cache() -> None:
    _ENGINE_JSON_CACHE.clear()


def summarize_results(rows: list[dict[str, Any]]) -> dict[str, Any]:
    total = len(rows)
    blocked = sum(bool(row.get("blocked")) for row in rows)
    evaluated = total - blocked
    hits = sum(row.get("result_class") in {"strong_hit", "weak_hit"} for row in rows if not row.get("blocked"))
    exact = sum(bool(row.get("matched_expected_label")) for row in rows if not row.get("blocked"))
    activation_rate = hits / evaluated if evaluated else None
    strong_rate = exact / evaluated if evaluated else None
    return {
        "total_events": total,
        "evaluated_events": evaluated,
        "strong_hits": sum(row.get("result_class") == "strong_hit" for row in rows),
        "weak_hits": sum(row.get("result_class") == "weak_hit" for row in rows),
        "misses": sum(row.get("result_class") == "miss" for row in rows),
        "blocked_events": blocked,
        "known_event_activation_rate": activation_rate,
        "strong_activation_rate": strong_rate,
        "positive_event_recall": activation_rate,
        "positive_event_recall_deprecated": True,
        "exact_label_rate": strong_rate,
        "exact_label_rate_deprecated": True,
        "blocked_rate": blocked / total if total else None,
        "balanced_accuracy": None,
        "balanced_accuracy_blocked_reason": "no_verified_negative_control_dates",
    }


def promotion_decision(v1: dict[str, Any], v2: dict[str, Any]) -> dict[str, Any]:
    if int(v2.get("blocked_events") or 0) > int(v1.get("blocked_events") or 0):
        return {"promote": False, "reason": "v2_increased_blocked_events"}
    recall1 = v1.get("positive_event_recall")
    recall2 = v2.get("positive_event_recall")
    exact1 = v1.get("exact_label_rate")
    exact2 = v2.get("exact_label_rate")
    if None in {recall1, recall2, exact1, exact2}:
        return {"promote": False, "reason": "comparison_metric_missing"}
    improved = recall2 >= recall1 and exact2 >= exact1 and (recall2 > recall1 or exact2 > exact1)
    return {"promote": improved, "reason": "holdout_metrics_improved" if improved else "no_holdout_improvement"}


def compare_reports(v1: dict[str, Any], v2: dict[str, Any]) -> dict[str, Any]:
    """Compare frozen rule versions without reinterpreting holdout outcomes."""
    v1_cases = {row["case_id"]: row for row in v1.get("cases") or []}
    v2_cases = {row["case_id"]: row for row in v2.get("cases") or []}
    deltas = []
    for case_id in sorted(v1_cases.keys() & v2_cases.keys()):
        before = v1_cases[case_id]
        after = v2_cases[case_id]
        before_signals = set(before.get("signals") or [])
        deltas.append({
            "case_id": case_id,
            "v1_score": before.get("score"),
            "v2_score": after.get("score"),
            "score_delta": (after.get("score") or 0) - (before.get("score") or 0),
            "v1_result_class": before.get("result_class"),
            "v2_result_class": after.get("result_class"),
            "added_signals": sorted(set(after.get("signals") or []) - before_signals),
        })
    return {
        "benchmark_id": "public_real_case_holdout_comparison_2026_07_11",
        "boundary": "Blind positive-event holdout comparison; no negative controls and no scientific accuracy claim.",
        "v1_summary": v1.get("summary") or {},
        "v2_summary": v2.get("summary") or {},
        "promotion": promotion_decision(v1.get("summary") or {}, v2.get("summary") or {}),
        "case_deltas": deltas,
    }


def combine_reports(reports: list[dict[str, Any]], promotion: dict[str, Any]) -> dict[str, Any]:
    rows = [row for report in reports for row in report.get("cases") or []]
    return {
        "benchmark_id": "public_real_case_20_case_closure_2026_07_11",
        "rule_version": "v2",
        "method": {
            "cohorts": ["batch1_discovery_10", "frozen_holdout_10"],
            "selection": "Rodden A/AA public figures with independently dated public career or legal-marriage events",
            "score_thresholds": {"strong_hit": ">=7", "weak_hit": "4-6", "miss": "<4"},
        },
        "summary": summarize_results(rows),
        "domain_summaries": {
            domain: summarize_results([row for row in rows if row.get("domain") == domain])
            for domain in ("career", "marriage")
        },
        "holdout_promotion": promotion,
        "boundary": "Twenty positive public events; no negative controls, specificity estimate, or scientific accuracy claim.",
        "technique_audit": [
            {"technique": "D1 + Functional Benefic/Malefic", "status": "used", "scope": "20/20"},
            {"technique": "D9/UL/Darakaraka", "status": "used", "scope": "10 marriage events"},
            {"technique": "D10/A10/Amatyakaraka", "status": "used", "scope": "10 career events"},
            {"technique": "Vimshottari MD/AD", "status": "used", "scope": "20/20"},
            {"technique": "Narayana Dasha", "status": "used", "scope": "20/20"},
            {"technique": "Double Transit PAC", "status": "used", "scope": "20/20"},
            {"technique": "Rahu/Ketu dispositor", "status": "used", "scope": "v2 scoring"},
            {"technique": "Vimshottari PD/PrAD", "status": "partial", "reason": "ratio expansion available but not externally validated or scored"},
            {"technique": "Tajika/Varshaphala/Muntha", "status": "partial", "reason": "local annual layer remains simplified and external oracle closure is incomplete"},
            {"technique": "KP exact cusp/significators", "status": "partial", "reason": "current local KP house layer uses sign-center approximation rather than exact cusps"},
            {"technique": "VedAstro official raw", "status": "blocked", "reason": "official_snapshot_budget_exhausted"},
            {"technique": "PyJHora/JHora/jyotishganit parity", "status": "blocked", "reason": "external canonical raw comparison incomplete"},
            {"technique": "MEVG / Global Web Evidence", "status": "used", "scope": "20 public birth/event source pairs"},
            {"technique": "Real Case Calibration", "status": "used", "scope": "10 discovery + 10 frozen holdout"},
            {"technique": "Negative controls", "status": "blocked", "reason": "no verified non-event dates"},
        ],
        "technique_debt": {
            "vimshottari_pd_prad": "available_ratio_expansion_not_scored_or_externally_validated",
            "tajika_varshaphala_muntha": "available_experimental_not_scored_due_simplified_year_lord_and_oracle_gap",
            "kp_cusp_significators": "partial_not_scored_house_centers_are_not_precise_cusps",
            "annual_transit_to_arudha_or_ul": "untested_candidate_layer",
            "negative_control_dates": "missing_blocks_balanced_accuracy",
        },
        "cases": rows,
    }


def node_dispositor_bonus(
    active_lords: set[str],
    domain: str,
    chart: dict[str, Any],
    roles: dict[str, Any],
) -> tuple[int, list[str]]:
    event_houses = set(EVENT_HOUSES[domain])
    score = 0
    signals: list[str] = []
    planets = chart.get("planets") or {}
    for node in sorted(active_lords & {"Rahu", "Ketu"}):
        node_sign = (planets.get(node) or {}).get("sign")
        dispositor = SIGN_LORDS.get(node_sign)
        if not dispositor:
            continue
        if set((roles.get("owned_houses") or {}).get(dispositor) or []) & event_houses:
            score += 1
            signals.append(f"{node}_dispositor_{dispositor}_owns_event_house")
        occupied = (planets.get(dispositor) or {}).get("house")
        if occupied in event_houses:
            score += 1
            signals.append(f"{node}_dispositor_{dispositor}_occupies_event_house:{occupied}")
    return score, signals


def _house_from_sign(ascendant: str, target: str) -> int | None:
    if ascendant not in SIGNS or target not in SIGNS:
        return None
    return (SIGNS.index(target) - SIGNS.index(ascendant)) % 12 + 1


def varga_and_karaka_bonus(
    active_lords: set[str],
    domain: str,
    varga: dict[str, Any],
    jaimini: dict[str, Any],
) -> tuple[int, list[str]]:
    chart_key = "D10_Dasamsa" if domain == "career" else "D9_Navamsa"
    chart = ((varga.get("divisional_charts") or {}).get(chart_key) or {})
    ascendant = chart.get("ascendant")
    primary_house = PRIMARY_HOUSE[domain]
    primary_sign = SIGNS[(SIGNS.index(ascendant) + primary_house - 1) % 12] if ascendant in SIGNS else None
    primary_lord = SIGN_LORDS.get(primary_sign)
    lagna_lord = SIGN_LORDS.get(ascendant)
    score = 0
    signals: list[str] = []
    label = "D10" if domain == "career" else "D9"
    for lord in sorted(active_lords):
        if lord == lagna_lord:
            score += 1
            signals.append(f"active_dasha_matches_{label}_Lagna_lord:{lord}")
        if lord == primary_lord:
            score += 1
            signals.append(f"active_dasha_matches_{label}_{primary_house}L:{lord}")
        lord_sign = (chart.get(lord) or {}).get("sign")
        if _house_from_sign(ascendant, lord_sign) == primary_house:
            score += 1
            signals.append(f"active_dasha_occupies_{label}_house_{primary_house}:{lord}")
    karaka_name = "Amatyakaraka" if domain == "career" else "Darakaraka"
    karaka_planet = ((((jaimini.get("chara_karaka_7") or {}).get("karaka_table") or {}).get(karaka_name) or {}).get("planet"))
    if karaka_planet in active_lords:
        score += 1
        signals.append(f"active_dasha_matches_{karaka_name}:{karaka_planet}")
    return score, signals


def _engine_json(command: str, subject: dict[str, Any], *extra: str, timeout: int = 30) -> dict[str, Any]:
    cache_key = json.dumps(
        {"command": command, "subject": subject, "extra": extra},
        sort_keys=True,
        ensure_ascii=True,
        default=str,
    )
    if cache_key in _ENGINE_JSON_CACHE:
        return copy.deepcopy(_ENGINE_JSON_CACHE[cache_key])
    args = [
        sys.executable, str(ENGINE), command,
        "--year", str(subject["year"]), "--month", str(subject["month"]),
        "--day", str(subject["day"]), "--hour", str(subject["hour"]),
        "--minute", str(subject["minute"]), "--lat", str(subject["lat"]),
        "--lon", str(subject["lon"]), "--tz", str(subject["tz"]),
        "--node-mode", str(subject.get("node_mode", "mean")),
        *extra,
    ]
    completed = subprocess.run(args, cwd=ROOT, check=True, capture_output=True, text=True, timeout=timeout)
    payload = json.loads(completed.stdout)
    _ENGINE_JSON_CACHE[cache_key] = payload
    return copy.deepcopy(payload)


def _find_dasha(dasha: dict[str, Any], event_date: str) -> tuple[str | None, str | None]:
    target = date.fromisoformat(event_date)
    for md in dasha.get("timeline") or []:
        if date.fromisoformat(md["start"][:10]) <= target < date.fromisoformat(md["end"][:10]):
            for ad in md.get("antardasha_timeline") or []:
                if date.fromisoformat(ad["start"][:10]) <= target < date.fromisoformat(ad["end"][:10]):
                    return md.get("lord"), ad.get("lord")
            return md.get("lord"), None
    return None, None


def _planet_score(planet: str | None, event_houses: set[int], chart: dict[str, Any], roles: dict[str, Any], karakas: set[str]) -> tuple[int, list[str]]:
    if not planet:
        return 0, []
    score = 0
    signals: list[str] = []
    owned = set((roles.get("owned_houses") or {}).get(planet) or [])
    occupied = (chart.get("planets") or {}).get(planet, {}).get("house")
    owned_hits = sorted(owned & event_houses)
    if owned_hits:
        score += 2
        signals.append(f"{planet}_owns_event_houses:{owned_hits}")
    if occupied in event_houses:
        score += 1
        signals.append(f"{planet}_occupies_event_house:{occupied}")
    if planet in karakas:
        score += 1
        signals.append(f"{planet}_domain_karaka")
    return score, signals


def score_active_dasha_lords(
    lords: list[str | None],
    event_houses: set[int],
    chart: dict[str, Any],
    roles: dict[str, Any],
    karakas: set[str],
) -> tuple[int, list[str]]:
    score = 0
    signals: list[str] = []
    for lord in dict.fromkeys(lord for lord in lords if lord):
        points, lord_signals = _planet_score(lord, event_houses, chart, roles, karakas)
        score += points
        signals.extend(lord_signals)
    return score, signals


def _transit_json(event_date: str, subject: dict[str, Any]) -> dict[str, Any]:
    target = date.fromisoformat(event_date)
    command = [
        sys.executable, str(ENGINE), "transit",
        "--year", str(target.year), "--month", str(target.month), "--day", str(target.day),
        "--planet", "Jupiter,Saturn", "--tz", str(subject["tz"]),
        "--node-mode", str(subject.get("node_mode", "mean")),
    ]
    completed = subprocess.run(command, cwd=ROOT, check=True, capture_output=True, text=True, timeout=30)
    return json.loads(completed.stdout)


def ashtakavarga_audit(domain: str, packet: dict[str, Any], transit: dict[str, Any]) -> dict[str, Any]:
    event_houses = EVENT_HOUSES[domain]
    sav = packet.get("sav") or {}
    bav = packet.get("bav") or {}
    event_house_sav = {
        str(house): (packet.get("house_scores") or {}).get(f"house_{house}")
        for house in event_houses
    }
    transit_support = {}
    for planet in ("Jupiter", "Saturn"):
        sign = ((transit.get("planets") or {}).get(planet) or {}).get("sign")
        sign_index = SIGNS.index(sign) if sign in SIGNS else None
        bindus = ((bav.get(planet) or {}).get("bindus") or [])
        transit_support[planet] = {
            "sign": sign,
            "sav": (sav.get("scores") or {}).get(sign),
            "bav": bindus[sign_index] if sign_index is not None and sign_index < len(bindus) else None,
        }
    return {
        "status": "used_non_scoring",
        "scoring_effect": 0,
        "method": packet.get("method"),
        "version": packet.get("version"),
        "sav_total": sav.get("total"),
        "sav_valid": sav.get("valid"),
        "all_bav_valid": packet.get("all_bav_valid"),
        "event_house_sav": event_house_sav,
        "transit_support": transit_support,
        "settings": {
            "ayanamsa": transit.get("ayanamsa"),
            "node_mode": transit.get("node_mode"),
        },
        "boundary": "Audit evidence only. SAV/BAV does not change V2.1 event scores until a fresh holdout validates it.",
    }


def _narayana_at_event(subject: dict[str, Any], chart: dict[str, Any], event_date: str) -> dict[str, Any]:
    asc_sign = chart["ascendant"]["sign"]
    asc_idx = SIGNS.index(asc_sign)
    planet_lons = {name: data["degree"] for name, data in chart["planets"].items() if "degree" in data}
    born = date(subject["year"], subject["month"], subject["day"])
    target = date.fromisoformat(event_date)
    age = (target - born).days / 365.2425
    report = narayana_dasha_full_report(asc_idx, planet_lons, current_age=age, birth_year=subject["year"])
    return report.get("current_dasha") or {}


def _arudha_lord(jaimini: dict[str, Any], domain: str) -> str | None:
    arudha = jaimini.get("arudha_padas") or {}
    if domain == "career":
        return ((arudha.get("padas") or {}).get("A10") or {}).get("lord")
    return (arudha.get("upapada") or {}).get("lord")


def _double_transit_score(packet: dict[str, Any]) -> tuple[int, list[str]]:
    strengths = [row.get("strength") for row in packet.get("double_transit") or []]
    if "strong" in strengths:
        return 2, ["double_transit_pac_strong"]
    if strengths:
        return 1, ["double_transit_pac_present"]
    return 0, []


def replay_case(case: dict[str, Any], rule_version: str = "v1") -> dict[str, Any]:
    subject = case["subject"]
    event = case["event_outcomes"][0]
    domain = event["domain"]
    event_houses = set(EVENT_HOUSES[domain])
    try:
        chart = _engine_json("chart", subject)
        dasha = _engine_json("dasha", subject, "--years", "100")
        varga = _engine_json("varga", subject, "--d10" if domain == "career" else "--d9")
        jaimini = _engine_json("jaimini", subject)
        pac = _engine_json(
            "double-transit-pac", subject,
            "--date", event["event_date"], "--house", str(PRIMARY_HOUSE[domain]),
        )
    except (subprocess.SubprocessError, json.JSONDecodeError, KeyError, ValueError) as exc:
        return {
            "case_id": case["case_id"], "name": subject["name"], "domain": domain,
            "event_date": event["event_date"], "blocked": True, "result_class": "blocked",
            "matched_expected_label": False, "blocked_reason": f"{type(exc).__name__}: {exc}",
        }

    roles = derive_functional_benefic_malefic(chart["ascendant"]["sign"])
    md, ad = _find_dasha(dasha, event["event_date"])
    score = 0
    signals: list[str] = []
    if rule_version == "v2_1":
        score, signals = score_active_dasha_lords([md, ad], event_houses, chart, roles, EVENT_KARAKAS[domain])
    else:
        for lord in (md, ad):
            points, lord_signals = _planet_score(lord, event_houses, chart, roles, EVENT_KARAKAS[domain])
            score += points
            signals.extend(lord_signals)

    active_lords = {lord for lord in (md, ad) if lord}
    if rule_version in {"v2", "v2_1"}:
        node_points, node_signals = node_dispositor_bonus(active_lords, domain, chart, roles)
        varga_points, varga_signals = varga_and_karaka_bonus(active_lords, domain, varga, jaimini)
        score += node_points + varga_points
        signals.extend(node_signals)
        signals.extend(varga_signals)

    arudha_lord = _arudha_lord(jaimini, domain)
    if arudha_lord in {md, ad}:
        score += 1
        signals.append(f"active_dasha_matches_{'A10' if domain == 'career' else 'UL'}_lord:{arudha_lord}")

    narayana = _narayana_at_event(subject, chart, event["event_date"])
    narayana_md = narayana.get("md") or {}
    event_sign = SIGNS[(SIGNS.index(chart["ascendant"]["sign"]) + PRIMARY_HOUSE[domain] - 1) % 12]
    if narayana_md.get("sign") == event_sign:
        score += 2
        signals.append(f"narayana_activates_primary_event_sign:{event_sign}")
    narayana_lord = narayana_md.get("lord")
    if set((roles.get("owned_houses") or {}).get(narayana_lord) or []) & event_houses:
        score += 1
        signals.append(f"narayana_lord_owns_event_house:{narayana_lord}")

    pac_points, pac_signals = _double_transit_score(pac)
    score += pac_points
    signals.extend(pac_signals)

    ashtakavarga = {"status": "not_run", "scoring_effect": 0}
    if rule_version == "v2_1":
        try:
            ashtakavarga_packet = _engine_json("ashtakavarga", subject)
            transit_packet = _transit_json(event["event_date"], subject)
            ashtakavarga = ashtakavarga_audit(domain, ashtakavarga_packet, transit_packet)
        except (subprocess.SubprocessError, json.JSONDecodeError, KeyError, ValueError) as exc:
            ashtakavarga = {
                "status": "blocked",
                "scoring_effect": 0,
                "blocked_reason": f"{type(exc).__name__}: {exc}",
            }

    if score >= 7:
        result_class = "strong_hit"
        actual_label = EXPECTED_LABEL[domain]
    elif score >= 4:
        result_class = "weak_hit"
        actual_label = "domain_activation"
    else:
        result_class = "miss"
        actual_label = None

    return {
        "case_id": case["case_id"],
        "name": subject["name"],
        "domain": domain,
        "event_date": event["event_date"],
        "outcome": event["outcome"],
        "birth_time_rating": subject["birth_source"]["time_accuracy_rating"],
        "rule_version": rule_version,
        "blocked": False,
        "result_class": result_class,
        "score": score,
        "expected_label": EXPECTED_LABEL[domain],
        "actual_label": actual_label,
        "matched_expected_label": actual_label == EXPECTED_LABEL[domain],
        "signals": signals,
        "evidence": {
            "ascendant": chart["ascendant"],
            "vimshottari": {"mahadasha": md, "antardasha": ad},
            "narayana": narayana,
            "functional_benefic_malefic": roles,
            "domain_varga": varga,
            "arudha_lord": arudha_lord,
            "double_transit_pac": pac,
            "ashtakavarga_audit": ashtakavarga,
            "birth_source": subject["birth_source"],
            "event_source": event["source"],
        },
    }


def build_report(
    manifest: dict[str, Any],
    strict_probe_blocked_reason: str | None = None,
    rule_version: str = "v1",
) -> dict[str, Any]:
    rows = [replay_case(case, rule_version=rule_version) for case in manifest.get("cases") or []]
    return {
        "benchmark_id": "public_real_case_benchmark_2026_07_11",
        "rule_version": rule_version,
        "method": {
            "selection": "Rodden A/AA public figures with independently dated public events",
            "pre_registered_layers": ["D1", "D9_or_D10", "UL_or_A10", "Functional Benefic/Malefic", "Vimshottari MD/AD", "Narayana Dasha", "Double Transit PAC"] + (["Rahu/Ketu dispositor", "D9/D10 Lagna and primary-house lord", "Amatyakaraka/Darakaraka"] if rule_version in {"v2", "v2_1"} else []) + (["SAV/BAV non-scoring audit", "deduplicated MD/AD lord scoring"] if rule_version == "v2_1" else []),
            "score_thresholds": {"strong_hit": ">=7", "weak_hit": "4-6", "miss": "<4"},
            "boundary": "Positive-event technical activation replay; not scientific predictive accuracy.",
        },
        "summary": summarize_results(rows),
        "strict_workflow_batch": {
            "status": "blocked" if strict_probe_blocked_reason else "not_run",
            "blocked_reason": strict_probe_blocked_reason,
        },
        "external_oracle_boundary": {
            "VedAstro": "diagnostic_only_unless_official_raw_present",
            "PyJHora": "blocked_or_benchmark_only_until_dependency_available",
            "JHora": "manual_oracle_not_automated",
            "jyotishganit": "parity_contract_separate_from_this_event_replay",
        },
        "cases": rows,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--manifest", default="references/real_case_calibration/replay_manifest.json")
    parser.add_argument("--output")
    parser.add_argument("--strict-probe-blocked-reason")
    parser.add_argument("--rule-version", choices=["v1", "v2", "v2_1", "compare"], default="v1")
    parser.add_argument("--comparison-v1")
    parser.add_argument("--comparison-v2")
    args = parser.parse_args()
    manifest = json.loads((ROOT / args.manifest).read_text(encoding="utf-8"))
    if args.rule_version == "compare":
        if not args.comparison_v1 or not args.comparison_v2:
            parser.error("compare requires --comparison-v1 and --comparison-v2 to avoid duplicate engine replay")
        v1 = json.loads((ROOT / args.comparison_v1).read_text(encoding="utf-8"))
        v2 = json.loads((ROOT / args.comparison_v2).read_text(encoding="utf-8"))
        report = compare_reports(v1, v2)
    else:
        report = build_report(manifest, args.strict_probe_blocked_reason, rule_version=args.rule_version)
    payload = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    if args.output:
        output_path = ROOT / args.output
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(payload, encoding="utf-8")
    print(payload, end="")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())