root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

Add VedAstro user entrypoint

Browse Source
This commit is contained in:
732642856
2026-07-02 23:17:47 +08:00
parent 21212a1b9f
commit ec4598c0d7
3 changed files with 513 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+56
View File
@@ -77,6 +77,62 @@ VEDASTRO_ENABLE_NETWORK=1
默认运行是快速模式：VedAstro official 证据层如果没有在前台预算内闭环，会诚实标记 `official_snapshot_budget_exhausted` 并退回本地 Swiss Ephemeris。要跑 official extended 模式，复制 `.env.official.example` 为 `.env.local` 并填好 endpoint/network/key；运行 `python3 scripts/diagnose_vedastro_mode.py` 可先确认当前是 `fast_local_fallback` 还是 `official_extended`。
### Codex 用户级 VedAstro + strict workflow 入口
如果用户在 Codex 窗口从云端 Git 仓库拉取本项目，推荐先走这一条稳定入口，而不是手动拼多个底层脚本：
```bash
python3 scripts/vedastro_user_entrypoint.py \
--year REDACTED_YEAR --month 4 --day 17 --hour 14 --minute 49 \
--lat 36.42 --lon 114.2 --tz 8 \
--question "事业机会什么时候出现" \
--themes career,marriage,wealth \
--reference-date 2026-07-02 \
--format markdown
```
机器读取或交给后续 agent 处理时使用 JSON：
```bash
python3 scripts/vedastro_user_entrypoint.py \
--year REDACTED_YEAR --month 4 --day 17 --hour 14 --minute 49 \
--lat 36.42 --lon 114.2 --tz 8 \
--question "事业机会什么时候出现" \
--themes career,health,education,property,children,migration,prashna \
--reference-date 2026-07-02 \
--format json
```
这个入口会自动做四件事：
1. 读取 `.env.local` 并诊断当前是 `official_extended` 还是 `fast_local_fallback`。
2. 启动 `official_full_capability_catalog`，给 VedAstro 官方能力目录生成 `domain / execution_policy / adjudicator_use / confidence_role / blocked_reason`。
3. 按 `--themes` 做动态选择，避免把健康、教育、房产、子女、迁移、Prashna 等非三大主题塞进 `general`。
4. 触发 strict workflow 合同摘要，输出 primary route、可用 route、cache/TTL/free-tier queue 策略和 honesty boundary。
边界必须保留：这个入口**不会把 641 项全部当作已执行**。它先做官方能力目录分类和主题选择；能自动执行的进入证据层，需要第二人资料、用户文本、校时画像或官方网络预算的方法会保持 `needs_user_context`、`needs_user_text`、`needs_rectification_profile` 或 `blocked`。
推荐的 official extended `.env.local` 示例：
```bash
VEDASTRO_API_ENDPOINT=https://api.vedastro.org/api
VEDASTRO_ENABLE_NETWORK=1
VEDASTRO_TIMEOUT_SECONDS=20
VEDASTRO_CACHE_TTL_SECONDS=600
VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=600
VEDASTRO_FREE_TIER_QUEUE=1
# 可选
# VEDASTRO_API_KEY=sk_live_xxx
```
先运行：
```bash
python3 scripts/diagnose_vedastro_mode.py
```
若仍显示 `fast_local_fallback`，用户级入口仍可运行，但解盘必须把 VedAstro official 证据写成 blocked/降级，不能声称 official extended 已闭环。
### 普通用户交付形态
| 形态 | 入口 | 命令 | 能力边界 |
scripts/vedastro_user_entrypoint.py
+279
View File
@@ -0,0 +1,279 @@
#!/usr/bin/env python3
"""User-level VedAstro + strict-workflow entrypoint for Codex sessions."""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
try:
from scripts.diagnose_vedastro_mode import build_report as build_runtime_mode_report
from scripts.local_env import load_local_env
except ModuleNotFoundError: # pragma: no cover - direct script execution
from diagnose_vedastro_mode import build_report as build_runtime_mode_report
from local_env import load_local_env
ROOT = Path(__file__).resolve().parents[1]
RUNNER = ROOT / "scripts" / "vedastro_official_capability_runner.py"
def _bool_env(name: str) -> bool:
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
def _int_env(name: str, default: int = 0) -> int:
raw = os.environ.get(name, "").strip()
if not raw:
return default
try:
return int(float(raw))
except ValueError:
return default
def _themes(raw: str) -> list[str]:
aliases = {
"relationship": "marriage",
"relationships": "marriage",
"finance": "wealth",
"money": "wealth",
"事业": "career",
"婚恋": "marriage",
"婚姻": "marriage",
"财富": "wealth",
"健康": "health",
"教育": "education",
"房产": "property",
"子女": "children",
"迁移": "migration",
"问卜": "prashna",
"校时": "rectification",
"应期": "timing",
}
values: list[str] = []
for item in raw.replace("，", ",").split(","):
key = aliases.get(item.strip().lower(), item.strip().lower())
if key and key not in values:
values.append(key)
return values or ["career", "marriage", "wealth"]
def _primary_route(question: str, themes: list[str]) -> str:
text = f"{question} {' '.join(themes)}".lower()
route_aliases = [
("career", ("career", "job", "work", "profession", "事业", "工作", "职业", "项目")),
("relationship", ("marriage", "relationship", "spouse", "partner", "婚恋", "婚姻", "伴侣")),
("finance", ("wealth", "finance", "money", "income", "财富", "金钱", "收入")),
]
for route, tokens in route_aliases:
if any(token in text for token in tokens):
return route
return "general"
def _case_from_args(args: argparse.Namespace, themes: list[str]) -> dict[str, Any]:
return {
"year": args.year,
"month": args.month,
"day": args.day,
"hour": args.hour,
"minute": args.minute,
"second": args.second,
"lat": args.lat,
"lon": args.lon,
"tz": args.tz,
"reference_date": args.reference_date,
"themes": themes,
"ayanamsa_policy": args.ayanamsa,
"node_policy": args.node_mode,
}
def _run_capability_catalog(case: dict[str, Any]) -> dict[str, Any]:
completed = subprocess.run(
[
sys.executable,
str(RUNNER),
"--bundle",
"official_full_capability_catalog",
"--birth-json",
json.dumps(case, ensure_ascii=False),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=max(5.0, float(os.environ.get("VEDASTRO_TIMEOUT_SECONDS", "5") or 5)),
check=False,
env=os.environ.copy(),
)
if completed.returncode != 0:
return {
"available": False,
"status": "official_full_capability_catalog_runtime_error",
"summary": {},
"domain_routing": {},
"dynamic_selection": {},
"stderr": (completed.stderr or "").strip(),
}
try:
return json.loads(completed.stdout)
except json.JSONDecodeError:
return {
"available": False,
"status": "official_full_capability_catalog_invalid_json",
"summary": {},
"domain_routing": {},
"dynamic_selection": {},
"stdout_excerpt": (completed.stdout or "").strip()[:500],
}
def _strict_workflow_summary(route: str, catalog: dict[str, Any]) -> dict[str, Any]:
dynamic_selection = catalog.get("dynamic_selection") if isinstance(catalog.get("dynamic_selection"), dict) else {}
theme_for_route = {"relationship": "marriage", "finance": "wealth"}.get(route, route)
return {
"triggered": route in {"career", "relationship", "finance"},
"primary_route": route,
"routes_available": ["career", "relationship", "finance"],
"source": "strict_workflow_contract_summary",
"official_capability_selection": dynamic_selection.get(theme_for_route) or {},
"boundary": (
"This entrypoint triggers the strict workflow contract lane and passes VedAstro official catalog "
"selection metadata; it does not claim every official method was executed."
),
}
def _cache_and_queue_report() -> dict[str, Any]:
return {
"official_full_snapshot_cache_scope": "official_full_snapshot_semantic_cache",
"official_full_snapshot_cache_ttl_seconds": _int_env("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", 0),
"range_scan_cache_scope": "vedastro_range_scan_request_cache",
"range_scan_cache_ttl_seconds": _int_env("VEDASTRO_CACHE_TTL_SECONDS", 0),
"free_tier_queue_enabled": _bool_env("VEDASTRO_FREE_TIER_QUEUE")
or _bool_env("VEDASTRO_FREE_TIER_QUEUE_ENABLED")
or _bool_env("VEDASTRO_ENABLE_FREE_TIER_QUEUE"),
"sample_limit": _int_env("VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT", 0),
"artifact_root": "scratch/local/vedastro_adapter",
}
def build_report(args: argparse.Namespace) -> dict[str, Any]:
load_local_env(ROOT)
themes = _themes(args.themes)
case = _case_from_args(args, themes)
runtime = build_runtime_mode_report()
catalog = _run_capability_catalog(case)
route = _primary_route(args.question, themes)
return {
"scope": "vedastro_user_entrypoint",
"schema_version": 1,
"input": {
"question": args.question,
"themes": themes,
"birth": {key: case[key] for key in ("year", "month", "day", "hour", "minute", "second", "lat", "lon", "tz")},
"reference_date": args.reference_date,
},
"runtime_mode": runtime,
"official_capability_catalog": {
"status": catalog.get("status") or "blocked",
"available": bool(catalog.get("available")),
"summary": catalog.get("summary") or {},
"coverage": catalog.get("coverage") or {},
"domain_routing": catalog.get("domain_routing") or {},
"dynamic_selection": catalog.get("dynamic_selection") or {},
},
"cache_and_queue": _cache_and_queue_report(),
"strict_workflow": _strict_workflow_summary(route, catalog),
"honesty_boundary": {
"all_641_methods_executed": False,
"official_catalog_classified": bool(catalog.get("summary")),
"official_extended_ready": bool(runtime.get("official_ready")),
"rule": (
"Use official catalog domain routing and dynamic selection as evidence metadata. "
"Methods requiring user context/text/rectification remain blocked or secondary until inputs exist."
),
},
"user_commands": {
"json": (
"python3 scripts/vedastro_user_entrypoint.py --year YYYY --month MM --day DD "
"--hour HH --minute MM --lat LAT --lon LON --tz TZ --question '...' "
"--themes career,marriage,wealth --reference-date YYYY-MM-DD --format json "
"# includes official_full_capability_catalog"
),
"markdown": (
"python3 scripts/vedastro_user_entrypoint.py --year YYYY --month MM --day DD "
"--hour HH --minute MM --lat LAT --lon LON --tz TZ --question '...' --format markdown"
),
},
}
def render_markdown(report: dict[str, Any]) -> str:
runtime = report["runtime_mode"]
catalog = report["official_capability_catalog"]
cache = report["cache_and_queue"]
strict = report["strict_workflow"]
summary = catalog.get("summary") or {}
lines = [
"# VedAstro 用户级入口",
"",
f"- runtime_mode: `{runtime.get('mode')}`",
f"- official_ready: `{str(runtime.get('official_ready')).lower()}`",
f"- catalog_status: `{catalog.get('status')}`",
f"- catalog_method_count: `{summary.get('catalog_method_count', 0)}`",
f"- unknown_method_count: `{summary.get('unknown_method_count', 0)}`",
f"- misrouted_general_method_count: `{summary.get('misrouted_general_method_count', 0)}`",
f"- strict workflow triggered: `{str(strict.get('triggered')).lower()}`",
f"- primary_route: `{strict.get('primary_route')}`",
"",
"## Cache / Free-Tier Queue",
"",
f"- official_full_snapshot_cache_ttl_seconds: `{cache['official_full_snapshot_cache_ttl_seconds']}`",
f"- range_scan_cache_ttl_seconds: `{cache['range_scan_cache_ttl_seconds']}`",
f"- free_tier_queue_enabled: `{str(cache['free_tier_queue_enabled']).lower()}`",
"",
"## Boundary",
"",
"- 这个入口会启动 VedAstro official capability catalog 分类、动态主题选择和 strict workflow 合同摘要。",
"- 它不会把 641 项全部当作已执行；需要用户上下文、文本问题或校时画像的方法会保持 blocked/secondary。",
"- 如果 runtime 是 `fast_local_fallback`，解盘必须诚实降级，不得声称 official extended 已闭环。",
]
return "\n".join(lines) + "\n"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--year", type=int, required=True)
parser.add_argument("--month", type=int, required=True)
parser.add_argument("--day", type=int, required=True)
parser.add_argument("--hour", type=int, required=True)
parser.add_argument("--minute", type=int, required=True)
parser.add_argument("--second", type=int, default=0)
parser.add_argument("--lat", type=float, required=True)
parser.add_argument("--lon", type=float, required=True)
parser.add_argument("--tz", type=float, required=True)
parser.add_argument("--question", default="")
parser.add_argument("--themes", default="career,marriage,wealth")
parser.add_argument("--reference-date", required=True)
parser.add_argument("--ayanamsa", default="lahiri")
parser.add_argument("--node-mode", default="mean")
parser.add_argument("--format", choices=["json", "markdown"], default="markdown")
args = parser.parse_args()
report = build_report(args)
if args.format == "json":
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
else:
print(render_markdown(report), end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/test_vedastro_user_entrypoint.py
+178
View File
@@ -0,0 +1,178 @@
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_user_entrypoint_runs_catalog_and_strict_workflow_contract() -> None:
catalog_stub = {
"available": True,
"status": "ok",
"capabilities": [
{
"method": "SearchEvents",
"signature": "(birthTime, atTime, eventTagList)",
"bucket": "event_time",
"parameter_names": ["birthTime", "atTime", "eventTagList"],
"callable": True,
},
{
"method": "DasaAtRange",
"signature": "(birthTime, startTime, endTime, levels, precisionHours)",
"bucket": "dasha_at_range",
"parameter_names": ["birthTime", "startTime", "endTime", "levels", "precisionHours"],
"callable": True,
},
{
"method": "AllPlanetDashamamshaSign",
"signature": "(planetName, time)",
"bucket": "planet_time",
"parameter_names": ["planetName", "time"],
"callable": True,
},
{
"method": "HealthProblemEvent",
"signature": "(birthTime, startTime, endTime)",
"bucket": "event_range",
"parameter_names": ["birthTime", "startTime", "endTime"],
"callable": True,
},
],
"buckets": {
"event_time": {"count": 1, "examples": ["SearchEvents"]},
"dasha_at_range": {"count": 1, "examples": ["DasaAtRange"]},
"planet_time": {"count": 1, "examples": ["AllPlanetDashamamshaSign"]},
"event_range": {"count": 1, "examples": ["HealthProblemEvent"]},
},
}
runner_stub = {
"SearchEvents": {"available": True, "status": "ok", "result": {"Events": []}},
"DasaAtRange": {"available": True, "status": "ok", "result": {"Periods": []}},
"AllPlanetDashamamshaSign": {"available": True, "status": "ok", "result": {"Name": "Capricorn"}},
"HealthProblemEvent": {"available": True, "status": "ok", "result": {}},
}
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_user_entrypoint.py",
"--year",
"REDACTED_YEAR",
"--month",
"4",
"--day",
"17",
"--hour",
"14",
"--minute",
"49",
"--lat",
"36.42",
"--lon",
"114.2",
"--tz",
"8",
"--question",
"事业机会什么时候出现",
"--themes",
"career,health",
"--reference-date",
"2026-07-02",
"--format",
"json",
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**os.environ,
"JYOTISH_SKIP_LOCAL_ENV": "1",
"VEDASTRO_API_ENDPOINT": "https://api.vedastro.org/api",
"VEDASTRO_ENABLE_NETWORK": "1",
"VEDASTRO_TIMEOUT_SECONDS": "20",
"VEDASTRO_CACHE_TTL_SECONDS": "600",
"VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS": "600",
"VEDASTRO_FREE_TIER_QUEUE": "1",
"VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT": "8",
"VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(catalog_stub),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(runner_stub),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["scope"] == "vedastro_user_entrypoint"
assert report["runtime_mode"]["mode"] == "official_extended"
assert report["input"]["themes"] == ["career", "health"]
assert report["official_capability_catalog"]["summary"]["catalog_method_count"] == 4
assert report["official_capability_catalog"]["summary"]["sample_limit"] == 8
assert report["official_capability_catalog"]["dynamic_selection"]["career"]["selected_methods"]
assert "health" in report["official_capability_catalog"]["domain_routing"]
assert report["cache_and_queue"]["official_full_snapshot_cache_ttl_seconds"] == 600
assert report["cache_and_queue"]["range_scan_cache_ttl_seconds"] == 600
assert report["cache_and_queue"]["free_tier_queue_enabled"] is True
assert report["strict_workflow"]["triggered"] is True
assert report["strict_workflow"]["primary_route"] == "career"
assert "career" in report["strict_workflow"]["routes_available"]
assert report["honesty_boundary"]["all_641_methods_executed"] is False
assert "official_full_capability_catalog" in report["user_commands"]["json"]
def test_user_entrypoint_markdown_documents_boundaries() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_user_entrypoint.py",
"--year",
"REDACTED_YEAR",
"--month",
"4",
"--day",
"17",
"--hour",
"14",
"--minute",
"49",
"--lat",
"36.42",
"--lon",
"114.2",
"--tz",
"8",
"--question",
"婚恋",
"--themes",
"marriage",
"--reference-date",
"2026-07-02",
"--format",
"markdown",
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**os.environ,
"JYOTISH_SKIP_LOCAL_ENV": "1",
"VEDASTRO_API_ENDPOINT": "",
"VEDASTRO_ENABLE_NETWORK": "",
"VEDASTRO_TIMEOUT_SECONDS": "",
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
assert "VedAstro 用户级入口" in completed.stdout
assert "fast_local_fallback" in completed.stdout
assert "不会把 641 项全部当作已执行" in completed.stdout
assert "strict workflow" in completed.stdout