from __future__ import annotations
|
|
import hashlib
|
import importlib
|
import inspect
|
import json
|
import sys
|
from dataclasses import dataclass
|
from pathlib import Path
|
from typing import Any
|
|
|
EXPECTED = {
|
"__init__.py": (257, "F39479C1A7B7F40A1A71D5EDE8A287F4E701B4F1473D674F04A2E1BA1B522D29"),
|
"core.py": (53165, "6AB34917A8B136F684EF655995428ACB5936050E1DA36B7B2371941A3E5A5F8F"),
|
"cli.py": (1811, "260692D62C8983B623483E0AA29C763869E30D40952700D46C636357A7376A1E"),
|
"source_registry.json": (4958, "0895D4F0559A1353787059D506DF16572E6818CD847C21042FE1D1106AA854E0"),
|
}
|
|
|
class V1ContractError(RuntimeError):
|
pass
|
|
|
def _sha(path: Path) -> str:
|
return hashlib.sha256(path.read_bytes()).hexdigest().upper()
|
|
|
@dataclass(frozen=True)
|
class V1Bridge:
|
module: Any
|
package_dir: Path
|
registry_path: Path
|
fingerprints: dict[str, dict[str, Any]]
|
|
@classmethod
|
def load(cls) -> "V1Bridge":
|
repo = Path(__file__).resolve().parents[3]
|
parent = repo / "dev" / "ana-dev"
|
package = parent / "stock_valuation_pipeline"
|
fingerprints: dict[str, dict[str, Any]] = {}
|
for name, (size, digest) in EXPECTED.items():
|
path = package / name
|
if not path.is_file() or path.stat().st_size != size or _sha(path) != digest:
|
raise V1ContractError(f"V1 合同漂移：{path}")
|
fingerprints[name] = {"bytes": size, "sha256": digest}
|
parent_text = str(parent)
|
if parent_text not in sys.path:
|
sys.path.insert(0, parent_text)
|
module = importlib.import_module("stock_valuation_pipeline")
|
module_file = Path(module.__file__).resolve()
|
if package.resolve() not in module_file.parents or module.__version__ != "1.0.0":
|
raise V1ContractError("V1 模块加载路径或版本不匹配")
|
expected_signatures = {
|
"run_pipeline": "(snapshot_path: 'Path', output_dir: 'Path', source_registry_path: 'Path', *, force: 'bool' = False) -> 'dict[str, Any]'",
|
"compute_valuation": "(snapshot: 'dict[str, Any]', registry: 'dict[str, Any]') -> 'dict[str, Any]'",
|
}
|
for name, expected in expected_signatures.items():
|
actual = str(inspect.signature(getattr(module, name)))
|
if actual != expected:
|
raise V1ContractError(f"V1 {name} 签名漂移：{actual}")
|
return cls(module, package, package / "source_registry.json", fingerprints)
|
|
def metadata(self) -> dict[str, Any]:
|
return {
|
"version": self.module.__version__,
|
"module_path": str(Path(self.module.__file__).resolve()),
|
"registry_path": str(self.registry_path.resolve()),
|
"files": self.fingerprints,
|
}
|
|
def run_input(self, snapshot_path: Path, output_dir: Path, force: bool) -> dict[str, Any]:
|
return self.module.run_pipeline(
|
snapshot_path=snapshot_path,
|
output_dir=output_dir,
|
source_registry_path=self.registry_path,
|
force=force,
|
)
|
|
def compute(self, snapshot_path: Path) -> dict[str, Any]:
|
snapshot = self.module.load_snapshot(snapshot_path)
|
registry = json.loads(self.registry_path.read_text(encoding="utf-8"))
|
return self.module.compute_valuation(snapshot, registry)
|
