from __future__ import annotations
|
|
import hashlib
|
import json
|
import os
|
import tempfile
|
import threading
|
import time
|
from contextlib import contextmanager
|
from datetime import date, datetime, timedelta
|
from decimal import Decimal
|
from pathlib import Path
|
from typing import Any, Iterator
|
|
|
def canonical_bytes(value: Any) -> bytes:
|
def encode(item: Any) -> Any:
|
if isinstance(item, Decimal):
|
return format(item, "f")
|
if isinstance(item, Path):
|
return str(item)
|
raise TypeError(f"Object of type {type(item).__name__} is not JSON serializable")
|
|
return json.dumps(
|
value,
|
ensure_ascii=False,
|
sort_keys=True,
|
separators=(",", ":"),
|
default=encode,
|
).encode("utf-8")
|
|
|
def sha256_bytes(data: bytes) -> str:
|
return hashlib.sha256(data).hexdigest()
|
|
|
def atomic_write(path: Path, data: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
|
temp = Path(raw_temp)
|
try:
|
with os.fdopen(fd, "wb") as handle:
|
handle.write(data)
|
handle.flush()
|
os.fsync(handle.fileno())
|
os.replace(temp, path)
|
finally:
|
if temp.exists():
|
temp.unlink()
|
|
|
def request_fingerprint(preimage: dict[str, Any]) -> str:
|
return sha256_bytes(canonical_bytes(preimage))
|
|
|
DATA_KIND_TTLS = {
|
"stock_identity": 7 * 86400,
|
"announcement_index_recent": 6 * 3600,
|
"announcement_index_historical": 30 * 86400,
|
"market_close": 30 * 86400,
|
"shares_market_cap": 15 * 60,
|
"finance_main_recent": 6 * 3600,
|
"finance_main_historical": 30 * 86400,
|
"finance_income_recent": 6 * 3600,
|
"finance_income_historical": 30 * 86400,
|
"finance_balance_recent": 6 * 3600,
|
"finance_balance_historical": 30 * 86400,
|
"finance_cashflow_recent": 6 * 3600,
|
"finance_cashflow_historical": 30 * 86400,
|
"forecast_summary": 6 * 3600,
|
"forecast_detail": 6 * 3600,
|
}
|
|
PROVIDER_KINDS = {
|
"announcements": ("stock_identity", "announcement_index"),
|
"market": ("market_close", "shares_market_cap"),
|
"finance": ("finance_main", "finance_income", "finance_balance", "finance_cashflow"),
|
"forecast": ("forecast_summary", "forecast_detail"),
|
}
|
|
|
class BlobIntegrityError(OSError):
|
"""A content-addressed object could not be written and verified safely."""
|
|
_PROCESS_LOCKS: dict[str, threading.Lock] = {}
|
_PROCESS_LOCKS_GUARD = threading.Lock()
|
|
|
@contextmanager
|
def _ticker_lock(path: Path) -> Iterator[None]:
|
key = os.path.normcase(os.path.abspath(path))
|
with _PROCESS_LOCKS_GUARD:
|
local = _PROCESS_LOCKS.setdefault(key, threading.Lock())
|
with local:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
handle = open(path, "a+b")
|
try:
|
handle.seek(0)
|
if handle.tell() == 0 and path.stat().st_size == 0:
|
handle.write(b"0")
|
handle.flush()
|
if os.name == "nt":
|
import msvcrt
|
|
handle.seek(0)
|
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
|
else:
|
import fcntl
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
yield
|
finally:
|
try:
|
handle.seek(0)
|
if os.name == "nt":
|
import msvcrt
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
else:
|
import fcntl
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
finally:
|
handle.close()
|
|
|
class ContentCache:
|
def __init__(self, root: Path):
|
self.root = root
|
|
def load_reusable(
|
self, provider: str, fingerprint: str, process_start: datetime
|
) -> dict[str, Any] | None:
|
index = self.root / "reusable" / provider / f"{fingerprint}.json"
|
if not index.is_file():
|
return None
|
try:
|
meta = json.loads(index.read_text(encoding="utf-8"))
|
expires = datetime.fromisoformat(meta["expires_at"])
|
blob = self.root / meta["blob_path"]
|
data = blob.read_bytes()
|
except (OSError, ValueError, KeyError, json.JSONDecodeError):
|
return None
|
if process_start > expires or sha256_bytes(data) != meta.get("blob_hash"):
|
return None
|
if not (200 <= int(meta.get("http_status", 0)) < 300):
|
return None
|
if not all(
|
meta.get(key) is True
|
for key in ("transport_complete", "parse_ok", "schema_ok", "as_of_ok")
|
) or meta.get("semantic_status") != "OK":
|
return None
|
return {"body": data, "meta": meta}
|
|
def store(
|
self,
|
provider: str,
|
fingerprint: str,
|
body: bytes,
|
fetched_at: datetime,
|
ttl_seconds: int,
|
reusable: bool,
|
raw_meta: dict[str, Any],
|
) -> dict[str, Any]:
|
digest = sha256_bytes(body)
|
blob_rel = Path("blobs") / "sha256" / digest[:2] / f"{digest}.bin"
|
blob = self.root / blob_rel
|
blob_valid = False
|
if blob.is_file():
|
try:
|
blob_valid = sha256_bytes(blob.read_bytes()) == digest
|
except OSError:
|
blob_valid = False
|
if not blob_valid:
|
try:
|
atomic_write(blob, body)
|
except OSError as exc:
|
raise BlobIntegrityError(
|
f"内容寻址 blob 无法原子写入：{digest}"
|
) from exc
|
try:
|
if sha256_bytes(blob.read_bytes()) != digest:
|
raise BlobIntegrityError(f"内容寻址 blob 写后校验失败：{digest}")
|
except BlobIntegrityError:
|
raise
|
except OSError as exc:
|
raise BlobIntegrityError(f"内容寻址 blob 无法形成可信实物：{digest}") from exc
|
meta = {
|
**raw_meta,
|
"schema_version": 1,
|
"complete": True,
|
"provider_id": provider,
|
"fingerprint": fingerprint,
|
"fetched_at": fetched_at.isoformat(),
|
"expires_at": (fetched_at + timedelta(seconds=ttl_seconds)).isoformat(),
|
"blob_path": blob_rel.as_posix(),
|
"blob_hash": digest,
|
"bytes": len(body),
|
}
|
run_id = raw_meta.get("run_id", "unknown")
|
atomic_write(
|
self.root / "raw-index" / provider / fingerprint / f"{run_id}.json",
|
canonical_bytes(meta),
|
)
|
if reusable:
|
atomic_write(
|
self.root / "reusable" / provider / f"{fingerprint}.json",
|
canonical_bytes(meta),
|
)
|
return meta
|
|
def promote_reusable(
|
self, provider: str, fingerprint: str, meta: dict[str, Any]
|
) -> dict[str, Any]:
|
promoted = {
|
**meta,
|
"parse_ok": True,
|
"schema_ok": True,
|
"as_of_ok": True,
|
"semantic_status": "OK",
|
}
|
atomic_write(
|
self.root / "reusable" / provider / f"{fingerprint}.json",
|
canonical_bytes(promoted),
|
)
|
return promoted
|
|
def _baseline_material(self, baseline: dict[str, Any]) -> dict[str, Any]:
|
return {k: v for k, v in baseline.items() if k not in {"created_at", "data_hash"}}
|
|
def _baseline_valid(self, baseline: dict[str, Any]) -> bool:
|
"""Validate only the immutable baseline envelope, not every kind blob."""
|
expected = baseline.get("data_hash")
|
if not isinstance(expected, str) or len(expected) != 64:
|
return False
|
if sha256_bytes(canonical_bytes(self._baseline_material(baseline))) != expected:
|
return False
|
return True
|
|
def _data_kind_blob_valid(self, entry: dict[str, Any]) -> bool:
|
digest = str(entry.get("raw_hash", "")).lower()
|
if len(digest) != 64:
|
return False
|
blob = self.root / "blobs" / "sha256" / digest[:2] / f"{digest}.bin"
|
try:
|
return sha256_bytes(blob.read_bytes()) == digest
|
except OSError:
|
return False
|
|
@staticmethod
|
def _selection_key(item: dict[str, Any]) -> tuple[str, bytes, str]:
|
return (
|
item.get("baseline_as_of", ""),
|
canonical_bytes(item.get("watermark", {})),
|
str(item.get("data_hash", "")).lower(),
|
)
|
|
def select_baseline(self, ticker: str, requested_as_of: str) -> dict[str, Any] | None:
|
root = self.root / "companies" / ticker / "baselines"
|
candidates: list[dict[str, Any]] = []
|
if not root.is_dir():
|
return None
|
for path in root.glob("*.json"):
|
try:
|
item = json.loads(path.read_text(encoding="utf-8"))
|
except (OSError, json.JSONDecodeError):
|
continue
|
if item.get("ticker") != ticker or item.get("baseline_as_of", "9999-12-31") > requested_as_of:
|
continue
|
if self._baseline_valid(item):
|
candidates.append(item)
|
return max(candidates, key=self._selection_key) if candidates else None
|
|
@staticmethod
|
def data_kind_state(
|
entry: dict[str, Any] | None,
|
requested_as_of: str,
|
process_start: datetime,
|
adapter_version: str = "1.0.0",
|
) -> str:
|
if entry is None:
|
return "MISSING"
|
required = (
|
"data_kind",
|
"provider_id",
|
"request_fingerprint",
|
"ticker",
|
"requested_as_of",
|
"baseline_as_of",
|
"data_date",
|
"fetched_at",
|
"expires_at",
|
"adapter_version",
|
"raw_hash",
|
"schema_status",
|
"as_of_status",
|
"semantic_status",
|
"watermark",
|
)
|
if any(key not in entry for key in required):
|
return "STALE"
|
try:
|
expires = datetime.fromisoformat(entry["expires_at"])
|
except (TypeError, ValueError):
|
return "STALE"
|
if entry["baseline_as_of"] > requested_as_of or entry["data_date"] > requested_as_of:
|
return "STALE"
|
if entry["requested_as_of"] != requested_as_of:
|
return "STALE"
|
publish = entry.get("publish_date")
|
if entry.get("requires_publish_date", False) and (
|
not publish or publish > requested_as_of
|
):
|
return "STALE"
|
if process_start > expires or entry["adapter_version"] != adapter_version:
|
return "STALE"
|
if entry["schema_status"] != "PASS" or entry["as_of_status"] != "PASS":
|
return "STALE"
|
if entry["semantic_status"] != "OK" or len(str(entry["raw_hash"])) != 64:
|
return "STALE"
|
kind = entry["data_kind"]
|
if kind == "market_close" and not entry.get("is_latest_eligible_trade_date"):
|
return "STALE"
|
if kind == "shares_market_cap" and not entry.get("historical_capture_valid"):
|
return "STALE"
|
if kind.startswith("finance_") and not entry.get("required_periods_complete"):
|
return "STALE"
|
if kind == "forecast_summary" and not entry.get("summary_complete"):
|
return "STALE"
|
if kind == "forecast_detail" and not entry.get("detail_complete"):
|
return "STALE"
|
return "FRESH"
|
|
def load_baseline_blob(
|
self,
|
entry: dict[str, Any],
|
provider_id: str,
|
request_fingerprint_value: str,
|
) -> dict[str, Any] | None:
|
"""Load a semantically validated per-kind baseline without a request index."""
|
if (
|
entry.get("provider_id") != provider_id
|
or entry.get("request_fingerprint") != request_fingerprint_value
|
):
|
return None
|
digest = str(entry.get("raw_hash", "")).lower()
|
if len(digest) != 64:
|
return None
|
blob = self.root / "blobs" / "sha256" / digest[:2] / f"{digest}.bin"
|
try:
|
body = blob.read_bytes()
|
except OSError:
|
return None
|
if sha256_bytes(body) != digest:
|
return None
|
meta = {
|
"schema_version": 1,
|
"provider_id": provider_id,
|
"fingerprint": request_fingerprint_value,
|
"blob_path": blob.relative_to(self.root).as_posix(),
|
"blob_hash": digest,
|
"bytes": len(body),
|
"http_status": 200,
|
"transport_complete": True,
|
"parse_ok": True,
|
"schema_ok": True,
|
"as_of_ok": True,
|
"semantic_status": "OK",
|
"fetched_at": entry["fetched_at"],
|
"expires_at": entry["expires_at"],
|
"data_kind": entry["data_kind"],
|
"started_at": None,
|
"finished_at": None,
|
"remaining_before": None,
|
"remaining_after": None,
|
"attempts": [],
|
"baseline_reused": True,
|
}
|
return {"body": body, "meta": meta}
|
|
def baseline_states(
|
self, ticker: str, requested_as_of: str, process_start: datetime
|
) -> tuple[dict[str, Any] | None, dict[str, str]]:
|
baseline = self.select_baseline(ticker, requested_as_of)
|
entries = (baseline or {}).get("data_kinds", {})
|
all_kinds = {kind for kinds in PROVIDER_KINDS.values() for kind in kinds}
|
states: dict[str, str] = {}
|
for kind in sorted(all_kinds):
|
entry = entries.get(kind)
|
state = self.data_kind_state(entry, requested_as_of, process_start)
|
if state == "FRESH" and not self._data_kind_blob_valid(entry):
|
state = "STALE"
|
states[kind] = state
|
return baseline, states
|
|
def baseline_entries_complete_and_valid(
|
self, entries: dict[str, dict[str, Any]]
|
) -> bool:
|
"""A company baseline advances only when every kind has a trusted blob."""
|
expected = {kind for kinds in PROVIDER_KINDS.values() for kind in kinds}
|
return set(entries) == expected and all(
|
self._data_kind_blob_valid(entries[kind]) for kind in expected
|
)
|
|
def fresh_provider_results(
|
self, ticker: str, requested_as_of: str, process_start: datetime
|
) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
|
baseline, states = self.baseline_states(ticker, requested_as_of, process_start)
|
fresh: dict[str, dict[str, Any]] = {}
|
if baseline:
|
for name, kinds in PROVIDER_KINDS.items():
|
if all(states.get(kind) == "FRESH" for kind in kinds):
|
result = json.loads(json.dumps(baseline["provider_results"][name]))
|
result["baseline_reused"] = True
|
result["request_telemetry"] = []
|
fresh[name] = result
|
return fresh, states
|
|
def write_baseline(
|
self,
|
ticker: str,
|
baseline: dict[str, Any],
|
supersede_data_hash: str | None = None,
|
) -> Path:
|
baseline = dict(baseline)
|
baseline["data_hash"] = sha256_bytes(canonical_bytes(self._baseline_material(baseline)))
|
target = (
|
self.root
|
/ "companies"
|
/ ticker
|
/ "baselines"
|
/ f"{baseline['baseline_as_of']}-{baseline['data_hash']}.json"
|
)
|
lock_path = self.root / "companies" / ticker / ".current.lock"
|
with _ticker_lock(lock_path):
|
atomic_write(target, canonical_bytes(baseline))
|
if supersede_data_hash and supersede_data_hash != baseline["data_hash"]:
|
old = (
|
self.root
|
/ "companies"
|
/ ticker
|
/ "baselines"
|
/ f"{baseline['baseline_as_of']}-{supersede_data_hash}.json"
|
)
|
if old.is_file():
|
history = self.root / "companies" / ticker / "history" / old.name
|
history.parent.mkdir(parents=True, exist_ok=True)
|
os.replace(old, history)
|
current = self.root / "companies" / ticker / "current.json"
|
candidate = {
|
"baseline_as_of": baseline["baseline_as_of"],
|
"watermark": baseline.get("watermark", {}),
|
"data_hash": baseline["data_hash"].lower(),
|
"path": target.relative_to(self.root).as_posix(),
|
}
|
existing = None
|
if current.is_file():
|
try:
|
existing = json.loads(current.read_text(encoding="utf-8"))
|
except (OSError, json.JSONDecodeError):
|
pass
|
existing_is_superseded = (
|
existing is not None
|
and supersede_data_hash is not None
|
and existing.get("data_hash") == supersede_data_hash.lower()
|
)
|
if (
|
existing is None
|
or existing_is_superseded
|
or self._selection_key(candidate) > self._selection_key(existing)
|
):
|
atomic_write(current, canonical_bytes(candidate))
|
return target
|
