From 2ca09398173020aa75300c978130d919e6639e7e Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Mon, 29 Jun 2026 06:09:17 +0800 Subject: [PATCH] Align VedAstro range scan with official SearchEvents contract --- scripts/vedastro_service_adapter.py | 160 ++++++++++++++++-- .../test_vedastro_service_adapter_executor.py | 122 ++++++++++++- 2 files changed, 266 insertions(+), 16 deletions(-) diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index aa0ab296..66b6d17a 100644 --- a/scripts/vedastro_service_adapter.py +++ b/scripts/vedastro_service_adapter.py @@ -76,6 +76,14 @@ PARITY_CASES = { SUPPORTED_RANGE_SCAN_DOMAINS = {"marriage", "wealth", "career"} SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS = {"marriage", "wealth", "career", "general"} +OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH = "/Calculate/SearchEvents" +OFFICIAL_SEARCH_EVENTS_METHOD = "POST" +OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION = "official_builder_search_events_v1" +OFFICIAL_RANGE_SCAN_EVENT_TAGS = { + "marriage": ["Marriage", "General"], + "wealth": ["LendingMoney", "BorrowingMoney", "General"], + "career": ["General", "Building", "Travel"], +} VEDASTRO_CALCULATION_COVERAGE = { "official_python_library_calculations": "596+", "official_api_builder_calculators": "600+", @@ -341,6 +349,22 @@ def schema() -> dict[str, Any]: "evidence_ledger", "source_metadata", ], + "official_search_events_profile_contract": { + "profile_version": OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION, + "base_url_requirement": "VEDASTRO_API_ENDPOINT must end with /api", + "route_template": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH, + "method": OFFICIAL_SEARCH_EVENTS_METHOD, + "content_type": "application/json", + "optional_auth_header": "x-api-key", + "body_fields": [ + "BirthTime", + "Ayanamsa", + "EventTagList", + ], + "range_mode_fields": [ + "AtTime | StartTime + EndTime + PrecisionHours", + ], + }, "vedastro_calculation_coverage": VEDASTRO_CALCULATION_COVERAGE, "external_technique_request_contract": [ "operation", @@ -416,7 +440,7 @@ def _request_preview(case: dict[str, Any]) -> dict[str, Any]: def _range_scan_preview(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]: - return { + preview = { "operation": "range_scan", "vedastro_event_method": "SearchEvents", "domain": domain, @@ -425,6 +449,68 @@ def _range_scan_preview(case: dict[str, Any], domain: str, start_date: str, end_ "event_model": "vedastro_events_at_range_candidate", **case, } + preview["official_request_profile"] = _build_official_search_events_profile(preview) + return preview + + +def _format_std_time(date_text: str, hour: Any, minute: Any, tz: Any) -> str: + year, month, day = str(date_text).split("-") + hour_int = int(float(hour)) + minute_int = int(float(minute)) + return f"{hour_int:02d}:{minute_int:02d} {day}/{month}/{year} {tz}" + + +def _time_json_from_case(case: dict[str, Any], date_text: str) -> dict[str, Any]: + return { + "StdTime": _format_std_time(date_text, case.get("hour", 0), case.get("minute", 0), case.get("tz", "+00:00")), + "Location": { + "Name": case.get("case_id") or "UserLocation", + "Latitude": case.get("lat"), + "Longitude": case.get("lon"), + }, + } + + +def _normalize_tz(case: dict[str, Any]) -> str: + tz = case.get("tz") + if isinstance(tz, str): + return tz + if tz is None: + return "+00:00" + sign = "+" if float(tz) >= 0 else "-" + value = abs(float(tz)) + hours = int(value) + minutes = int(round((value - hours) * 60)) + return f"{sign}{hours:02d}:{minutes:02d}" + + +def _build_official_search_events_profile(request_preview: dict[str, Any]) -> dict[str, Any]: + case = dict(request_preview) + case["tz"] = _normalize_tz(case) + body = { + "BirthTime": _time_json_from_case(case, f"{case['year']:04d}-{case['month']:02d}-{case['day']:02d}"), + "Ayanamsa": str(case.get("ayanamsa_policy") or "lahiri"), + "EventTagList": OFFICIAL_RANGE_SCAN_EVENT_TAGS.get(str(request_preview.get("domain") or ""), ["General"]), + } + start_time = _time_json_from_case(case, str(request_preview["start_date"])) + end_time = _time_json_from_case(case, str(request_preview["end_date"])) + if str(request_preview["start_date"]) == str(request_preview["end_date"]): + body["AtTime"] = start_time + else: + body["StartTime"] = start_time + body["EndTime"] = end_time + body["PrecisionHours"] = 100 + headers: dict[str, str] = {"Content-Type": "application/json"} + api_key = os.environ.get("VEDASTRO_API_KEY", "").strip() + if api_key: + headers["x-api-key"] = api_key + return { + "profile_version": OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION, + "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH, + "method": OFFICIAL_SEARCH_EVENTS_METHOD, + "headers": headers, + "body": body, + } def _external_technique_preview( @@ -451,7 +537,8 @@ def _base_live_metadata( attempt_count: int = 1, retry_error_codes: list[int] | None = None, ) -> dict[str, Any]: - return { + official_request_profile = request_preview.get("official_request_profile") if isinstance(request_preview, dict) else None + metadata = { "transport": "http_json_service_boundary", "endpoint": endpoint, "endpoint_host": _endpoint_host(endpoint), @@ -467,6 +554,18 @@ def _base_live_metadata( "attempt_count": attempt_count, "retry_error_codes": retry_error_codes or [], } + if isinstance(official_request_profile, dict): + redacted_headers = dict(official_request_profile.get("headers") or {}) + if "x-api-key" in redacted_headers: + redacted_headers["x-api-key"] = "[redacted]" + redacted_profile = { + **official_request_profile, + "headers": redacted_headers, + } + metadata["official_endpoint_path"] = official_request_profile.get("endpoint_path") + metadata["official_request_profile"] = redacted_profile + metadata["official_request_profile_hash"] = _hash_payload(redacted_profile) + return metadata def _normalize_success( @@ -640,11 +739,18 @@ def _source_metadata(endpoint: str) -> dict[str, Any]: def _post_json(endpoint: str, request_preview: dict[str, Any]) -> dict[str, Any] | str: + official_request_profile = request_preview.get("official_request_profile") if isinstance(request_preview, dict) else None + request_url = endpoint + headers = {"Content-Type": "application/json"} vedastro_payload = request_preview + if isinstance(official_request_profile, dict): + request_url = f"{endpoint.rstrip('/')}{official_request_profile.get('endpoint_path', '')}" + headers = dict(official_request_profile.get("headers") or headers) + vedastro_payload = dict(official_request_profile.get("body") or {}) req = request.Request( - endpoint, + request_url, data=json.dumps(vedastro_payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=headers, method="POST", ) with request.urlopen(req, timeout=_timeout_seconds()) as resp: @@ -747,14 +853,7 @@ def run_case(case_id: str) -> dict[str, Any]: return _normalize_success(payload, endpoint, request_preview) -def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> dict[str, Any]: - if case_id not in PARITY_CASES: - return { - "backend": "vedastro_service_adapter_candidate", - "available": False, - "status": "unknown_case_id", - "reason": f"Unknown parity case: {case_id}", - } +def _run_range_scan_case(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]: if domain not in SUPPORTED_RANGE_SCAN_DOMAINS: return { "backend": "vedastro_service_adapter_candidate", @@ -763,7 +862,7 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> "reason": f"Unsupported range scan domain: {domain}", } - request_preview = _range_scan_preview(PARITY_CASES[case_id], domain, start_date, end_date) + request_preview = _range_scan_preview(case, domain, start_date, end_date) endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() if not endpoint: result = _unconfigured("VEDASTRO_API_ENDPOINT is not configured; range scan stops before network access.") @@ -824,6 +923,41 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> return _normalize_range_scan_success(payload, endpoint, request_preview, attempt_count, retry_error_codes) +def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> dict[str, Any]: + if case_id not in PARITY_CASES: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "unknown_case_id", + "reason": f"Unknown parity case: {case_id}", + } + return _run_range_scan_case(PARITY_CASES[case_id], domain, start_date, end_date) + + +def run_range_scan_for_case( + case: dict[str, Any], + domain: str, + start_date: str, + end_date: str, + case_id: str = "user_chart", +) -> dict[str, Any]: + user_case = { + "case_id": case_id, + "year": case.get("year"), + "month": case.get("month"), + "day": case.get("day"), + "hour": case.get("hour"), + "minute": case.get("minute"), + "second": case.get("second", 0), + "lat": case.get("lat"), + "lon": case.get("lon"), + "tz": case.get("tz"), + "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri", + "node_policy": case.get("node_policy") or case.get("node_mode") or "mean", + } + return _run_range_scan_case(user_case, domain, start_date, end_date) + + def run_external_technique(case_id: str, domain: str, method: str, api_endpoint: str) -> dict[str, Any]: if case_id not in PARITY_CASES: return { diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py index 2e90c55a..d6d551f9 100644 --- a/tests/test_vedastro_service_adapter_executor.py +++ b/tests/test_vedastro_service_adapter_executor.py @@ -189,6 +189,115 @@ def test_vedastro_range_scan_unconfigured_still_returns_official_search_events_p assert report["request_preview"]["operation"] == "range_scan" assert report["request_preview"]["vedastro_event_method"] == "SearchEvents" assert report["request_preview"]["domain"] == "wealth" + assert report["request_preview"]["official_request_profile"]["endpoint_path"] == "/Calculate/SearchEvents" + assert report["request_preview"]["official_request_profile"]["method"] == "POST" + assert report["request_preview"]["official_request_profile"]["headers"] == {"Content-Type": "application/json"} + assert report["request_preview"]["official_request_profile"]["body"]["Ayanamsa"] == "lahiri" + assert report["request_preview"]["official_request_profile"]["body"]["EventTagList"] == ["LendingMoney", "BorrowingMoney", "General"] + assert "AtTime" not in report["request_preview"]["official_request_profile"]["body"] + assert report["request_preview"]["official_request_profile"]["body"]["StartTime"]["StdTime"] == "12:00 01/01/2026 +08:00" + assert report["request_preview"]["official_request_profile"]["body"]["EndTime"]["StdTime"] == "12:00 01/01/2031 +08:00" + assert report["request_preview"]["official_request_profile"]["body"]["PrecisionHours"] == 100 + + +def test_vedastro_schema_declares_official_search_events_live_contract() -> None: + completed = subprocess.run( + [sys.executable, "scripts/vedastro_service_adapter.py", "--print-schema"], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + contract = report["official_search_events_profile_contract"] + assert contract["route_template"] == "/Calculate/SearchEvents" + assert contract["method"] == "POST" + assert contract["optional_auth_header"] == "x-api-key" + assert "BirthTime" in contract["body_fields"] + assert "EventTagList" in contract["body_fields"] + assert "AtTime | StartTime + EndTime + PrecisionHours" in contract["range_mode_fields"] + + +def test_vedastro_service_adapter_posts_official_search_events_contract() -> None: + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + assert self.path == "/api/Calculate/SearchEvents" + assert self.headers.get("Content-Type") == "application/json" + assert self.headers.get("x-api-key") == "sk_live_test" + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + assert payload["Ayanamsa"] == "lahiri" + assert payload["EventTagList"] == ["Marriage", "General"] + assert payload["BirthTime"]["StdTime"] == "12:00 01/01/1990 +08:00" + assert payload["AtTime"]["StdTime"] == "12:00 01/01/2026 +08:00" + assert "StartTime" not in payload + response = { + "Status": "Pass", + "Payload": [ + { + "Name": "JupiterSupportsMarriageAxis", + "StartTime": "2026-05-01", + "EndTime": "2026-06-01", + "EventTags": ["marriage", "transit"], + "Nature": "Good", + } + ], + } + body = json.dumps(response).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args) -> None: # noqa: A003 + return + + server = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + env = os.environ.copy() + env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/api" + env["VEDASTRO_ENABLE_NETWORK"] = "1" + env["VEDASTRO_API_KEY"] = "sk_live_test" + completed = subprocess.run( + [ + sys.executable, + "scripts/vedastro_service_adapter.py", + "--range-scan", + "--domain", + "marriage", + "--case", + "beijing_first_use_demo", + "--start-date", + "2026-01-01", + "--end-date", + "2026-01-01", + ], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + env=env, + ) + finally: + server.shutdown() + thread.join(timeout=5) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["status"] == "ok" + assert report["source_metadata"]["endpoint"].endswith("/api") + assert report["source_metadata"]["official_endpoint_path"] == "/Calculate/SearchEvents" + assert report["source_metadata"]["official_request_profile"]["method"] == "POST" + assert report["source_metadata"]["official_request_profile"]["headers"]["x-api-key"] == "[redacted]" + assert report["source_metadata"]["official_request_profile_hash"] + assert report["source_metadata"]["request_hash"] != report["source_metadata"]["official_request_profile_hash"] def test_vedastro_service_adapter_can_normalize_mock_http_response() -> None: @@ -259,9 +368,12 @@ def test_vedastro_service_adapter_can_normalize_mock_range_scan_response() -> No def do_POST(self) -> None: # noqa: N802 length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(length).decode("utf-8")) - assert payload["operation"] == "range_scan" - assert payload["domain"] == "marriage" - assert payload["start_date"] == "2026-01-01" + assert payload["Ayanamsa"] == "lahiri" + assert payload["EventTagList"] == ["Marriage", "General"] + assert payload["BirthTime"]["StdTime"] == "12:00 01/01/1990 +08:00" + assert payload["StartTime"]["StdTime"] == "12:00 01/01/2026 +08:00" + assert payload["EndTime"]["StdTime"] == "12:00 01/01/2031 +08:00" + assert payload["PrecisionHours"] == 100 response = { "events": [ { @@ -417,6 +529,10 @@ def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None: assert metadata["method"] == "POST" assert metadata["operation"] == "range_scan" assert metadata["vedastro_event_method"] == "SearchEvents" + assert metadata["official_endpoint_path"] == "/Calculate/SearchEvents" + assert metadata["official_request_profile"]["method"] == "POST" + assert metadata["official_request_profile"]["body"]["EventTagList"] == ["Marriage", "General"] + assert metadata["official_request_profile_hash"] assert metadata["allowlist_domain"] == "marriage" assert metadata["allowlist_event_count"] == 1 assert metadata["filtered_event_count"] == 1