From 92358f2645ccf456f451769c44b600d843081722 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Wed, 8 Jul 2026 19:08:48 +0800 Subject: [PATCH] feat: add VedAstro gateway polling queue --- scripts/jyotish_api_server.py | 35 ++++++++++++ scripts/vedastro_gateway.py | 90 +++++++++++++++++++++++++++++++ tests/test_api_server_security.py | 46 ++++++++++++++++ tests/test_vedastro_gateway.py | 40 ++++++++++++++ 4 files changed, 211 insertions(+) diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index d80a8e62..0751b077 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -877,6 +877,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): self._json(self._vedastro_status()) elif path == '/api/vedastro_gateway/status': self._json(self._compute_vedastro_gateway_status()) + elif path.startswith('/api/vedastro_gateway/jobs/'): + job_id = path.rsplit('/', 1)[-1] + result = self._compute_vedastro_gateway_job(job_id) + if result is None: + self._error_json('Not found', 404, 'ERR_NOT_FOUND') + else: + self._json(result) elif path.startswith('/api/chart/jobs/'): job_id = path.rsplit('/', 1)[-1] result = self._get_chart_job(job_id) @@ -943,6 +950,9 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): elif path == '/api/vedastro_gateway/run': result = self._compute_vedastro_gateway_run(body) self._json(result) + elif path == '/api/vedastro_gateway/enqueue': + result = self._compute_vedastro_gateway_enqueue(body) + self._json(result) elif path == '/api/professional_reading': result = self._compute_professional_reading(body) self._json(result) @@ -1749,6 +1759,31 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): return gateway_status() + def _compute_vedastro_gateway_job(self, job_id): + from scripts.vedastro_gateway import get_gateway_job + + return get_gateway_job(str(job_id)) + + def _compute_vedastro_gateway_enqueue(self, body): + from scripts.vedastro_gateway import enqueue_gateway_job + + payload = dict(body or {}) + birth_payload = self._high_rigor_birth_payload(payload) + themes = self._high_rigor_requested_themes(payload) + reference_date = ( + payload.get('reference_date') + or payload.get('transit_date') + or payload.get('today') + or payload.get('current_date') + or datetime.now().strftime('%Y-%m-%d') + ) + return enqueue_gateway_job( + birth_payload, + question=str(payload.get('question') or payload.get('query') or ''), + themes=themes, + reference_date=str(reference_date), + ) + def _compute_vedastro_gateway_run(self, body): from scripts.vedastro_gateway import run_gateway_packet diff --git a/scripts/vedastro_gateway.py b/scripts/vedastro_gateway.py index 9167ebb6..2b09abc9 100644 --- a/scripts/vedastro_gateway.py +++ b/scripts/vedastro_gateway.py @@ -3,13 +3,18 @@ from __future__ import annotations +import hashlib +import json import os +from datetime import datetime, timezone +from pathlib import Path from types import SimpleNamespace from typing import Any BACKEND_PRIORITY = ["self_host", "official", "cache", "queue", "local_fallback"] BOUNDARY_TEXT = "Users never call VedAstro directly; backend gateway owns cache, queue, and fallback." +ROOT = Path(__file__).resolve().parents[1] def _bool_env(name: str) -> bool: @@ -53,6 +58,91 @@ def _active_backend(config: dict[str, Any]) -> str: return "local_fallback" +def _queue_dir() -> Path: + raw = os.environ.get("VEDASTRO_GATEWAY_QUEUE_DIR", "").strip() + return Path(raw).expanduser() if raw else ROOT / "scratch" / "local" / "vedastro_gateway_jobs" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _job_path(job_id: str) -> Path: + allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" + if not job_id or any(ch not in allowed for ch in job_id): + raise ValueError("invalid VedAstro gateway job id") + return _queue_dir() / f"{job_id}.json" + + +def _write_job(job: dict[str, Any]) -> dict[str, Any]: + path = _job_path(str(job["job_id"])) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(job, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8") + return job + + +def enqueue_gateway_job( + case: dict[str, Any], + question: str = "", + themes: list[str] | tuple[str, ...] | None = None, + reference_date: str = "", +) -> dict[str, Any]: + created_at = _now_iso() + request = { + "case": dict(case or {}), + "question": question or "", + "themes": list(themes or []), + "reference_date": reference_date or "", + } + digest = hashlib.sha256( + json.dumps({"created_at": created_at, "request": request}, ensure_ascii=False, sort_keys=True).encode("utf-8") + ).hexdigest()[:20] + job_id = f"vgw_{digest}" + return _write_job( + { + "scope": "vedastro_gateway_job", + "schema_version": 1, + "job_id": job_id, + "status": "queued", + "created_at": created_at, + "updated_at": created_at, + "poll_path": f"/api/vedastro_gateway/jobs/{job_id}", + "request": request, + "result": None, + "raw_response_archive": { + "status": "pending", + "official_raw_response_available": False, + "boundary": "Queued jobs do not prove VedAstro official raw response availability.", + }, + } + ) + + +def get_gateway_job(job_id: str) -> dict[str, Any] | None: + try: + path = _job_path(job_id) + except ValueError: + return None + if not path.exists(): + return None + return json.loads(path.read_text(encoding="utf-8")) + + +def complete_gateway_job(job_id: str, result: dict[str, Any]) -> dict[str, Any]: + job = get_gateway_job(job_id) + if job is None: + raise FileNotFoundError(job_id) + job["status"] = "completed" + job["updated_at"] = _now_iso() + job["result"] = dict(result or {}) + job["raw_response_archive"] = { + "status": "stored_gateway_packet_not_official_raw", + "official_raw_response_available": False, + "boundary": "Gateway packet was archived; VedAstro official raw response is still separate evidence.", + } + return _write_job(job) + + def gateway_status() -> dict[str, Any]: config = build_gateway_config() return { diff --git a/tests/test_api_server_security.py b/tests/test_api_server_security.py index c4748118..9dd96a55 100644 --- a/tests/test_api_server_security.py +++ b/tests/test_api_server_security.py @@ -301,6 +301,52 @@ def test_vedastro_gateway_run_route_returns_gateway_packet(monkeypatch) -> None: assert result['user_visibility']['mainland_cn_safe'] is True +def test_vedastro_gateway_enqueue_and_poll_routes(monkeypatch, tmp_path) -> None: + monkeypatch.setenv('VEDASTRO_GATEWAY_QUEUE_DIR', str(tmp_path)) + handler = _PostCaptureHandler('/api/vedastro_gateway/enqueue', { + 'year': 1955, + 'month': 2, + 'day': 24, + 'hour': 19, + 'minute': 15, + 'second': 0, + 'lat': 37.7749, + 'lon': -122.4194, + 'tz': 8, + 'question': '事业机会什么时候出现', + 'themes': ['career'], + 'reference_date': '2026-07-02', + }) + + handler.do_POST() + + assert handler.status_code == 200 + queued = handler.payload() + assert queued['status'] == 'queued' + assert queued['poll_path'].startswith('/api/vedastro_gateway/jobs/') + + poller = _ResponseCaptureHandler() + poller.path = queued['poll_path'] + poller.do_GET() + + assert poller.status_code == 200 + payload = poller.payload() + assert payload['job_id'] == queued['job_id'] + assert payload['status'] == 'queued' + assert payload['raw_response_archive']['official_raw_response_available'] is False + + +def test_vedastro_gateway_poll_rejects_invalid_job_id(monkeypatch, tmp_path) -> None: + monkeypatch.setenv('VEDASTRO_GATEWAY_QUEUE_DIR', str(tmp_path)) + poller = _ResponseCaptureHandler() + poller.path = '/api/vedastro_gateway/jobs/../../secret' + + poller.do_GET() + + assert poller.status_code == 404 + assert poller.payload()['error_code'] == 'ERR_NOT_FOUND' + + def test_professional_reading_composes_high_rigor_and_gateway(monkeypatch) -> None: handler = _handler() diff --git a/tests/test_vedastro_gateway.py b/tests/test_vedastro_gateway.py index ab78aa3e..6128ae70 100644 --- a/tests/test_vedastro_gateway.py +++ b/tests/test_vedastro_gateway.py @@ -71,3 +71,43 @@ def test_gateway_run_packet_uses_user_entrypoint_and_marks_not_all_641(monkeypat assert packet["official_capability_catalog"]["summary"]["catalog_method_count"] >= 0 assert packet["honesty_boundary"]["all_641_methods_executed"] is False assert packet["user_visibility"]["mainland_cn_safe"] is True + + +def test_gateway_queue_lifecycle_uses_file_job_store(monkeypatch, tmp_path): + from scripts import vedastro_gateway + + monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path)) + job = vedastro_gateway.enqueue_gateway_job( + { + "year": 1955, + "month": 2, + "day": 24, + "hour": 19, + "minute": 15, + "second": 0, + "lat": 37.7749, + "lon": -122.4194, + "tz": 8, + }, + question="事业机会什么时候出现", + themes=["career"], + reference_date="2026-07-02", + ) + + assert job["status"] == "queued" + assert job["poll_path"].startswith("/api/vedastro_gateway/jobs/") + + stored = vedastro_gateway.get_gateway_job(job["job_id"]) + assert stored["status"] == "queued" + assert stored["raw_response_archive"]["status"] == "pending" + assert stored["raw_response_archive"]["official_raw_response_available"] is False + + completed = vedastro_gateway.complete_gateway_job( + job["job_id"], + {"scope": "vedastro_gateway_run", "status": "local_fallback"}, + ) + assert completed["status"] == "completed" + + polled = vedastro_gateway.get_gateway_job(job["job_id"]) + assert polled["result"]["status"] == "local_fallback" + assert polled["raw_response_archive"]["status"] == "stored_gateway_packet_not_official_raw"