fix(qizheng): copy vendored stem-branch files from local upstream checkout
Independent Staging Quality Gate / validate (push) Canceled after 4m51s
Independent Staging Quality Gate / publish (push) Canceled after 0s

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.
This commit is contained in:
jesse-ux
2026-09-15 16:21:04 +08:00
parent 25851dd338
commit e776cf9d83
8 changed files with 225 additions and 137 deletions
+1 -1
View File
@@ -4,7 +4,7 @@
- **无 Docker** 本机 `docker` 不在 PATH。无法构建 `deploy/railway-api.Dockerfile`,也无法在容器内跑 `node --version` 或断言 `Path('/app/vendor/stem-branch/dist/cli.cjs').exists()`。**不得把镜像构建或容器内验证写成通过。** Dockerfile 已按任务加上 `nodejs` 运行时与 `COPY vendor ./vendor`,待有 Docker 的环境验证。
- **替代证据(宿主机):** `node --version` = `v22.23.2`。vendored `dist/cli.cjs` 45,963 行 / 2,682,464 字节,与任务书 45,963 行 / 2.68 MB 一致。用任务书虚构资料 `1990-04-09T13:24:00+08:00 / 31.19N 121.44E` 实跑 `getSevenGovernorsChart`:11 曜、12 宫、命宫星宿·午宫、太阳宿度 177.94 奎 7.44°、相位 27、神煞 0。
- **上游 Git 仓不可取:** `https://github.com/732642856/yinduzhanxing``a911c890` 本机 `git ls-remote` HTTPS 认证失败,SSH 无密钥,网页 404。`vendor/stem-branch/**` 取自 npm 发布包 `@4n6h4x0r/stem-branch@0.8.0`Apache-2.0),不是从该 commit 原样 `git show``scripts/qizheng_chart_engine.py` 按任务 2 合同在本仓实现(派别参数、boundary 重写、坐标系标注、五种错误路径),不是 168 行原文件的逐字拷贝。CLI 不暴露 `ketuMode`/`siderealMode`故额外 vendored `dist/index.cjs` 以支持 BUG-701
- **上游 Git 仓不再从 GitHub 拉。** 本机目录 `G:\Ferti\yinduzhanxing-codex-add-birth-time-rectification-skill` 即任务书 `/workspace/yinduzhanxing`(分支 `codex/add-birth-time-rectification-skill` / `a911c890`)。任务书列出的 `vendor/stem-branch/{LICENSE,README.md,package.json,dist/cli.cjs}` 已按内容拷贝到本 worktree 同名路径,运行时只读仓库内相对路径`scripts/qizheng_chart_engine.py` 以上游 168 行适配器为起点,按任务 2 改了派别参数、boundary、坐标系和错误路径。CLI 不暴露 `ketuMode`/`siderealMode`默认走 `cli.cjs`,覆盖派别时用同包 `dist/index.cjs`
- **本机无仓库 `.venv` 目录:** 主仓 `.venv` 是指向 `/workspace/Jyotisha/.venv` 的 git symlinkWindows 检出为 25 字节指针文件。测试与预检使用 Anaconda Python 3.11.7(已有 `pytest``pyswisseph`)。`timezonefinder` 仍缺,与既有 BLOCKED 条目一致。
## ~~快速门 timezone 推断依赖~~2026-09-15 解除:验收机已装 `timezonefinder 9.0.0`
+6 -4
View File
@@ -7,16 +7,18 @@ This product includes @4n6h4x0r/stem-branch 0.8.0 (Apache License 2.0).
Upstream: https://github.com/h4x0r/stem-branch
npm: https://www.npmjs.com/package/@4n6h4x0r/stem-branch
The following files are vendored unmodified from the published 0.8.0 tarball:
The following files were copied from the local upstream checkout
`yinduzhanxing` branch `codex/add-birth-time-rectification-skill`
(commit a911c890), same paths, unmodified:
vendor/stem-branch/LICENSE
vendor/stem-branch/README.md
vendor/stem-branch/package.json
vendor/stem-branch/dist/cli.cjs
vendor/stem-branch/dist/index.cjs
index.cjs is included because the published CLI does not expose ketuMode /
siderealMode flags, and those school choices must be request parameters.
index.cjs is the same 0.8.0 library entry. The published CLI does not
expose ketuMode / siderealMode flags; that file is kept so request-level
school parameters can actually change the chart.
Jyotisha does not modify the vendored files. This NOTICE is the attribution
required by Apache-2.0 §4.
@@ -10,15 +10,14 @@
| --- | --- |
| `git status -sb` | `codex/qizheng-native-chart-20260915...origin/staging` |
| `scripts/jyotish_api_server.py` 行数 | 开工 11313;上限 11363 |
| `docs/BUG_HISTORY.md` 最大号 | BUG-699。本单使用 **700703**,未被占用 |
| `docs/BUG_HISTORY.md` 最大号 | 开工时 BUG-699。本单按预分配写入 **700703**。rebase 时发现 `3e58e85f``TASK-rectification-tiebreak-card-loss-20260915.md` 把 702–704 写进了另一份任务书(尚未写入 BUG_HISTORY)。本单记录已落盘,另一单须整体后移,不得覆盖 700–703 |
| Skill | `6.9.16`**未 bump** |
| `pre_work_check.py` | python_runtime / fragment_scan / external_engine_adapters / remote_visibility 均 ok。focused_tests 两条 `test_preflight_fragment_scan.py` 红:本机没有 `.workbuddy/skills/jyotish-vedic-astrology` 镜像路径,且 fragment candidate_count 5 < residue 14。与本单改动无关,未修 |
## 任务 1 · vendored 归档与镜像 Node
- 上游 Git `https://github.com/732642856/yinduzhanxing` @ `a911c890` 本机取不到(HTTPS 认证失败、SSH 无密钥、页面 404)。见 `BLOCKED.md`
- `vendor/stem-branch/{LICENSE,README.md,package.json,dist/cli.cjs}` 取自 npm `@4n6h4x0r/stem-branch@0.8.0`Apache-2.0)。`cli.cjs` **45,963 行 / 2,682,464 B**,与任务书一致
- 另归档 `dist/index.cjs`:发布 CLI 没有 `ketuMode` / `siderealMode` 旗标,BUG-701 需要库函数入口。
- 上游不从 GitHub 拉。本机 `G:\Ferti\yinduzhanxing-codex-add-birth-time-rectification-skill` = 任务书 `/workspace/yinduzhanxing``a911c890`)。已按内容拷贝 `vendor/stem-branch/{LICENSE,README.md,package.json,dist/cli.cjs}` 到本仓同名路径。`cli.cjs` **45,963 行 / 2,682,464 B**sha256 前缀 `cf1bca10507ffcc9`,与上游文件逐字节相同
- `scripts/qizheng_chart_engine.py` 以上游 168 行适配器为起点(`calculate_qizheng_chart` + CLI `--seven-governors --json`),再按任务 2 改派别参数、boundary、坐标系和错误路径。默认走 `cli.cjs`;覆盖 `ketu_mode` / `sidereal_mode` 时用同包 `dist/index.cjs`CLI 无对应旗标)
- 仓库根 `NOTICE`:包名、0.8.0、Apache-2.0、上游地址、取入方式。
- `deploy/railway-api.Dockerfile``nodejs` 运行时(不装 npm/构建链),`COPY vendor ./vendor`
- `deploy/gated-paths.txt`:新增 `vendor/**`
+1 -1
View File
@@ -222,7 +222,7 @@
| `TASK-rectification-house-lord-gochara-research-20260913.md` | `PROGRESS-rectification-house-lord-gochara-research-20260913.md` | 研究单:宫主触发与木星/土星过运(合冲本命宫主、罗睺紧密合、年精度、用于 block 选上升)四种放宽,20 例公开 AA 离线量 block 层与 minute 层两组指标;引擎里已有宫主/功能吉凶/受控过运,只量缺的四条 | 待验收(无收益,关闭;不立实现单) | `codex/rectification-house-lord-gochara-research-20260913` |
| `TASK-qizheng-native-chart-20260915.md` | `PROGRESS-qizheng-native-chart-20260915.md` | **后端单(独占 `scripts/jyotish_api_server.py`**vendored `stem-branch` 0.8.0Apache-2.0)归档 + API 镜像 Node runtime + 七政适配器(计都派别与宿度坐标系参数化、boundary 按实测重写)+ 三个只读端点 `/api/qizheng`、`/api/western`、`/api/ephemeris_events`。实证三条:四柱时柱按 UTC 算(BUG-700,本轮不修不调用)、`ketuMode` 写死未暴露(BUG-701)、boundary 把空神煞与未闭合庙旺说成已生成(BUG-702)。BUG 段 700703 | 已实现,推送中 | `codex/qizheng-native-chart-20260915` |
| `TASK-qizheng-native-chart-20260915.md` | `PROGRESS-qizheng-native-chart-20260915.md` | **后端单(独占 `scripts/jyotish_api_server.py`**vendored `stem-branch` 0.8.0Apache-2.0)归档 + API 镜像 Node runtime + 七政适配器(计都派别与宿度坐标系参数化、boundary 按实测重写)+ 三个只读端点 `/api/qizheng`、`/api/western`、`/api/ephemeris_events`。实证三条:四柱时柱按 UTC 算(BUG-700,本轮不修不调用)、`ketuMode` 写死未暴露(BUG-701)、boundary 把空神煞与未闭合庙旺说成已生成(BUG-702)。BUG 段 700703 | 已实现,待推 staging | `codex/qizheng-native-chart-20260915` |
| `TASK-chart-page-20260915.md` | `PROGRESS-chart-page-20260915.md` | **前端单(独占 `app-sidebar.tsx`,同时加星盘与星历两个入口)**:P0 只读星盘页,五个 Tab(星盘 / 基础信息 / 大运 / 西洋盘 / 七政四余),中宫排盘参数卡,三套坐标系各自标注且禁止互相换算。不扣点不调模型不出 spinner;不碰 `page.tsx`1951/2000);**不搬 `vedic-chart-svg.tsx`**rectification-board 也在用)。BUG 段 704706 | 待领取 | — |
| `TASK-ephemeris-page-20260915.md` | `PROGRESS-ephemeris-page-20260915.md` | **前端单**:P1 星历页,今日五要素 + 当日行运(相对本命宫位)+ 未来九十天换座与停滞,底部「带这天去提问」出口。页面不得出现任何运势判断。含实证缺陷:panchanga 写死 Lahiri 与账户 Raman 分裂(BUG-707,只标注不修)。侧边栏入口由 chart-page 单交付。BUG 段 707709 | 待领取 | — |
| `TASK-api-server-decomposition-20260916.md` | `PROGRESS-api-server-decomposition-20260916.md` | **重构单(串行在 qizheng 单之后)**:把业务逻辑搬出 `JyotishAPIHandler`。核心不是行数,是全仓 3 处靠 `JyotishAPIHandler.__new__` 伪造空壳 handler 借方法(`consultation_workflow_service` ×2、`capture_report_blocked_repairs_golden`、`local_accuracy_report`,MCP 也走这条),依赖方向反了、handler 没有 `headers`/`wfile` 随时可炸。四阶段:拆 `__new__` 后门 → 抽 ≥150 行业务方法 → `do_POST`/`do_GET` 改路由表 → 重新冻结行数 baseline(余量 300→50)。纯搬运不改行为,`test_api_server_security.py` 3841 行断言一条不许改。预计 11,314 → 约 9,230 行。BUG 段 710+ | 待领取 | — |
+1 -1
View File
@@ -3500,7 +3500,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
elif path == '/api/qizheng':
qizheng = _load_local_module('qizheng_chart_engine')
try:
result = qizheng.build_qizheng_natal_chart(body)
result = qizheng.calculate_qizheng_chart(body)
except qizheng.QizhengChartError as exc:
raise BadRequest(str(exc)) from exc
self._json(result)
+197 -111
View File
@@ -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",
]
+1 -1
View File
@@ -118,5 +118,5 @@ def test_qizheng_node_unavailable_returns_structured_error(api_env, monkeypatch)
payload = handler.payload()
assert payload["success"] is False
assert payload["error_code"] == "ERR_BAD_REQUEST"
assert "node" in payload["error"]
assert "unavailable" in payload["error"].lower()
assert "Traceback" not in payload["error"]
+15 -14
View File
@@ -11,6 +11,7 @@ from types import SimpleNamespace
import pytest
from scripts import qizheng_chart_engine as engine
from scripts.qizheng_chart_engine import calculate_qizheng_chart
ROOT = Path(__file__).resolve().parents[1]
GOLDEN = ROOT / "tests" / "golden" / "qizheng_stem_branch_19900409.json"
@@ -49,7 +50,7 @@ def test_golden_fixture_is_real_engine_output() -> None:
def test_live_chart_matches_golden_and_product_contract() -> None:
_require_node()
golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
result = engine.build_qizheng_natal_chart(SAMPLE)
result = calculate_qizheng_chart(SAMPLE)
assert result["coordinate_system"] == "qizheng_mansion_degrees_from_jiao"
assert set(result["bodies"]) == set(engine.REQUIRED_BODIES)
assert len(result["palaces"]) == 12
@@ -70,7 +71,7 @@ def test_live_chart_matches_golden_and_product_contract() -> None:
def test_ketu_mode_request_overrides_and_is_echoed() -> None:
_require_node()
result = engine.build_qizheng_natal_chart({**SAMPLE, "ketu_mode": "descending-node"})
result = calculate_qizheng_chart({**SAMPLE, "ketu_mode": "descending-node"})
assert result["calculation"]["ketu_mode"] == "descending-node"
assert result["calculation"]["engine_ketu_mode"] == "descending-node"
rahu = result["bodies"]["rahu"]["siderealLon"]
@@ -81,23 +82,23 @@ def test_ketu_mode_request_overrides_and_is_echoed() -> None:
def test_missing_lat_is_structured_input_error() -> None:
with pytest.raises(engine.QizhengChartError, match="lat is required") as caught:
engine.build_qizheng_natal_chart({k: v for k, v in SAMPLE.items() if k != "lat"})
calculate_qizheng_chart({k: v for k, v in SAMPLE.items() if k != "lat"})
assert caught.value.error_code == "ERR_QIZHENG_INPUT"
def test_node_missing_error(monkeypatch) -> None:
monkeypatch.setattr(engine.shutil, "which", lambda name: None)
with pytest.raises(engine.QizhengChartError, match="node runtime is not available") as caught:
engine.build_qizheng_natal_chart(SAMPLE)
with pytest.raises(engine.QizhengChartError, match="unavailable") as caught:
calculate_qizheng_chart(SAMPLE)
assert caught.value.error_code == "ERR_QIZHENG_NODE_MISSING"
def test_cli_missing_error(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setattr(engine, "CLI_PATH", tmp_path / "cli.cjs")
monkeypatch.setattr(engine, "LIB_PATH", tmp_path / "index.cjs")
monkeypatch.setattr(engine, "VENDORED_CLI", tmp_path / "cli.cjs")
monkeypatch.setattr(engine, "VENDORED_LIB", tmp_path / "index.cjs")
monkeypatch.setattr(engine.shutil, "which", lambda name: "node")
with pytest.raises(engine.QizhengChartError, match="missing") as caught:
engine.build_qizheng_natal_chart(SAMPLE)
with pytest.raises(engine.QizhengChartError, match="unavailable") as caught:
calculate_qizheng_chart(SAMPLE)
assert caught.value.error_code == "ERR_QIZHENG_CLI_MISSING"
@@ -108,8 +109,8 @@ def test_timeout_error(monkeypatch) -> None:
raise subprocess.TimeoutExpired(cmd="node", timeout=20)
monkeypatch.setattr(engine.subprocess, "run", _boom)
with pytest.raises(engine.QizhengChartError, match="timed out") as caught:
engine.build_qizheng_natal_chart(SAMPLE)
with pytest.raises(engine.QizhengChartError, match="超时") as caught:
calculate_qizheng_chart(SAMPLE)
assert caught.value.error_code == "ERR_QIZHENG_TIMEOUT"
@@ -123,7 +124,7 @@ def test_nonzero_exit_does_not_leak_stderr(monkeypatch) -> None:
),
)
with pytest.raises(engine.QizhengChartError, match="exited with an error") as caught:
engine.build_qizheng_natal_chart(SAMPLE)
calculate_qizheng_chart(SAMPLE)
assert caught.value.error_code == "ERR_QIZHENG_ENGINE_EXIT"
assert "secret" not in str(caught.value)
assert "cli.cjs" not in str(caught.value)
@@ -136,8 +137,8 @@ def test_non_json_stdout_error(monkeypatch) -> None:
"run",
lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="not-json", stderr=""),
)
with pytest.raises(engine.QizhengChartError, match="non-JSON") as caught:
engine.build_qizheng_natal_chart(SAMPLE)
with pytest.raises(engine.QizhengChartError, match="JSON") as caught:
calculate_qizheng_chart(SAMPLE)
assert caught.value.error_code == "ERR_QIZHENG_BAD_JSON"