jesse-ux
e776cf9d83
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.
342 lines
14 KiB
Python
342 lines
14 KiB
Python
"""Self-contained 七政四余 chart adapter.
|
|
|
|
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]
|
|
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"}
|
|
ALLOWED_KETU_MODES = frozenset({"apogee", "descending-node"})
|
|
REQUIRED_BODIES = (
|
|
"sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn",
|
|
"rahu", "ketu", "yuebei", "purpleQi",
|
|
)
|
|
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({sevenGovernors:chart}));"
|
|
)
|
|
|
|
|
|
class QizhengChartError(ValueError):
|
|
"""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 _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:
|
|
result = float(value)
|
|
except (TypeError, ValueError) as 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 _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:
|
|
result = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
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 _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(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("_", "-")
|
|
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(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):
|
|
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]
|
|
return value
|
|
|
|
|
|
def _node_bin() -> str:
|
|
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.js runtime is unavailable", error_code="ERR_QIZHENG_NODE_MISSING")
|
|
return found
|
|
|
|
|
|
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")
|
|
return _run_node(command, timeout_seconds=timeout_seconds)
|
|
|
|
|
|
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,
|
|
}
|
|
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 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"] = 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": {
|
|
"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": 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": 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,
|
|
}
|
|
|
|
|
|
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",
|
|
]
|