Jyotisha/tests/run_real_case_revalidation.py

#!/usr/bin/env python3
"""Revalidate public real-person chart fixtures against the local engine.

This is a chart-calculation regression gate, not an event_prediction_accuracy
claim. It checks sign-level and optional degree-level agreement for public
reference cases, while keeping known controversial_reference rows visible in
the JSON report.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
ENGINE = ROOT / "scripts" / "jyotish_engine.py"
PUBLIC_CASES_RELATIVE = "tests/celebrity_cases.json"
INDASTRO_CASES_RELATIVE = "tests/indastro_cases.json"
PUBLIC_CASES = ROOT / PUBLIC_CASES_RELATIVE
INDASTRO_CASES = ROOT / INDASTRO_CASES_RELATIVE

DEFAULT_MIN_PASS_RATE = 0.98
DEFAULT_DEGREE_TOLERANCE = 1.0
CONTROVERSIAL_HINTS = [
    "内部矛盾",
    "需进一步验证",
    "存在重大偏差",
    "存在约",
    "边界案例",
]


def load_cases(path: Path) -> list[dict[str, Any]]:
    return json.loads(path.read_text(encoding="utf-8"))


def run_engine(case: dict[str, Any], python: str) -> dict[str, Any]:
    cmd = [
        python,
        str(ENGINE),
        "chart",
        "--year",
        str(case["year"]),
        "--month",
        str(case["month"]),
        "--day",
        str(case["day"]),
        "--hour",
        str(case["hour"]),
        "--minute",
        str(case["minute"]),
        "--lat",
        str(case["lat"]),
        "--lon",
        str(case["lon"]),
        "--tz",
        str(case["tz"]),
    ]
    completed = subprocess.run(
        cmd,
        cwd=ROOT,
        text=True,
        capture_output=True,
        timeout=45,
        check=False,
    )
    if completed.returncode != 0:
        raise RuntimeError(completed.stderr.strip() or completed.stdout.strip())
    return json.loads(completed.stdout)


def is_controversial(case: dict[str, Any]) -> bool:
    note = case.get("tz_note", "")
    return any(hint in note for hint in CONTROVERSIAL_HINTS)


def add_check(
    checks: list[dict[str, Any]],
    label: str,
    expected: Any,
    actual: Any,
    *,
    gated: bool,
    tolerance: float | None = None,
) -> None:
    if expected in (None, ""):
        return
    if tolerance is None:
        passed = actual == expected
    else:
        passed = actual is not None and abs(float(actual) - float(expected)) <= tolerance
    checks.append(
        {
            "label": label,
            "expected": expected,
            "actual": actual,
            "passed": passed,
            "gated": gated,
        }
    )


def validate_public_case(case: dict[str, Any], chart: dict[str, Any]) -> dict[str, Any]:
    checks: list[dict[str, Any]] = []
    planets = chart.get("planets", {})
    add_check(checks, "lagna_sign", case.get("known_lagna"), chart.get("ascendant", {}).get("sign"), gated=True)
    add_check(checks, "sun_sign", case.get("known_sun_sign"), planets.get("Sun", {}).get("sign"), gated=True)
    add_check(checks, "moon_sign", case.get("known_moon_sign"), planets.get("Moon", {}).get("sign"), gated=True)
    return summarize_case(case, "public_reference", checks)


def validate_indastro_case(case: dict[str, Any], chart: dict[str, Any], tolerance: float) -> dict[str, Any]:
    checks: list[dict[str, Any]] = []
    planets = chart.get("planets", {})
    gated = not is_controversial(case)
    category = "controversial_reference" if not gated else "public_reference"
    asc = chart.get("ascendant", {})
    sun = planets.get("Sun", {})
    moon = planets.get("Moon", {})

    add_check(checks, "lagna_sign", case.get("expected_lagna"), asc.get("sign"), gated=gated)
    add_check(checks, "sun_sign", case.get("expected_sun"), sun.get("sign"), gated=gated)
    add_check(checks, "moon_sign", case.get("expected_moon"), moon.get("sign"), gated=gated)
    add_check(checks, "lagna_degree", case.get("expected_lagna_degree"), asc.get("degree_in_sign"), gated=False, tolerance=tolerance)
    add_check(checks, "sun_degree", case.get("expected_sun_degree"), sun.get("degree_in_sign"), gated=False, tolerance=tolerance)
    add_check(checks, "moon_degree", case.get("expected_moon_degree"), moon.get("degree_in_sign"), gated=False, tolerance=tolerance)

    result = summarize_case(case, category, checks)
    result["source"] = case.get("source", "Indastro.com")
    result["note"] = case.get("tz_note", "")
    return result


def summarize_case(case: dict[str, Any], category: str, checks: list[dict[str, Any]]) -> dict[str, Any]:
    gated = [check for check in checks if check["gated"]]
    return {
        "id": case["id"],
        "name": case["name"],
        "category": category,
        "passed_checks": sum(1 for check in checks if check["passed"]),
        "total_checks": len(checks),
        "gated_passed_checks": sum(1 for check in gated if check["passed"]),
        "gated_total_checks": len(gated),
        "checks": checks,
    }


def build_report(args: argparse.Namespace) -> dict[str, Any]:
    results: list[dict[str, Any]] = []
    for case in load_cases(PUBLIC_CASES):
        chart = run_engine(case, args.python)
        results.append(validate_public_case(case, chart))
    for case in load_cases(INDASTRO_CASES):
        chart = run_engine(case, args.python)
        results.append(validate_indastro_case(case, chart, args.degree_tolerance))

    total_checks = sum(item["total_checks"] for item in results)
    passed_checks = sum(item["passed_checks"] for item in results)
    gated_total_checks = sum(item["gated_total_checks"] for item in results)
    gated_passed_checks = sum(item["gated_passed_checks"] for item in results)
    pass_rate = gated_passed_checks / gated_total_checks if gated_total_checks else 1.0
    failures = [
        {
            "id": item["id"],
            "name": item["name"],
            "category": item["category"],
            "failed_checks": [check for check in item["checks"] if check["gated"] and not check["passed"]],
        }
        for item in results
        if any(check["gated"] and not check["passed"] for check in item["checks"])
    ]
    controversial = [item for item in results if item["category"] == "controversial_reference"]

    return {
        "valid": pass_rate >= args.min_pass_rate and not failures,
        "scope": "public real-person chart revalidation; not event_prediction_accuracy",
        "min_pass_rate": args.min_pass_rate,
        "pass_rate": round(pass_rate, 4),
        "passed_checks": passed_checks,
        "total_checks": total_checks,
        "gated_passed_checks": gated_passed_checks,
        "gated_total_checks": gated_total_checks,
        "controversial_reference_cases": len(controversial),
        "failures": failures,
        "results": results,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Run public real-case Jyotish chart revalidation")
    parser.add_argument("--python", default=sys.executable)
    parser.add_argument("--min-pass-rate", type=float, default=DEFAULT_MIN_PASS_RATE)
    parser.add_argument("--degree-tolerance", type=float, default=DEFAULT_DEGREE_TOLERANCE)
    parser.add_argument("--summary", action="store_true", help="Print a compact human-readable line before JSON")
    args = parser.parse_args()

    report = build_report(args)
    if args.summary:
        print(
            "真实案例复验: "
            f"gated={report['gated_passed_checks']}/{report['gated_total_checks']} "
            f"all={report['passed_checks']}/{report['total_checks']} "
            f"controversial={report['controversial_reference_cases']}"
        )
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0 if report["valid"] else 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except BrokenPipeError:
        raise SystemExit(0)