fix(qizheng): copy vendored stem-branch files from local upstream checkout

Read G:/Ferti/yinduzhanxing-codex-add-birth-time-rectification-skill (a911c890) and copy the listed vendor files by content. Rebuild the adapter from the 168-line CLI wrapper, then apply school parameters, boundary rewrite, and sanitized errors.

scripts/qizheng_chart_engine.py

@@ -1,32 +1,35 @@
-#!/usr/bin/env python3
-"""Native 七政四余 adapter over vendored @4n6h4x0r/stem-branch 0.8.0.
+"""Self-contained 七政四余 chart adapter.
 
-Only the seven-governors library entry is invoked. The CLI subcommands
---pillars, --luck, --polaris, --qimen, and --liuren are never called.
+The vendored CLI is the Apache-2.0 stem-branch seven-governors implementation.
+This module owns the application contract and keeps the upstream raw chart
+visible for audit. Starting point is the 168-line native adapter; school
+parameters, boundary text, coordinate labels, and error sanitization are
+product requirements on top of that file.
 """
+
 from __future__ import annotations
 
 import json
 import os
 import shutil
 import subprocess
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
+
 REPO_ROOT = Path(__file__).resolve().parents[1]
-CLI_PATH = REPO_ROOT / "vendor" / "stem-branch" / "dist" / "cli.cjs"
-LIB_PATH = REPO_ROOT / "vendor" / "stem-branch" / "dist" / "index.cjs"
-ENGINE_TIMEOUT_SECONDS = 20
+VENDORED_CLI = REPO_ROOT / "vendor" / "stem-branch" / "dist" / "cli.cjs"
+VENDORED_LIB = REPO_ROOT / "vendor" / "stem-branch" / "dist" / "index.cjs"
+DEFAULT_TIMEOUT_SECONDS = 20
 COORDINATE_SYSTEM = "qizheng_mansion_degrees_from_jiao"
 DEFAULT_KETU_MODE = "apogee"
 DEFAULT_SIDEREAL_MODE: dict[str, Any] = {"type": "modern"}
-ENGINE_NAME = "@4n6h4x0r/stem-branch"
-ENGINE_VERSION = "0.8.0"
+ALLOWED_KETU_MODES = frozenset({"apogee", "descending-node"})
 REQUIRED_BODIES = (
     "sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn",
     "rahu", "ketu", "yuebei", "purpleQi",
 )
-ALLOWED_KETU_MODES = frozenset({"apogee", "descending-node"})
 FORBIDDEN_CLI_FLAGS = frozenset({
     "--pillars", "--luck", "--polaris", "--qimen", "--liuren", "--chuanren",
 })
@@ -54,57 +57,74 @@ _NODE_EVAL = (
     "{lat:payload.lat,lon:payload.lon},"
     "{ketuMode:payload.ketuMode,siderealMode:payload.siderealMode}"
     ");"
-    "process.stdout.write(JSON.stringify(chart));"
+    "process.stdout.write(JSON.stringify({sevenGovernors:chart}));"
 )
 
 
 class QizhengChartError(ValueError):
-    """Structured seven-governors adapter failure. Callers map this to HTTP 400."""
+    """Raised when the native 七政四余 chart cannot be computed."""
 
     def __init__(self, message: str, *, error_code: str = "ERR_QIZHENG_ENGINE") -> None:
         super().__init__(message)
         self.error_code = error_code
 
 
-def _field(body: dict[str, Any], *keys: str, required: bool = True) -> Any:
-    for key in keys:
-        if key in body and body[key] not in (None, ""):
-            return body[key]
-    if required:
-        raise QizhengChartError(f"{keys[0]} is required", error_code="ERR_QIZHENG_INPUT")
-    return None
-
-
-def _as_int(value: Any, key: str) -> int:
+def _number(value: Any, name: str, minimum: float, maximum: float) -> float:
+    if value in (None, ""):
+        raise QizhengChartError(f"{name} is required", error_code="ERR_QIZHENG_INPUT")
     try:
-        return int(value)
+        result = float(value)
     except (TypeError, ValueError) as exc:
-        raise QizhengChartError(f"{key} must be an integer", error_code="ERR_QIZHENG_INPUT") from exc
+        raise QizhengChartError(f"{name} must be a number", error_code="ERR_QIZHENG_INPUT") from exc
+    if not minimum <= result <= maximum:
+        raise QizhengChartError(f"{name} must be between {minimum} and {maximum}", error_code="ERR_QIZHENG_INPUT")
+    return result
 
 
-def _as_float(value: Any, key: str) -> float:
+def _required_int(body: dict[str, Any], name: str, minimum: int, maximum: int, default: int | None = None) -> int:
+    value = body.get(name, default)
+    if value in (None, "") and default is not None:
+        value = default
     try:
-        number = float(value)
+        result = int(value)
     except (TypeError, ValueError) as exc:
-        raise QizhengChartError(f"{key} must be a number", error_code="ERR_QIZHENG_INPUT") from exc
-    if number != number or number in {float("inf"), float("-inf")}:
-        raise QizhengChartError(f"{key} must be finite", error_code="ERR_QIZHENG_INPUT")
-    return number
+        raise QizhengChartError(f"{name} must be an integer", error_code="ERR_QIZHENG_INPUT") from exc
+    if not minimum <= result <= maximum:
+        raise QizhengChartError(f"{name} must be between {minimum} and {maximum}", error_code="ERR_QIZHENG_INPUT")
+    return result
 
 
-def _iso_local(year: int, month: int, day: int, hour: int, minute: int, second: int, tz: float) -> str:
-    """ISO local civil time with numeric offset. Correct for astronomy; unsafe for pillars."""
-    total_minutes = int(round(tz * 60))
-    sign = "+" if total_minutes >= 0 else "-"
-    abs_minutes = abs(total_minutes)
-    offset_hours, offset_minutes = divmod(abs_minutes, 60)
+def _timezone_offset(body: dict[str, Any]) -> float:
+    raw = body.get("tz", body.get("timezone_offset", body.get("timezone")))
+    if raw in (None, ""):
+        raise QizhengChartError("tz is required", error_code="ERR_QIZHENG_INPUT")
+    return _number(raw, "tz", -14, 14)
+
+
+def _iso_local(body: dict[str, Any]) -> str:
+    year = _required_int(body, "year", 1, 9999)
+    month = _required_int(body, "month", 1, 12)
+    day = _required_int(body, "day", 1, 31)
+    hour = _required_int(body, "hour", 0, 23)
+    minute = _required_int(body, "minute", 0, 59)
+    second = _required_int(body, "second", 0, 59, default=0)
+    offset = _timezone_offset(body)
+    datetime(year, month, day, hour, minute, second)
+    sign = "+" if offset >= 0 else "-"
+    absolute = abs(offset)
+    offset_hours = int(absolute)
+    offset_minutes = round((absolute - offset_hours) * 60)
+    if offset_minutes == 60:
+        offset_hours += 1
+        offset_minutes = 0
     return (
         f"{year:04d}-{month:02d}-{day:02d}T{hour:02d}:{minute:02d}:{second:02d}"
         f"{sign}{offset_hours:02d}:{offset_minutes:02d}"
     )
 
 
-def _ketu_mode(raw: Any) -> str:
+def _ketu_mode(body: dict[str, Any]) -> str:
+    raw = body.get("ketu_mode", body.get("ketuMode"))
     if raw in (None, ""):
         return DEFAULT_KETU_MODE
     key = str(raw).strip().lower().replace("_", "-")
@@ -113,7 +133,8 @@ def _ketu_mode(raw: Any) -> str:
     return key
 
 
-def _sidereal_mode(raw: Any) -> dict[str, Any]:
+def _sidereal_mode(body: dict[str, Any]) -> dict[str, Any]:
+    raw = body.get("sidereal_mode", body.get("siderealMode"))
     if raw in (None, ""):
         return dict(DEFAULT_SIDEREAL_MODE)
     if isinstance(raw, str):
@@ -149,107 +170,172 @@ def _sanitize_raw(value: Any) -> Any:
         return {key: _sanitize_raw(item) for key, item in value.items() if key not in dropped}
     if isinstance(value, list):
         return [_sanitize_raw(item) for item in value]
-    if isinstance(value, str) and ("\\" in value or value.startswith("/")) and ("vendor" in value or "stem-branch" in value):
-        return "[redacted-path]"
     return value
 
 
 def _node_bin() -> str:
-    found = shutil.which("node") or shutil.which("node.exe")
+    configured = os.environ.get("NODE_BINARY", "node")
+    found = shutil.which(configured) or shutil.which("node") or shutil.which("node.exe")
     if not found:
-        raise QizhengChartError("node runtime is not available", error_code="ERR_QIZHENG_NODE_MISSING")
+        raise QizhengChartError("Node.js runtime is unavailable", error_code="ERR_QIZHENG_NODE_MISSING")
     return found
 
 
-def _run_engine(payload: dict[str, Any]) -> dict[str, Any]:
-    if not CLI_PATH.is_file() or not LIB_PATH.is_file():
-        raise QizhengChartError("vendored seven-governors engine is missing", error_code="ERR_QIZHENG_CLI_MISSING")
-    node_bin = _node_bin()
-    argv = [node_bin, "-e", _NODE_EVAL]
-    if FORBIDDEN_CLI_FLAGS.intersection(argv):
+def _run_cli(iso: str, lat: float, lon: float, timeout_seconds: int) -> dict[str, Any]:
+    command = [
+        _node_bin(),
+        str(VENDORED_CLI),
+        "--date",
+        iso,
+        "--lat",
+        str(lat),
+        "--lng",
+        str(lon),
+        "--seven-governors",
+        "--json",
+    ]
+    if FORBIDDEN_CLI_FLAGS.intersection(command):
         raise QizhengChartError("forbidden engine subcommand blocked", error_code="ERR_QIZHENG_ENGINE")
-    try:
-        completed = subprocess.run(
-            argv,
-            input=json.dumps(payload, ensure_ascii=False),
-            capture_output=True,
-            text=True,
-            encoding="utf-8",
-            timeout=ENGINE_TIMEOUT_SECONDS,
-            cwd=str(REPO_ROOT),
-            check=False,
-        )
-    except FileNotFoundError as exc:
-        raise QizhengChartError("node runtime is not available", error_code="ERR_QIZHENG_NODE_MISSING") from exc
-    except subprocess.TimeoutExpired as exc:
-        raise QizhengChartError("seven-governors engine timed out", error_code="ERR_QIZHENG_TIMEOUT") from exc
-    if completed.returncode != 0:
-        raise QizhengChartError("seven-governors engine exited with an error", error_code="ERR_QIZHENG_ENGINE_EXIT")
-    stdout = (completed.stdout or "").strip()
-    if not stdout:
-        raise QizhengChartError("seven-governors engine returned empty output", error_code="ERR_QIZHENG_BAD_JSON")
-    try:
-        parsed = json.loads(stdout)
-    except json.JSONDecodeError as exc:
-        raise QizhengChartError("seven-governors engine returned non-JSON output", error_code="ERR_QIZHENG_BAD_JSON") from exc
-    if not isinstance(parsed, dict):
-        raise QizhengChartError("seven-governors engine returned non-JSON output", error_code="ERR_QIZHENG_BAD_JSON")
-    return parsed
+    return _run_node(command, timeout_seconds=timeout_seconds)
 
 
-def build_qizheng_natal_chart(body: dict[str, Any] | None) -> dict[str, Any]:
-    if not isinstance(body, dict):
-        raise QizhengChartError("JSON body must be an object", error_code="ERR_QIZHENG_INPUT")
-    year = _as_int(_field(body, "year"), "year")
-    month = _as_int(_field(body, "month"), "month")
-    day = _as_int(_field(body, "day"), "day")
-    hour = _as_int(_field(body, "hour"), "hour")
-    minute = _as_int(_field(body, "minute"), "minute")
-    second = _as_int(_field(body, "second", required=False) or 0, "second")
-    lat = _as_float(_field(body, "lat", "latitude"), "lat")
-    lon = _as_float(_field(body, "lon", "longitude", "lng"), "lon")
-    tz = _as_float(_field(body, "tz", "timezone"), "tz")
-    ketu_mode = _ketu_mode(body.get("ketu_mode", body.get("ketuMode")))
-    sidereal_mode = _sidereal_mode(body.get("sidereal_mode", body.get("siderealMode")))
-    iso = _iso_local(year, month, day, hour, minute, second, tz)
-    raw = _run_engine({
-        "library": os.fspath(LIB_PATH),
+def _run_library(iso: str, lat: float, lon: float, ketu_mode: str, sidereal_mode: dict[str, Any], timeout_seconds: int) -> dict[str, Any]:
+    if not VENDORED_LIB.is_file():
+        raise QizhengChartError("vendored 七政四余 library is unavailable", error_code="ERR_QIZHENG_CLI_MISSING")
+    payload = {
+        "library": os.fspath(VENDORED_LIB),
         "iso": iso,
         "lat": lat,
         "lon": lon,
         "ketuMode": ketu_mode,
         "siderealMode": sidereal_mode,
-    })
-    bodies = raw.get("bodies") if isinstance(raw.get("bodies"), dict) else {}
-    palaces = raw.get("palaces") if isinstance(raw.get("palaces"), list) else []
+    }
+    return _run_node(
+        [_node_bin(), "-e", _NODE_EVAL],
+        timeout_seconds=timeout_seconds,
+        stdin=json.dumps(payload, ensure_ascii=False),
+    )
+
+
+def _run_node(command: list[str], *, timeout_seconds: int, stdin: str | None = None) -> dict[str, Any]:
+    try:
+        completed = subprocess.run(
+            command,
+            cwd=str(REPO_ROOT),
+            input=stdin,
+            check=False,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            timeout=timeout_seconds,
+        )
+    except FileNotFoundError as exc:
+        raise QizhengChartError("Node.js runtime is unavailable", error_code="ERR_QIZHENG_NODE_MISSING") from exc
+    except subprocess.TimeoutExpired as exc:
+        raise QizhengChartError("七政四余计算超时", error_code="ERR_QIZHENG_TIMEOUT") from exc
+    if completed.returncode != 0:
+        raise QizhengChartError("seven-governors engine exited with an error", error_code="ERR_QIZHENG_ENGINE_EXIT")
+    stdout = (completed.stdout or "").strip()
+    if not stdout:
+        raise QizhengChartError("native 七政四余 output is not valid JSON", error_code="ERR_QIZHENG_BAD_JSON")
+    try:
+        raw = json.loads(stdout)
+    except json.JSONDecodeError as exc:
+        raise QizhengChartError("native 七政四余 output is not valid JSON", error_code="ERR_QIZHENG_BAD_JSON") from exc
+    if not isinstance(raw, dict):
+        raise QizhengChartError("native 七政四余 output is not valid JSON", error_code="ERR_QIZHENG_BAD_JSON")
+    return raw
+
+
+def _normalize(raw: dict[str, Any], body: dict[str, Any], *, ketu_mode: str, sidereal_mode: dict[str, Any], iso: str) -> dict[str, Any]:
+    chart = raw.get("sevenGovernors")
+    if not isinstance(chart, dict):
+        raise QizhengChartError("native seven-governors output is missing", error_code="ERR_QIZHENG_ENGINE")
+    bodies = chart.get("bodies") if isinstance(chart.get("bodies"), dict) else {}
+    palaces = chart.get("palaces") if isinstance(chart.get("palaces"), list) else []
     missing = [name for name in REQUIRED_BODIES if name not in bodies]
-    if missing:
-        raise QizhengChartError("seven-governors engine returned an incomplete chart", error_code="ERR_QIZHENG_ENGINE")
-    if len(palaces) != 12:
-        raise QizhengChartError("seven-governors engine returned an incomplete chart", error_code="ERR_QIZHENG_ENGINE")
-    if not isinstance(raw.get("ascendant"), dict):
-        raise QizhengChartError("seven-governors engine returned an incomplete chart", error_code="ERR_QIZHENG_ENGINE")
+    if missing or len(palaces) != 12 or not isinstance(chart.get("ascendant"), dict):
+        raise QizhengChartError("native seven-governors output is missing", error_code="ERR_QIZHENG_ENGINE")
     dignities = dict(DIGNITY_CLOSURE)
-    dignities["values"] = raw.get("dignities") if isinstance(raw.get("dignities"), dict) else {}
+    dignities["values"] = chart.get("dignities") if isinstance(chart.get("dignities"), dict) else {}
     return {
+        "schema": "qizheng_native_chart.v1",
         "success": True,
         "endpoint": "qizheng",
+        "status": "executed",
         "coordinate_system": COORDINATE_SYSTEM,
+        "engine": "stem-branch-seven-governors",
+        "engine_version": "0.8.0",
+        "source": {
+            "project": "@4n6h4x0r/stem-branch",
+            "license": "Apache-2.0",
+            "vendored_cli": "vendor/stem-branch/dist/cli.cjs",
+            "source_boundary": "Native chart computation only; interpretive rules remain separately audited.",
+        },
         "calculation": {
-            "engine": ENGINE_NAME,
-            "engine_version": ENGINE_VERSION,
+            "birth_iso": iso,
+            "iso_local": iso,
+            "latitude": float(body["lat"]),
+            "longitude": float(body["lon"]),
             "ketu_mode": ketu_mode,
             "sidereal_mode": sidereal_mode,
-            "engine_ketu_mode": raw.get("ketuMode"),
-            "engine_sidereal_mode": raw.get("siderealMode"),
-            "iso_local": iso,
-            "timeout_seconds": ENGINE_TIMEOUT_SECONDS,
+            "engine_ketu_mode": chart.get("ketuMode"),
+            "engine_sidereal_mode": chart.get("siderealMode"),
+            "timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
         },
+        "raw_engine_output": _sanitize_raw(raw),
         "bodies": bodies,
         "palaces": palaces,
-        "ascendant": raw.get("ascendant"),
-        "aspects": raw.get("aspects") if isinstance(raw.get("aspects"), list) else [],
+        "ascendant": chart.get("ascendant"),
+        "aspects": chart.get("aspects") or [],
         "dignities": dignities,
+        "chart": {
+            "date": chart.get("date"),
+            "location": chart.get("location"),
+            "sidereal_mode": chart.get("siderealMode"),
+            "ketu_mode": chart.get("ketuMode"),
+            "ascendant": chart.get("ascendant"),
+            "bodies": bodies,
+            "palaces": palaces,
+            "aspects": chart.get("aspects") or [],
+            "dignities": dignities,
+        },
+        "counts": {
+            "body_count": len(bodies),
+            "palace_count": len(palaces),
+            "aspect_count": len(chart.get("aspects") or []),
+        },
         "boundary": BOUNDARY,
-        "raw_engine_output": _sanitize_raw(raw),
     }
+
+
+def calculate_qizheng_chart(body: dict[str, Any], *, timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS) -> dict[str, Any]:
+    """Compute a complete native 七政四余 natal chart from JSON-like input."""
+    if not isinstance(body, dict):
+        raise QizhengChartError("request body must be an object", error_code="ERR_QIZHENG_INPUT")
+    if not VENDORED_CLI.exists():
+        raise QizhengChartError("vendored 七政四余 engine is unavailable", error_code="ERR_QIZHENG_CLI_MISSING")
+    lat = _number(body.get("lat", body.get("latitude")), "lat", -90, 90)
+    lon = _number(body.get("lon", body.get("longitude", body.get("lng"))), "lon", -180, 180)
+    iso = _iso_local(body)
+    ketu_mode = _ketu_mode(body)
+    sidereal_mode = _sidereal_mode(body)
+    uses_defaults = ketu_mode == DEFAULT_KETU_MODE and sidereal_mode == DEFAULT_SIDEREAL_MODE
+    if uses_defaults:
+        raw = _run_cli(iso, lat, lon, timeout_seconds)
+    else:
+        raw = _run_library(iso, lat, lon, ketu_mode, sidereal_mode, timeout_seconds)
+    return _normalize(raw, {**body, "lat": lat, "lon": lon}, ketu_mode=ketu_mode, sidereal_mode=sidereal_mode, iso=iso)
+
+
+def build_qizheng_natal_chart(body: dict[str, Any]) -> dict[str, Any]:
+    """Alias used by the thin API registration."""
+    return calculate_qizheng_chart(body)
+
+
+__all__ = [
+    "QizhengChartError",
+    "calculate_qizheng_chart",
+    "build_qizheng_natal_chart",
+    "VENDORED_CLI",
+]