#!/usr/bin/env python3 """VedAstro official REST bridge. The official MCP ``tools/call`` resolver returned "Invalid or Outdated Call" in the 2026-09-01 probe. This bridge calls ``https://api.vedastro.org/api/Calculate`` directly for external comparison data. Birth arguments default to a fictional smoke chart (1990-01-01 Beijing). They are never a real person. The match/synastry subcommand is not implemented and remains ``official_blocked``. """ from __future__ import annotations import argparse import json import os import sys import threading import time import urllib.error import urllib.request from pathlib import Path from typing import Any API = "https://api.vedastro.org/api/Calculate" RATE_LIMIT_PER_MINUTE = 5 # Fictional smoke only. Do not replace with a real birth record. SMOKE_DEFAULT = dict( name="Beijing, China", lat=39.9042, lon=116.4074, birth="12:00 01/01/1990 +08:00", ayanamsa="LAHIRI", ) _RATE_LOCK = threading.Lock() _RATE_WINDOW = 0.0 _RATE_COUNT = 0 class RestRateLimited(RuntimeError): """Raised when the local 5/min guard trips before a network call.""" def _time(birth: str, name: str, lat: float, lon: float) -> dict[str, Any]: return {"StdTime": birth, "Location": {"Name": name, "Longitude": lon, "Latitude": lat}} def _date_to_vedastro_std(value: str) -> str: parts = value.split() if len(parts) != 3: raise ValueError("time must use 'YYYY-MM-DD HH:MM +08:00' format") date_part, time_part, tz = parts y, m, d = date_part.split("-") return f"{time_part} {d}/{m}/{y} {tz}" def consume_rate_token(*, now: float | None = None) -> None: global _RATE_WINDOW, _RATE_COUNT current = time.time() if now is None else now with _RATE_LOCK: if current - _RATE_WINDOW >= 60: _RATE_WINDOW = current _RATE_COUNT = 0 if _RATE_COUNT >= RATE_LIMIT_PER_MINUTE: raise RestRateLimited("vedastro_rest_rate_limited") _RATE_COUNT += 1 def reset_rate_limiter() -> None: global _RATE_WINDOW, _RATE_COUNT with _RATE_LOCK: _RATE_WINDOW = 0.0 _RATE_COUNT = 0 def build_horoscope_payload(args: argparse.Namespace) -> dict[str, Any]: return { "Ayanamsa": args.ayanamsa, "BirthTime": _time(args.birth, args.name, args.lat, args.lon), "SortByWeight": False, } def build_dasa_payload(args: argparse.Namespace) -> dict[str, Any]: return { "Ayanamsa": args.ayanamsa, "BirthTime": _time(args.birth, args.name, args.lat, args.lon), "StartTime": _time(_date_to_vedastro_std(args.start), args.name, args.lat, args.lon), "EndTime": _time(_date_to_vedastro_std(args.end), args.name, args.lat, args.lon), "Levels": 3, "PrecisionHours": 100, } def match_not_implemented() -> dict[str, Any]: return { "Status": "official_blocked", "official_closure_state": "official_blocked", "official_closure_reason": "match_subcommand_not_implemented", "boundary": "VedAstro REST match/synastry is not implemented; do not fabricate a comparison.", } def call( method: str, payload: dict[str, Any], out: str | None = None, *, opener: Any | None = None, timeout: float = 90, skip_rate_limit: bool = False, ) -> dict[str, Any]: if not skip_rate_limit: consume_rate_token() req = urllib.request.Request( f"{API}/{method}", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, method="POST", ) try: if opener is None: with urllib.request.urlopen(req, timeout=timeout) as response: data = json.loads(response.read().decode()) else: with opener(req, timeout=timeout) as response: data = json.loads(response.read().decode()) except urllib.error.HTTPError as error: if error.code == 429: return { "Status": "official_blocked", "official_closure_state": "official_blocked", "official_closure_reason": "rate_limited", "http_status": 429, } raise if out: Path(out).write_text(json.dumps(data, ensure_ascii=False, indent=1) + "\n", encoding="utf-8") print(f"[已落盘] {out}") return data def probe_calculate_health(*, opener: Any | None = None, timeout: float = 8) -> dict[str, Any]: """POST a fictional smoke horoscope and report whether Status is Pass.""" args = argparse.Namespace(**SMOKE_DEFAULT) try: payload = build_horoscope_payload(args) data = call("HoroscopePredictions", payload, opener=opener, timeout=timeout) except RestRateLimited: return { "status": "official_blocked", "reason": "rate_limited", "endpoint": API, "transport": "rest", } except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as error: return { "status": "official_blocked", "reason": type(error).__name__, "endpoint": API, "transport": "rest", } status = str(data.get("Status") or "") if status.lower() == "pass": return {"status": "official_verified", "reason": "status_pass", "endpoint": API, "transport": "rest"} if str(data.get("official_closure_reason") or "") == "rate_limited": return {"status": "official_blocked", "reason": "rate_limited", "endpoint": API, "transport": "rest"} return { "status": "official_blocked", "reason": f"status_{status or 'missing'}", "endpoint": API, "transport": "rest", } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="VedAstro official REST bridge (fictional smoke defaults)") sub = parser.add_subparsers(dest="cmd", required=True) for command in ("horoscope", "dasa"): item = sub.add_parser(command) item.add_argument("--birth", default=SMOKE_DEFAULT["birth"], help="VedAstro StdTime; default is fictional smoke") item.add_argument("--name", default=SMOKE_DEFAULT["name"]) item.add_argument("--lat", type=float, default=SMOKE_DEFAULT["lat"]) item.add_argument("--lon", type=float, default=SMOKE_DEFAULT["lon"]) item.add_argument("--ayanamsa", default=SMOKE_DEFAULT["ayanamsa"]) item.add_argument("--out") sub._name_parser_map["dasa"].add_argument("--start", default="2026-01-01 00:00 +08:00") sub._name_parser_map["dasa"].add_argument("--end", default="2032-12-31 00:00 +08:00") match = sub.add_parser("match", help="Not implemented; remains official_blocked") match.add_argument("--out") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.cmd == "match": result = match_not_implemented() if args.out: Path(args.out).write_text(json.dumps(result, ensure_ascii=False, indent=1) + "\n", encoding="utf-8") print(json.dumps(result, ensure_ascii=False)) return 0 if args.cmd == "horoscope": result = call("HoroscopePredictions", build_horoscope_payload(args), args.out) elif args.cmd == "dasa": result = call("DasaAtRange", build_dasa_payload(args), args.out) else: sys.exit("未知子命令") if not args.out: print(json.dumps(result, ensure_ascii=False)[:2000]) return 0 if __name__ == "__main__": os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1") raise SystemExit(main())