Jyotisha/scripts/research/api_service_extract_golden.py

"""Canonical hashes for the compute/HTTP class split.

Fictional smoke chart only. Clock fields are removed before hashing.
Absolute repository paths in string values are replaced with <ROOT>.
Payload values are not written; the golden file stores sha256 and key-path types.
"""
from __future__ import annotations

import hashlib
import json
import os
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"
GOLDEN_PATH = ROOT / "tests" / "golden" / "api_server_service_extract_v1.json"
CLOCK_KEYS = {"cache_created_at", "cache_expires_at"}
THEMES = ("career", "wealth", "relationship")
BASELINE_SHA = "fbf04f0b5f31cd8421bc4546840b1c833aaed810"
CACHE_TTL_ENV = "JYOTISH_API_CHART_CACHE_TTL_SECONDS"
ROOT_PLACEHOLDER = "<ROOT>"

# Synthetic positions reused from tests/test_offline_compute_mixins.py.
# Not a birth record.
EVIDENCE_CHART = {
    "success": True,
    "ascendant": {"sign": "Aries", "sign_idx": 0, "degree": 12.0, "lon": 12.0},
    "planets": {
        "Sun": {"sign": "Leo", "lon": 130.0},
        "Moon": {"sign": "Cancer", "lon": 100.0},
    },
}


def prepare_env() -> None:
    os.environ["VEDASTRO_ENABLE_NETWORK"] = "0"
    os.environ.setdefault("JYOTISH_SKIP_LOCAL_ENV", "1")
    os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
    for path in (str(ROOT), str(SCRIPTS)):
        if path not in sys.path:
            sys.path.insert(0, path)


@contextmanager
def chart_cache_disabled() -> Iterator[None]:
    """Turn the shared chart cache off. Do not delete the repository directory."""
    previous = os.environ.get(CACHE_TTL_ENV)
    os.environ[CACHE_TTL_ENV] = "0"
    try:
        yield
    finally:
        if previous is None:
            os.environ.pop(CACHE_TTL_ENV, None)
        else:
            os.environ[CACHE_TTL_ENV] = previous


def root_spellings(root: Path = ROOT) -> tuple[str, ...]:
    native = os.path.normpath(str(root))
    forward = native.replace("\\", "/")
    spellings = {native, forward}
    if os.name == "nt":
        spellings.add("\\\\?\\" + native)
    return tuple(sorted(spellings, key=len, reverse=True))


def neutralize_path_string(text: str, root: Path = ROOT) -> str:
    for spelling in root_spellings(root):
        text = text.replace(spelling, ROOT_PLACEHOLDER)
    if ROOT_PLACEHOLDER not in text:
        return text
    pieces = text.split(ROOT_PLACEHOLDER)
    rebuilt = [pieces[0]]
    for piece in pieces[1:]:
        if piece.startswith("\\") or piece.startswith("/"):
            end = 0
            while end < len(piece) and piece[end] not in " \t\r\n\"'":
                end += 1
            piece = piece[:end].replace("\\", "/") + piece[end:]
        rebuilt.append(piece)
    return ROOT_PLACEHOLDER.join(rebuilt)


def neutralize_paths(value: Any, root: Path = ROOT) -> Any:
    if isinstance(value, str):
        return neutralize_path_string(value, root)
    if isinstance(value, dict):
        return {key: neutralize_paths(item, root) for key, item in value.items()}
    if isinstance(value, list):
        return [neutralize_paths(item, root) for item in value]
    return value


def theme_body(theme: str) -> dict[str, Any]:
    """Fictional smoke chart from the task brief. Coordinates only, no person."""
    return {
        "year": 1990,
        "month": 1,
        "day": 1,
        "hour": 12,
        "minute": 0,
        "second": 0,
        "lat": 28.6139,
        "lon": 77.2090,
        "tz": 5.5,
        "theme": [theme],
        "defer_optional_external_evidence": True,
        "today": "2026-10-13",
    }


def strip_clock(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: strip_clock(item)
            for key, item in value.items()
            if key not in CLOCK_KEYS
        }
    if isinstance(value, list):
        return [strip_clock(item) for item in value]
    return value


def json_type(value: Any) -> str:
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    return type(value).__name__


def flatten_keypaths(value: Any, prefix: str = "", *, depth: int = 0, max_depth: int = 2) -> dict[str, str]:
    out: dict[str, str] = {}
    label = prefix or "$"
    out[label] = json_type(value)
    if depth >= max_depth:
        return out
    if isinstance(value, dict):
        for key in sorted(item for item in value if isinstance(item, str)):
            child = f"{prefix}.{key}" if prefix else key
            out.update(flatten_keypaths(value[key], child, depth=depth + 1, max_depth=max_depth))
    elif isinstance(value, list) and value:
        child = f"{prefix}[]" if prefix else "[]"
        out.update(flatten_keypaths(value[0], child, depth=depth + 1, max_depth=max_depth))
    return out


def canonical_text(value: Any) -> str:
    return json.dumps(neutralize_paths(strip_clock(value)), sort_keys=True, ensure_ascii=False)


