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
|
