from __future__ import annotations import hashlib import json import os import shutil import stat import tempfile import time import uuid import warnings from datetime import date, datetime from pathlib import Path from typing import Any, Callable from .acquisition import acquire_all from .cache import ContentCache, atomic_write, canonical_bytes from .full_report import render_report from .http_client import HttpClient from .judgment import apply_overlay, load_overlay from .report_qa import run_qa, verify_manifest from .snapshot_builder import CoreInputError, build_data_snapshot from .telemetry import RunClock, aware_now, iso from .v1_bridge import V1Bridge class InputContractError(ValueError): pass def _hash_file(path: Path) -> dict[str, Any]: data = path.read_bytes() return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} def _atomic_workflow_write( path: Path, data: bytes, inject: Callable[[str], None], label: str, ) -> None: """Atomic write with every durable boundary exposed to fault injection.""" path.parent.mkdir(parents=True, exist_ok=True) inject(f"file:{label}:temp") fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temp = Path(raw_temp) try: with os.fdopen(fd, "wb") as handle: inject(f"file:{label}:write") handle.write(data) inject(f"file:{label}:flush") handle.flush() inject(f"file:{label}:fsync") os.fsync(handle.fileno()) inject(f"file:{label}:replace") os.replace(temp, path) finally: if temp.exists(): temp.unlink() def _is_reparse(path: Path) -> bool: try: attrs = path.lstat().st_file_attributes # type: ignore[attr-defined] except (AttributeError, OSError): return path.is_symlink() return bool(attrs & stat.FILE_ATTRIBUTE_REPARSE_POINT) def _lexical_absolute(path: Path) -> Path: return Path(os.path.abspath(os.fspath(path))) def _lexists(path: Path) -> bool: return os.path.lexists(path) def _validate_parent_chain(path: Path, label: str) -> None: for parent in path.parents: if not _lexists(parent): continue if parent.is_symlink() or _is_reparse(parent): raise InputContractError(f"{label} 父链不得含链接/reparse：{parent}") if not parent.is_dir(): raise InputContractError(f"{label} 父链必须是目录：{parent}") def _validate_directory(path: Path, label: str, allow_existing: bool) -> None: _validate_parent_chain(path, label) if not _lexists(path): return if path.is_symlink() or _is_reparse(path): raise InputContractError(f"{label} 不得是链接或 reparse point：{path}") if not path.is_dir(): raise InputContractError(f"{label} 必须是目录：{path}") if not allow_existing: raise InputContractError(f"{label} 已存在：{path}") def _validate_file(path: Path, label: str, must_exist: bool) -> None: _validate_parent_chain(path, label) if not _lexists(path): if must_exist: raise InputContractError(f"{label} 不存在：{path}") return if path.is_symlink() or _is_reparse(path) or not path.is_file(): raise InputContractError(f"{label} 必须是普通文件：{path}") def _input_summary( clock: RunClock, error: BaseException, task_start_valid: bool = False ) -> dict[str, Any]: terminal = aware_now() process = max(0.0, time.monotonic() - clock.process_start_mono) task = None if task_start_valid and clock.task_start_wall is not None: task, _ = clock.walls(terminal) return { "schema_version": 1, "status": "INPUT_ERROR", "run_id": None, "exit_code": 2, "output_dir": None, "failed_dir": None, "manifest_path": None, "manifest_sha256": None, "receipt_path": None, "receipt_sha256": None, "receipt_bytes": None, "task_start": iso(clock.task_start_wall) if task is not None else None, "task_start_source": clock.task_start_source if task is not None else None, "process_start": iso(clock.process_start_wall), "terminal_at": iso(terminal), "task_wall_seconds": task, "process_wall_seconds": process, "gap_count": None, "error_code": "E_INPUT_CONTRACT", "error": f"{type(error).__name__}: {error}", } def _terminal_summary( *, clock: RunClock, status: str, exit_code: int, run_id: str, output_dir: Path, failed_dir: Path | None, manifest_path: Path | None, manifest_sha256: str | None, receipt_path: Path | None, receipt_sha256: str | None, receipt_bytes: int | None, terminal: datetime, task_wall: float, process_wall: float, gap_count: int, error_code: str | None, error: str | None, ) -> dict[str, Any]: return { "schema_version": 1, "status": status, "run_id": run_id, "exit_code": exit_code, "output_dir": str(output_dir), "failed_dir": str(failed_dir) if failed_dir else None, "manifest_path": str(manifest_path) if manifest_path else None, "manifest_sha256": manifest_sha256, "receipt_path": str(receipt_path) if receipt_path else None, "receipt_sha256": receipt_sha256, "receipt_bytes": receipt_bytes, "task_start": iso(clock.task_start_wall), "task_start_source": clock.task_start_source, "process_start": iso(clock.process_start_wall), "terminal_at": iso(terminal), "task_wall_seconds": task_wall, "process_wall_seconds": process_wall, "gap_count": gap_count, "error_code": error_code, "error": error, } def _clean_owned(path: Path) -> None: if _lexists(path) and path.is_dir() and not path.is_symlink() and not _is_reparse(path): shutil.rmtree(path) def _failure_package( failed: Path, clock: RunClock, run_id: str, status: str, error_code: str, original: BaseException, rollback_errors: list[str], provider_results: dict[str, Any] | None, ) -> None: _clean_owned(failed) failed.mkdir(parents=True) terminal = aware_now() task, process = clock.walls(terminal) atomic_write( failed / "failure.json", canonical_bytes( { "schema_version": 2, "complete": False, "run_id": run_id, "status": status, "error_code": error_code, "error_type": type(original).__name__, "error": str(original), "rollback_errors": rollback_errors, "recorded_at": iso(terminal), } ), ) atomic_write( failed / "runtime_metrics.json", canonical_bytes( { "schema_version": 1, "run_id": run_id, "status": status, "task_start": iso(clock.task_start_wall), "task_start_source": clock.task_start_source, "process_start": iso(clock.process_start_wall), "terminal_at": iso(terminal), "task_wall_seconds": task, "process_wall_seconds": process, "wall_scope": "through_failure_recorded", "phases": clock.phases, "providers": clock.providers, } ), ) message = f"{status} {error_code}: {type(original).__name__}: {original}\n" if rollback_errors: message += "rollback_errors=" + json.dumps(rollback_errors, ensure_ascii=False) + "\n" atomic_write(failed / "runtime.log", message.encode("utf-8")) if provider_results: atomic_write(failed / "provider_results.json", canonical_bytes(provider_results)) raw_hashes = sorted( { digest for result in provider_results.values() for digest in result.get("raw_artifact_hashes", []) } ) atomic_write( failed / "source_evidence.json", canonical_bytes({"schema_version": 1, "raw_hashes": raw_hashes}), ) def _reserved_conflicts(parent: Path, output_name: str) -> list[Path]: prefixes = ( f".{output_name}.staging-", f".{output_name}.backup-", f".{output_name}.backup-receipt-", f".{output_name}.failed-new-", f"{output_name}.failed-", ) if not parent.is_dir(): return [] return [item for item in parent.iterdir() if any(item.name.startswith(prefix) for prefix in prefixes)] def run_ticker_pipeline( *, ticker: str, as_of: str, output_dir: Path, cache_dir: Path, judgment_path: Path | None = None, fixture_dir: Path | None = None, task_start: str | None = None, force: bool = False, network_budget_seconds: float = 90.0, fault_injector: Callable[[str], None] | None = None, ) -> tuple[int, dict[str, Any]]: clock = RunClock() task_valid = False try: clock.set_task_start(task_start) task_valid = True if not __import__("re").fullmatch(r"\d{6}\.(SZ|SH|BJ)", ticker): raise InputContractError("ticker 必须是 NNNNNN.SZ|SH|BJ") as_of_date = date.fromisoformat(as_of) if as_of_date > date.today(): raise InputContractError("as-of 不得晚于当前日期") output_dir = _lexical_absolute(output_dir) cache_dir = _lexical_absolute(cache_dir) if fixture_dir is not None: fixture_dir = _lexical_absolute(fixture_dir) _validate_directory(fixture_dir, "fixture-dir", True) if judgment_path is not None: judgment_path = _lexical_absolute(judgment_path) _validate_file(judgment_path, "judgment", True) _validate_directory(output_dir, "output-dir", force) _validate_directory(cache_dir, "cache-dir", True) parent = output_dir.parent _validate_directory(parent, "output parent", True) receipt = parent / f"{output_dir.name}.commit-receipt.json" _validate_file(receipt, "receipt", False) if _lexists(receipt) and not force: raise InputContractError(f"receipt 已存在：{receipt}") conflicts = _reserved_conflicts(parent, output_dir.name) if conflicts: raise InputContractError(f"运行路径冲突：{conflicts[0]}") except (ValueError, OSError) as exc: return 2, _input_summary(clock, exc, task_valid) run_id = uuid.uuid4().hex receipt = output_dir.parent / f"{output_dir.name}.commit-receipt.json" stage = output_dir.parent / f".{output_dir.name}.staging-{run_id}" backup = output_dir.parent / f".{output_dir.name}.backup-{run_id}" backup_receipt = output_dir.parent / f".{output_dir.name}.backup-receipt-{run_id}.json" failed = output_dir.parent / f"{output_dir.name}.failed-{run_id}" provider_results: dict[str, Any] | None = None gaps: list[dict[str, Any]] = [] old_moved = False old_receipt_moved = False new_installed = False receipt_written = False def inject(point: str) -> None: if fault_injector: fault_injector(point) def stage_write(relative: str, value: Any, raw: bool = False) -> None: inject(f"write:{relative}") data = value if raw else canonical_bytes(value) _atomic_workflow_write(stage / relative, data, inject, relative) try: inject("mkdir_staging") stage.mkdir(parents=True) bridge = V1Bridge.load() registry_path = Path(__file__).with_name("provider_registry.json") registry = json.loads(registry_path.read_text(encoding="utf-8")) cache_dir.mkdir(parents=True, exist_ok=True) cache = ContentCache(cache_dir) selected_baseline, baseline_states = cache.baseline_states( ticker, as_of, clock.process_start_wall ) baseline_entries = dict((selected_baseline or {}).get("data_kinds", {})) clock.begin("acquisition") network_started = time.monotonic() client = HttpClient( registry, cache, network_started + network_budget_seconds, run_id, clock.process_start_wall, fixture_dir, baseline_states, baseline_entries, ) provider_results = acquire_all(client, ticker, as_of) clock.end("acquisition") clock.providers = list(client.telemetry) gaps = [gap for result in provider_results.values() for gap in result.get("gaps", [])] clock.begin("snapshot") data_snapshot, build_report = build_data_snapshot(ticker, as_of, provider_results) clock.end("snapshot") build_report["baseline_states_before"] = baseline_states results = None working_snapshot = data_snapshot if judgment_path: clock.begin("judgment_and_v1") overlay = load_overlay(judgment_path) working_snapshot = apply_overlay(data_snapshot, overlay) stage_write("valuation_snapshot.json", working_snapshot) results = bridge.compute(stage / "valuation_snapshot.json") stage_write("valuation_results.json", results) clock.end("judgment_and_v1") status = "COMPLETE_WITH_GAPS" if gaps else "COMPLETE" else: gaps.append( { "gap_id": "W_JUDGMENT_REQUIRED", "provider": "human_judgment", "field": "judgment_overlay", "reason": "机械数据已就绪，尚未提供最小人工判断覆盖层", "impact": "不得生成方向性估值结论", "blocking": False, "budget_used_seconds": None, "manual_action": "提供受 schema 约束的 judgment overlay", } ) status = "DATA_READY_NEEDS_JUDGMENT" build_report["gaps"] = gaps stage_write("data_snapshot.json", data_snapshot) stage_write("snapshot_build_report.json", build_report) stage_write("provider_results.json", provider_results) stage_write("gaps.json", gaps) stage_write( "source_evidence.json", { "schema_version": 2, "sources": data_snapshot["sources"], "raw_hashes": build_report["raw_hashes"], "field_lineage": build_report["field_lineage"], }, ) data_kinds = { kind: entry for result in provider_results.values() for kind, entry in result.get("data_kinds", {}).items() } baseline = { "schema_version": 1, "adapter_bundle_version": "2.1.0", "ticker": ticker, "baseline_as_of": as_of, "created_at": iso(aware_now()), "watermark": { "announcement_time": max( (source["publish_date"] for source in data_snapshot["sources"]), default="", ), "report_date": data_snapshot["meta"]["report_period_end"], "market_date": data_snapshot["market"]["shares_date"], "forecast_date": max( ( item.get("report_date") or "" for item in data_snapshot["institutions"].get("forecasts", []) ), default="", ), }, "provider_results": provider_results, "data_kinds": data_kinds, "field_lineage": build_report["field_lineage"], "source_hashes": build_report["raw_hashes"], } if client.remaining() <= 0: raise CoreInputError("E_NETWORK_DEADLINE：截止后禁止提交公司基线") cache_integrity_failure = any( result.get("cache_integrity_failure") is True for result in provider_results.values() ) if ( not cache_integrity_failure and cache.baseline_entries_complete_and_valid(data_kinds) ): cache.write_baseline( ticker, baseline, supersede_data_hash=(selected_baseline or {}).get("data_hash"), ) clock.begin("report_and_qa") report = render_report(working_snapshot, build_report, gaps, results) stage_write("report.md", report.encode("utf-8"), raw=True) qa = run_qa(report, working_snapshot, build_report, results, stage) stage_write("qa_report.json", qa) if qa["status"] != "PASS": raise RuntimeError("QA_FAIL: " + "；".join(qa["errors"])) clock.end("report_and_qa") stage_write("runtime.log", b"pipeline completed through commit-ready\n", raw=True) commit_ready = aware_now() stage_write("runtime_metrics.json", clock.metrics(run_id, status, commit_ready)) inject("manifest") artifacts = { path.relative_to(stage).as_posix(): _hash_file(path) for path in sorted(stage.rglob("*")) if path.is_file() } manifest = { "schema_version": 1, "run_id": run_id, "status": status, "complete": True, "ticker": ticker, "as_of": as_of, "artifacts": artifacts, "v1_contract": bridge.metadata(), } _atomic_workflow_write( stage / "manifest.json", canonical_bytes(manifest), inject, "manifest.json" ) manifest_meta = _hash_file(stage / "manifest.json") if _lexists(output_dir): inject("rename_old_to_backup") os.replace(output_dir, backup) old_moved = True if _lexists(receipt): os.replace(receipt, backup_receipt) old_receipt_moved = True inject("rename_stage_to_output") os.replace(stage, output_dir) new_installed = True manifest_errors = verify_manifest(output_dir) if manifest_errors: raise RuntimeError("MANIFEST_VERIFY_FAIL: " + "；".join(manifest_errors)) receipt_write_started = aware_now() receipt_payload = { "schema_version": 2, "run_id": run_id, "status": status, "output_path": str(output_dir), "manifest_path": str(output_dir / "manifest.json"), "manifest_sha256": manifest_meta["sha256"], "manifest_bytes": manifest_meta["bytes"], "commit_ready_at": iso(commit_ready), "receipt_write_started_at": iso(receipt_write_started), "task_start": iso(clock.task_start_wall), "task_start_source": clock.task_start_source, "process_start": iso(clock.process_start_wall), "wall_scope": "through_receipt_write_started", } inject("receipt") _atomic_workflow_write( receipt, canonical_bytes(receipt_payload), inject, "receipt" ) inject("receipt:read") receipt_bytes_value = receipt.read_bytes() inject("receipt:hash") receipt_digest = hashlib.sha256(receipt_bytes_value).hexdigest() inject("receipt:stat") receipt_size = receipt.stat().st_size if receipt_size != len(receipt_bytes_value): raise RuntimeError("RECEIPT_VERIFY_FAIL: bytes changed during validation") if json.loads(receipt_bytes_value.decode("utf-8")) != receipt_payload: raise RuntimeError("RECEIPT_VERIFY_FAIL: canonical payload mismatch") receipt_meta = {"bytes": receipt_size, "sha256": receipt_digest} receipt_written = True cleanup_warning: BaseException | None = None if _lexists(backup): try: inject("cleanup_backup") shutil.rmtree(backup) if _lexists(backup_receipt): backup_receipt.unlink() except BaseException as exc: cleanup_warning = exc elif _lexists(backup_receipt): try: backup_receipt.unlink() except BaseException as exc: cleanup_warning = exc if cleanup_warning is not None: # New output and receipt are already the durable truth. Retain backup evidence. warnings.warn( f"backup cleanup failed; new output remains authoritative: {cleanup_warning}", RuntimeWarning, ) terminal = aware_now() task_wall, process_wall = clock.walls(terminal) return 0, _terminal_summary( clock=clock, status=status, exit_code=0, run_id=run_id, output_dir=output_dir, failed_dir=None, manifest_path=output_dir / "manifest.json", manifest_sha256=manifest_meta["sha256"], receipt_path=receipt, receipt_sha256=receipt_meta["sha256"], receipt_bytes=receipt_meta["bytes"], terminal=terminal, task_wall=task_wall, process_wall=process_wall, gap_count=len(gaps), error_code=None, error=None, ) except BaseException as original: rollback_errors: list[str] = [] try: if old_moved and not new_installed and _lexists(backup): inject("restore") os.replace(backup, output_dir) old_moved = False elif new_installed and not receipt_written: orphan = output_dir.parent / f".{output_dir.name}.failed-new-{run_id}" if _lexists(output_dir): os.replace(output_dir, orphan) if _lexists(backup): inject("restore") os.replace(backup, output_dir) old_moved = False _clean_owned(orphan) if not old_receipt_moved and _lexists(receipt): receipt.unlink() if old_receipt_moved and _lexists(backup_receipt): os.replace(backup_receipt, receipt) old_receipt_moved = False except BaseException as rollback_exc: rollback_errors.append(f"{type(rollback_exc).__name__}: {rollback_exc}") try: _clean_owned(stage) except BaseException as cleanup_exc: rollback_errors.append(f"cleanup {type(cleanup_exc).__name__}: {cleanup_exc}") status = "BLOCKED" if isinstance(original, CoreInputError) else "FAILED" exit_code = 4 if status == "BLOCKED" else 5 error_code = "E_CORE_INPUT" if status == "BLOCKED" else "E_RUNTIME" _failure_package( failed, clock, run_id, status, error_code, original, rollback_errors, provider_results, ) terminal = aware_now() task_wall, process_wall = clock.walls(terminal) summary = _terminal_summary( clock=clock, status=status, exit_code=exit_code, run_id=run_id, output_dir=output_dir, failed_dir=failed, manifest_path=None, manifest_sha256=None, receipt_path=None, receipt_sha256=None, receipt_bytes=None, terminal=terminal, task_wall=task_wall, process_wall=process_wall, gap_count=len(gaps), error_code=error_code, error=f"{type(original).__name__}: {original}", ) if isinstance(original, (KeyboardInterrupt, SystemExit)): raise return exit_code, summary