def digest(value: Any) -> dict[str, Any]:
    text = canonical_text(value)
    return {
        "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
        "canonicalBytes": len(text.encode("utf-8")),
        "keypaths": flatten_keypaths(strip_clock(value)),
    }


def capture_mcp(theme: str) -> Any:
    prepare_env()
    with chart_cache_disabled():
        from consultation_workflow_service import execute_consultation_workflow

        return execute_consultation_workflow(theme_body(theme), surface="skill_mcp")


def capture_http(theme: str) -> Any:
    """HTTP handler method, on the compute class after the split."""
    prepare_env()
    with chart_cache_disabled():
        from jyotish_api_server import JyotishComputeService

        return JyotishComputeService()._compute_consultation_workflow(theme_body(theme))


def capture_repairs() -> dict[str, Any]:
    prepare_env()
    with chart_cache_disabled():
        from capture_report_blocked_repairs_golden import (
            BIRTH,
            MODULE_KEEP,
            THEMES as REPAIR_THEMES,
            VARGA_KEEP,
            _body,
            _handler,
            _trim_workflow,
        )

        payload: dict[str, Any] = {
            "source": "scripts/jyotish_api_server.py _compute_consultation_workflow",
            "smoke_birth": {**BIRTH, "note": "fictional capture identity; not a real person"},
            "trim": {
                "keep_varga_keys": list(VARGA_KEEP),
                "keep_modules": list(MODULE_KEEP),
                "drop": [
                    "ai_prompt_pack",
                    "vedastro",
                    "varga_spectrum",
                    "guided_topics",
                    "western_spectrum",
                    "technique_audit_table",
                ],
            },
            "themes": {},
        }
        for theme in REPAIR_THEMES:
            raw = _handler()._compute_consultation_workflow(_body(theme))
            payload["themes"][theme] = _trim_workflow(raw)
        return payload


def capture_evidence() -> Any:
    prepare_env()
    from consultation_workflow_service import build_runtime_evidence_helpers

    return build_runtime_evidence_helpers(EVIDENCE_CHART)


def capture_cases(*, include_http: bool = True, include_repairs: bool = True) -> dict[str, Any]:
    cases: dict[str, Any] = {}
    for theme in THEMES:
        cases[f"mcp.{theme}"] = capture_mcp(theme)
        if include_http:
            cases[f"http.{theme}"] = capture_http(theme)
    if include_repairs:
        cases["repairs"] = capture_repairs()
    cases["evidence"] = capture_evidence()
    return cases


def summarize(cases: dict[str, Any]) -> dict[str, Any]:
    return {name: digest(value) for name, value in cases.items()}


def provenance() -> dict[str, Any]:
    return {
        "fictional": True,
        "command": "python scripts/research/api_service_extract_golden.py --check",
        "baselineSha": BASELINE_SHA,
        "env": {
            "VEDASTRO_ENABLE_NETWORK": "0",
            "JYOTISH_SKIP_LOCAL_ENV": "1",
            "JYOTISH_API_CHART_CACHE_TTL_SECONDS": "0",
        },
        "clockFieldsRemoved": sorted(CLOCK_KEYS),
        "rootPlaceholder": ROOT_PLACEHOLDER,
        "mcpSurface": "skill_mcp",
        "note": (
            "fictional smoke chart from the task brief; payload values are not stored; "
            "repository absolute paths in string values are replaced with <ROOT>"
        ),
    }


def document_from_summary(summary: dict[str, Any]) -> dict[str, Any]:
    return {
        "kind": "sha256_and_keypath_types",
        "fixtureProvenance": provenance(),
        "cases": summary,
    }


def write_golden(summary: dict[str, Any], path: Path = GOLDEN_PATH) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(document_from_summary(summary), ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    path.write_text(text, encoding="utf-8")


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


def check_golden(path: Path = GOLDEN_PATH) -> list[str]:
    expected = load_golden(path)["cases"]
    actual = summarize(capture_cases())
    mismatches: list[str] = []
    if set(expected) != set(actual):
        mismatches.append(f"case names {sorted(expected)} != {sorted(actual)}")
    for name in sorted(set(expected) & set(actual)):
        if expected[name]["sha256"] != actual[name]["sha256"]:
            mismatches.append(
                f"{name} sha256 {expected[name]['sha256']} != {actual[name]['sha256']} "
                f"bytes {expected[name]['canonicalBytes']} != {actual[name]['canonicalBytes']}"
            )
        elif expected[name]["keypaths"] != actual[name]["keypaths"]:
            mismatches.append(f"{name} keypaths differ")
    return mismatches


def main(argv: list[str]) -> int:
    prepare_env()
    if "--check" in argv:
        mismatches = check_golden()
        if mismatches:
            print("MISMATCH")
            for item in mismatches:
                print(item)
            return 1
        print("MATCH", len(load_golden()["cases"]))
        return 0
    summary = summarize(capture_cases())
    for name, row in summary.items():
        print(f"{name} {row['sha256']} {row['canonicalBytes']}")
    if "--write" in argv:
        write_golden(summary)
        print("WROTE", GOLDEN_PATH)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))