from __future__ import annotations import json import socket import threading import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from datetime import datetime from datetime import date, timedelta from pathlib import Path from typing import Any, Iterable from .cache import ContentCache, canonical_bytes, request_fingerprint, sha256_bytes from .telemetry import aware_now, iso class NetworkBudgetExceeded(TimeoutError): pass class SourceContractError(RuntimeError): pass QueryValue = str | int | float | bool | list[str] | tuple[str, ...] @dataclass(frozen=True) class HttpRequest: provider_id: str adapter_version: str data_kind: str method: str url: str ticker: str as_of: str fixture_id: str query: dict[str, QueryValue] | list[tuple[str, str]] = field(default_factory=dict) body: bytes | None = None content_type: str = "application/json" ttl_seconds: int = 21600 headers: dict[str, str] = field(default_factory=dict) class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def] raise SourceContractError(f"未登记重定向：{newurl}") def _query_items(query: dict[str, QueryValue] | list[tuple[str, str]]) -> list[tuple[str, str]]: if isinstance(query, list): return [(str(key), str(value)) for key, value in query] items: list[tuple[str, str]] = [] for key in sorted(query): value = query[key] if isinstance(value, (list, tuple)): items.extend((str(key), str(item)) for item in value) else: items.append((str(key), str(value))) return items def _canonical_body(body: bytes | None, content_type: str) -> bytes: if not body: return b"" base = content_type.split(";", 1)[0].lower() if base == "application/json": try: return canonical_bytes(json.loads(body.decode("utf-8"))) except (UnicodeDecodeError, json.JSONDecodeError): return body if base == "application/x-www-form-urlencoded": pairs = urllib.parse.parse_qsl(body.decode("utf-8"), keep_blank_values=True) return urllib.parse.urlencode(sorted(pairs), doseq=True).encode("utf-8") return body class HttpClient: def __init__( self, registry: dict[str, Any], cache: ContentCache, deadline_mono: float, run_id: str, process_start: datetime, fixture_dir: Path | None = None, baseline_states: dict[str, str] | None = None, baseline_entries: dict[str, dict[str, Any]] | None = None, ): self.registry = registry self.cache = cache self.deadline_mono = deadline_mono self.run_id = run_id self.process_start = process_start self.fixture_dir = fixture_dir self.baseline_states = baseline_states self.baseline_entries = baseline_entries or {} self.opener = urllib.request.build_opener(NoRedirect) self.fixture_manifest: dict[str, Any] | None = None self.fixture_identity: str | None = None self.cancel_event = threading.Event() self.network_attempts = 0 self.cache_mutations = 0 self.telemetry: list[dict[str, Any]] = [] self._telemetry_lock = threading.Lock() self._negative: dict[str, tuple[float, str]] = {} if fixture_dir: manifest_path = fixture_dir / "fixture_manifest.json" raw = manifest_path.read_bytes() self.fixture_manifest = json.loads(raw.decode("utf-8")) self.fixture_identity = sha256_bytes(raw) def remaining(self) -> float: if self.cancel_event.is_set(): return 0.0 return max(0.0, self.deadline_mono - time.monotonic()) def cancel(self) -> None: self.cancel_event.set() def _contract(self, request: HttpRequest) -> dict[str, Any]: provider = self.registry.get("providers", {}).get(request.provider_id) if not provider or provider.get("adapter_version") != request.adapter_version: raise SourceContractError(f"未登记 provider：{request.provider_id}") contract = provider.get("contracts", {}).get(request.data_kind) if not contract: raise SourceContractError(f"未登记 data_kind：{request.provider_id}/{request.data_kind}") return {**provider, **contract} def _validate(self, request: HttpRequest) -> dict[str, Any]: contract = self._contract(request) parsed = urllib.parse.urlsplit(request.url) if parsed.query or parsed.fragment or parsed.username or parsed.password: raise SourceContractError(f"URL 必须只含登记 scheme/host/path：{request.url}") if parsed.scheme != "https" or parsed.hostname != contract["host"]: raise SourceContractError(f"未登记域名：{request.url}") if parsed.path != contract["path"] or request.method != contract["method"]: raise SourceContractError(f"未登记 endpoint/method：{request.method} {request.url}") items = _query_items(request.query) keys = [key for key, _ in items] allowed = contract.get("query_keys", []) if sorted(keys) != sorted(allowed): raise SourceContractError(f"query 键不匹配：{keys}") values = {key: value for key, value in items} for key, expected in contract.get("fixed_query", {}).items(): if values.get(key) != str(expected): raise SourceContractError(f"query 固定值不匹配：{key}") code, market = request.ticker.split(".") secid = f"{1 if market == 'SH' else 0}.{code}" if "secid" in values and values["secid"] != secid: raise SourceContractError("secid 与 ticker 不匹配") if request.data_kind == "market_close": expected_beg = (date.fromisoformat(request.as_of) - timedelta(days=14)).strftime("%Y%m%d") if values.get("beg") != expected_beg or values.get("end") != request.as_of.replace("-", ""): raise SourceContractError("K 线窗口不匹配") if request.data_kind.startswith("finance_") and values.get("filter") != f'(SECUCODE="{request.ticker}")': raise SourceContractError("finance filter 与 ticker 不匹配") if request.data_kind == "forecast_summary" and values.get("filter") != f'(SECURITY_CODE="{code}")': raise SourceContractError("forecast filter 与 ticker 不匹配") if request.data_kind == "forecast_detail" and values.get("code") != f"{market}{code}": raise SourceContractError("forecast detail code 与 ticker 不匹配") if contract.get("body_keys") is not None: if request.content_type.split(";", 1)[0].lower() != "application/x-www-form-urlencoded": raise SourceContractError("登记 body 必须为 form") try: body_items = urllib.parse.parse_qsl( (request.body or b"").decode("utf-8"), keep_blank_values=True ) except UnicodeDecodeError as exc: raise SourceContractError("form body 不是 UTF-8") from exc body_values = {key: value for key, value in body_items} if sorted(key for key, _ in body_items) != sorted(contract["body_keys"]): raise SourceContractError("form body 键不匹配") for key, expected in contract.get("fixed_body", {}).items(): if body_values.get(key) != str(expected): raise SourceContractError(f"form body 固定值不匹配：{key}") if body_values.get("stock", "").split(",", 1)[0] != code: raise SourceContractError("CNInfo stock body 与 ticker 不匹配") if body_values.get("seDate", "").split("~")[-1] != request.as_of: raise SourceContractError("CNInfo seDate 与 as-of 不匹配") if request.content_type.split(";", 1)[0].lower() not in contract.get( "request_content_types", ["application/json"] ): raise SourceContractError(f"未登记 request content-type：{request.content_type}") return contract def _headers(self, request: HttpRequest) -> dict[str, str]: headers = { "User-Agent": "project-info-stock-valuation-v2/2.0", "Accept": "application/json,text/html;q=0.8", **request.headers, } if request.body is not None: headers["Content-Type"] = request.content_type return headers def _fingerprint(self, request: HttpRequest) -> str: parsed = urllib.parse.urlsplit(request.url) headers = sorted((key.lower().strip(), value.strip()) for key, value in self._headers(request).items()) preimage = { "schema_version": 1, "provider_id": request.provider_id, "adapter_version": request.adapter_version, "method": request.method, "canonical_scheme_host_path": f"{parsed.scheme}://{parsed.hostname}{parsed.path}", "sorted_query": _query_items(request.query), "canonical_headers": headers, "canonical_body_sha256": sha256_bytes(_canonical_body(request.body, request.content_type)), "ticker": request.ticker, "as_of": request.as_of, "data_kind": request.data_kind, "fixture_identity": ( f"{self.fixture_identity}:{request.fixture_id}" if self.fixture_identity else None ), } return request_fingerprint(preimage) def _check_active(self, message: str = "network deadline exhausted") -> float: remaining = self.remaining() if remaining <= 0: raise NetworkBudgetExceeded(message) return remaining def fetch(self, request: HttpRequest) -> dict[str, Any]: contract = self._validate(request) fingerprint = self._fingerprint(request) negative = self._negative.get(fingerprint) if negative and time.monotonic() <= negative[0]: raise SourceContractError(f"本次运行负结果去重：{negative[1]}") state = self.baseline_states.get(request.data_kind) if self.baseline_states else None if state == "FRESH": baseline_hit = self.cache.load_baseline_blob( self.baseline_entries.get(request.data_kind, {}), request.provider_id, fingerprint, ) if baseline_hit is None: raise SourceContractError( f"FRESH baseline blob/fingerprint invalid: {request.data_kind}" ) return { "body": baseline_hit["body"], "meta": baseline_hit["meta"], "from_cache": True, "from_baseline": True, "fingerprint": fingerprint, } allow_logical_reuse = ( self.baseline_states is None or state == "FRESH" ) hit = ( self.cache.load_reusable(request.provider_id, fingerprint, self.process_start) if allow_logical_reuse else None ) if hit: self._record_telemetry(request, fingerprint, hit["meta"], True) return { "body": hit["body"], "meta": hit["meta"], "from_cache": True, "from_baseline": False, "fingerprint": fingerprint, } remaining_before = self._check_active() started = aware_now() if self.fixture_dir: assert self.fixture_manifest is not None entry = self.fixture_manifest["requests"].get(request.fixture_id) if entry is None: raise SourceContractError(f"fixture request 缺失：{request.fixture_id}") body = (self.fixture_dir / entry["file"]).read_bytes() status = int(entry.get("status", 200)) headers = {key.lower(): value for key, value in entry.get( "headers", {"content-type": request.content_type} ).items()} attempts: list[dict[str, Any]] = [] else: body, status, headers, attempts = self._fetch_network(request, contract) self._check_active("network deadline exhausted before cache archive") content_type = headers.get("content-type", "").split(";", 1)[0].strip().lower() allowed_response = contract.get("response_content_types", []) max_bytes = int(contract["max_response_bytes"]) ok = ( 200 <= status < 300 and 0 < len(body) <= max_bytes and content_type in allowed_response ) fetched = aware_now() raw_meta = { "run_id": self.run_id, "transport_complete": True, "http_status": status, "content_type": content_type, "parse_ok": False, "schema_ok": False, "as_of_ok": False, "semantic_status": "PENDING_PARSE" if ok else "GAP", "adapter_version": request.adapter_version, "method": request.method, "canonical_url": request.url, "query": _query_items(request.query), "canonical_headers": sorted( (key.lower(), value) for key, value in self._headers(request).items() ), "body_sha256": sha256_bytes(_canonical_body(request.body, request.content_type)), "ticker": request.ticker, "as_of": request.as_of, "data_kind": request.data_kind, "fixture_identity": self.fixture_identity, "started_at": iso(started), "finished_at": iso(fetched), "remaining_before": remaining_before, "remaining_after": self.remaining(), "attempts": attempts, } self._check_active("network deadline exhausted before raw archive") meta = self.cache.store( request.provider_id, fingerprint, body, fetched, request.ttl_seconds if ok else 60, reusable=False, raw_meta=raw_meta, ) self.cache_mutations += 1 self._record_telemetry(request, fingerprint, meta, False) if not ok: self._negative[fingerprint] = ( time.monotonic() + min(60.0, self.remaining()), f"status={status} bytes={len(body)} content-type={content_type}", ) raise SourceContractError( f"HTTP/响应合同失败：status={status} bytes={len(body)} content-type={content_type}" ) return { "body": body, "meta": meta, "from_cache": False, "from_baseline": False, "fingerprint": fingerprint, } def _record_telemetry( self, request: HttpRequest, fingerprint: str, meta: dict[str, Any], from_cache: bool, ) -> None: item = { "provider_id": request.provider_id, "data_kind": request.data_kind, "fingerprint": fingerprint, "raw_hash": meta.get("blob_hash"), "bytes": meta.get("bytes"), "http_status": meta.get("http_status"), "from_cache": from_cache, "started_at": meta.get("started_at"), "finished_at": meta.get("finished_at"), "remaining_before": meta.get("remaining_before"), "remaining_after": meta.get("remaining_after"), "attempts": meta.get("attempts", []), } with self._telemetry_lock: self.telemetry.append(item) def confirm_reusable(self, request: HttpRequest, response: dict[str, Any]) -> None: if response["from_cache"]: return self._check_active("network deadline exhausted before reusable promotion") response["meta"] = self.cache.promote_reusable( request.provider_id, response["fingerprint"], response["meta"] ) self.cache_mutations += 1 @staticmethod def _set_response_timeout(response: Any, timeout: float) -> None: candidates: Iterable[Any] = ( getattr(response, "fp", None), getattr(getattr(response, "fp", None), "raw", None), getattr(getattr(getattr(response, "fp", None), "raw", None), "_sock", None), ) for candidate in candidates: setter = getattr(candidate, "settimeout", None) if setter: try: setter(timeout) return except OSError: continue def _read_chunk_bounded(self, response: Any, size: int) -> bytes: """Bound even transports that ignore socket timeouts (notably test/error bodies).""" remaining = self._check_active("network deadline exhausted during read") self._set_response_timeout(response, remaining) completed = threading.Event() outcome: dict[str, Any] = {} def read_once() -> None: try: outcome["chunk"] = response.read(size) except BaseException as exc: # propagated in the caller thread outcome["error"] = exc finally: completed.set() worker = threading.Thread( target=read_once, name="valuation-v2-bounded-read", daemon=True, ) worker.start() if not completed.wait(timeout=remaining): closer = getattr(response, "close", None) if closer: try: closer() except OSError: pass raise NetworkBudgetExceeded("network deadline exhausted during read") error = outcome.get("error") if error is not None: raise error return bytes(outcome.get("chunk", b"")) def _read_response_bounded(self, response: Any, max_bytes: int) -> bytes: chunks: list[bytes] = [] total = 0 while True: chunk = self._read_chunk_bounded( response, min(65536, max_bytes - total + 1) ) if not chunk: break chunks.append(chunk) total += len(chunk) if total > max_bytes: raise SourceContractError( f"响应超过登记上限 {max_bytes} bytes" ) return b"".join(chunks) def _fetch_network( self, request: HttpRequest, contract: dict[str, Any] ) -> tuple[bytes, int, dict[str, str], list[dict[str, Any]]]: query = urllib.parse.urlencode(_query_items(request.query), doseq=True) url = request.url + ("?" + query if query else "") headers = self._headers(request) max_bytes = int(contract["max_response_bytes"]) last: Exception | None = None attempts: list[dict[str, Any]] = [] for attempt, backoff in enumerate((0.0, 0.25, 0.75), start=1): before = self._check_active() if backoff: time.sleep(min(backoff, before)) before = self._check_active("network deadline exhausted after backoff") timeout = min(20.0, before) if timeout <= 0: raise NetworkBudgetExceeded("network deadline exhausted before open") self.network_attempts += 1 req = urllib.request.Request( url, data=request.body, headers=headers, method=request.method ) attempt_started = aware_now() try: with self.opener.open(req, timeout=timeout) as response: body = self._read_response_bounded(response, max_bytes) total = len(body) attempts.append( { "attempt": attempt, "started_at": iso(attempt_started), "finished_at": iso(aware_now()), "remaining_before": before, "remaining_after": self.remaining(), "http_status": response.status, "bytes": total, "deadline_cause": None, } ) return body, response.status, { key.lower(): value for key, value in response.headers.items() }, attempts except urllib.error.HTTPError as exc: try: body = self._read_response_bounded(exc, max_bytes) finally: try: exc.close() except OSError: pass attempts.append( { "attempt": attempt, "started_at": iso(attempt_started), "finished_at": iso(aware_now()), "remaining_before": before, "remaining_after": self.remaining(), "http_status": exc.code, "bytes": len(body), "deadline_cause": None, } ) if 500 <= exc.code < 600 and attempt < 3: last = exc continue return body, exc.code, { key.lower(): value for key, value in exc.headers.items() }, attempts except NetworkBudgetExceeded: raise except (urllib.error.URLError, socket.timeout, TimeoutError, OSError) as exc: last = exc attempts.append( { "attempt": attempt, "started_at": iso(attempt_started), "finished_at": iso(aware_now()), "remaining_before": before, "remaining_after": self.remaining(), "http_status": None, "bytes": 0, "deadline_cause": type(exc).__name__, } ) if attempt < 3: continue break raise SourceContractError(f"网络请求失败：{last}")