jesse-ux
25851dd338
Add Apache-2.0 @4n6h4x0r/stem-branch 0.8.0, a seven-governors adapter, and POST /api/qizheng, /api/western, /api/ephemeris_events. BUG-700 remains blocked (do not call --pillars). BUG-701 and BUG-702 are resolved. BUG-703 is investigating (panchanga Lahiri). Skill is not bumped.
256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Native 七政四余 adapter over vendored @4n6h4x0r/stem-branch 0.8.0.
|
|
|
|
Only the seven-governors library entry is invoked. The CLI subcommands
|
|
--pillars, --luck, --polaris, --qimen, and --liuren are never called.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
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
|
|
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"
|
|
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",
|
|
})
|
|
DIGNITY_CLOSURE = {
|
|
"status": "unclosed",
|
|
"may_enter_conclusions": False,
|
|
"runtime_promotable_count": 0,
|
|
"closed_cells": 9,
|
|
"total_cells": 132,
|
|
"closure_ref": "references/oracle/qizheng_runtime_truth_closure_status_2026_08_29.json",
|
|
}
|
|
BOUNDARY = (
|
|
"已生成七政、四余、二十八宿、十二宫、命宫与相位。"
|
|
"宿度自角宿初度起算，不是 Raman/Lahiri 恒星黄经，也不是回归黄道；"
|
|
"三套坐标系不得互相换算，也不得叠加成双重印证。"
|
|
"庙旺判定未闭合（十一体×十二宫共 132 格仅 9 格有直接证据，"
|
|
"runtime_promotable_count = 0），不得进入解读、报告或咨询结论。"
|
|
)
|
|
_NODE_EVAL = (
|
|
"const fs=require('fs');"
|
|
"const payload=JSON.parse(fs.readFileSync(0,'utf8'));"
|
|
"const lib=require(payload.library);"
|
|
"const chart=lib.getSevenGovernorsChart("
|
|
"new Date(payload.iso),"
|
|
"{lat:payload.lat,lon:payload.lon},"
|
|
"{ketuMode:payload.ketuMode,siderealMode:payload.siderealMode}"
|
|
");"
|
|
"process.stdout.write(JSON.stringify(chart));"
|
|
)
|
|
|
|
|
|
class QizhengChartError(ValueError):
|
|
"""Structured seven-governors adapter failure. Callers map this to HTTP 400."""
|
|
|
|
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:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise QizhengChartError(f"{key} must be an integer", error_code="ERR_QIZHENG_INPUT") from exc
|
|
|
|
|
|
def _as_float(value: Any, key: str) -> float:
|
|
try:
|
|
number = float(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
|
|
|
|
|
|
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)
|
|
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:
|
|
if raw in (None, ""):
|
|
return DEFAULT_KETU_MODE
|
|
key = str(raw).strip().lower().replace("_", "-")
|
|
if key not in ALLOWED_KETU_MODES:
|
|
raise QizhengChartError("ketu_mode must be apogee or descending-node", error_code="ERR_QIZHENG_INPUT")
|
|
return key
|
|
|
|
|
|
def _sidereal_mode(raw: Any) -> dict[str, Any]:
|
|
if raw in (None, ""):
|
|
return dict(DEFAULT_SIDEREAL_MODE)
|
|
if isinstance(raw, str):
|
|
key = raw.strip().lower()
|
|
if key == "modern":
|
|
return {"type": "modern"}
|
|
if key in {"classical", "kaiyuan"}:
|
|
return {"type": "classical", "epoch": "kaiyuan"}
|
|
if key == "chongzhen":
|
|
return {"type": "classical", "epoch": "chongzhen"}
|
|
raise QizhengChartError("sidereal_mode is not supported", error_code="ERR_QIZHENG_INPUT")
|
|
if not isinstance(raw, dict):
|
|
raise QizhengChartError("sidereal_mode is not supported", error_code="ERR_QIZHENG_INPUT")
|
|
kind = str(raw.get("type") or "").strip().lower()
|
|
if kind == "modern":
|
|
return {"type": "modern"}
|
|
if kind == "classical":
|
|
epoch = raw.get("epoch", "kaiyuan")
|
|
if epoch not in {"kaiyuan", "chongzhen"} and not isinstance(epoch, (int, float)):
|
|
raise QizhengChartError("sidereal_mode epoch is not supported", error_code="ERR_QIZHENG_INPUT")
|
|
return {"type": "classical", "epoch": epoch}
|
|
if kind == "ayanamsa":
|
|
try:
|
|
return {"type": "ayanamsa", "value": float(raw.get("value"))}
|
|
except (TypeError, ValueError) as exc:
|
|
raise QizhengChartError("sidereal_mode value must be a number", error_code="ERR_QIZHENG_INPUT") from exc
|
|
raise QizhengChartError("sidereal_mode is not supported", error_code="ERR_QIZHENG_INPUT")
|
|
|
|
|
|
def _sanitize_raw(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
dropped = {"cwd", "path", "argv", "command", "stderr", "stdout", "library", "cli", "executable"}
|
|
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")
|
|
if not found:
|
|
raise QizhengChartError("node runtime is not available", 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):
|
|
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
|
|
|
|
|
|
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),
|
|
"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 []
|
|
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")
|
|
dignities = dict(DIGNITY_CLOSURE)
|
|
dignities["values"] = raw.get("dignities") if isinstance(raw.get("dignities"), dict) else {}
|
|
return {
|
|
"success": True,
|
|
"endpoint": "qizheng",
|
|
"coordinate_system": COORDINATE_SYSTEM,
|
|
"calculation": {
|
|
"engine": ENGINE_NAME,
|
|
"engine_version": ENGINE_VERSION,
|
|
"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,
|
|
},
|
|
"bodies": bodies,
|
|
"palaces": palaces,
|
|
"ascendant": raw.get("ascendant"),
|
|
"aspects": raw.get("aspects") if isinstance(raw.get("aspects"), list) else [],
|
|
"dignities": dignities,
|
|
"boundary": BOUNDARY,
|
|
"raw_engine_output": _sanitize_raw(raw),
|
|
}
|