from __future__ import annotations
|
|
import hashlib
|
import json
|
import re
|
from pathlib import Path
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
RESULT_ROOT = ROOT / "ana-data" / "result" / "股票估值"
|
BATCH_DIR = RESULT_ROOT / "20260805_batch_missing_image_stocks_valuation"
|
CASE_DIR = ROOT / "ana-data" / "cases" / "股票估值" / "BATCH-STOCK-VALUATION-20260805-004"
|
|
|
def read_json(path: Path):
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
def sha256(path: Path) -> str:
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def main() -> None:
|
batch = read_json(BATCH_DIR / "batch_results.json")
|
manifest = read_json(CASE_DIR / "batch_manifest.json")
|
rows = batch["items"]
|
errors: list[str] = []
|
checks = {
|
"row_count": len(rows),
|
"unique_tickers": len({item["ticker"] for item in rows}),
|
"formal_reports": 0,
|
"snapshots": 0,
|
"source_manifests": 0,
|
"calculation_results": 0,
|
"run_manifests": 0,
|
"reports_with_16_sections": 0,
|
"reports_without_placeholders": 0,
|
"qa_error_count": 0,
|
"qa_warning_count": 0,
|
"market_cap_reconciliation_max_gap": 0.0,
|
"summary_links_checked": 0,
|
}
|
report_hashes = []
|
for row in rows:
|
formal = RESULT_ROOT / row["formal_path"]
|
out_dir = formal.parent
|
snapshots = list(out_dir.glob("*估值快照_20260804.json"))
|
evidence = out_dir / "source_evidence_manifest.json"
|
results_path = out_dir / "calculation" / "valuation_results.json"
|
run_manifest = out_dir / "calculation" / "run_manifest.json"
|
for path, key in (
|
(formal, "formal_reports"),
|
(evidence, "source_manifests"),
|
(results_path, "calculation_results"),
|
(run_manifest, "run_manifests"),
|
):
|
if path.exists():
|
checks[key] += 1
|
else:
|
errors.append(f"missing {path}")
|
if len(snapshots) != 1:
|
errors.append(f"snapshot count {len(snapshots)} for {row['ticker']}")
|
continue
|
checks["snapshots"] += 1
|
text = formal.read_text(encoding="utf-8")
|
headings = {int(item) for item in re.findall(r"^## (\d+)\.", text, flags=re.MULTILINE)}
|
if headings == set(range(16)):
|
checks["reports_with_16_sections"] += 1
|
else:
|
errors.append(f"section mismatch {row['ticker']}: {sorted(headings)}")
|
if not re.search(r"\b(?:TODO|TBD|PLACEHOLDER)\b", text, flags=re.IGNORECASE):
|
checks["reports_without_placeholders"] += 1
|
else:
|
errors.append(f"placeholder {row['ticker']}")
|
snapshot = read_json(snapshots[0])
|
results = read_json(results_path)
|
checks["qa_error_count"] += results["qa"]["error_count"]
|
checks["qa_warning_count"] += results["qa"]["warning_count"]
|
expected_cap = float(snapshot["market"]["price"]) * float(snapshot["market"]["diluted_shares"])
|
actual_cap = float(results["metrics"]["market_cap"])
|
gap = abs(expected_cap - actual_cap) / expected_cap
|
checks["market_cap_reconciliation_max_gap"] = max(checks["market_cap_reconciliation_max_gap"], gap)
|
if gap > 1e-12:
|
errors.append(f"market cap mismatch {row['ticker']}: {gap}")
|
report_hashes.append({"ticker": row["ticker"], "path": row["formal_path"], "sha256": sha256(formal)})
|
|
summary = BATCH_DIR / "八张图片未评估股票估值批次汇总_20260805.md"
|
summary_text = summary.read_text(encoding="utf-8")
|
for relative in re.findall(r"\]\((\.\./[^)]+)\)", summary_text):
|
checks["summary_links_checked"] += 1
|
if not (summary.parent / relative).resolve().exists():
|
errors.append(f"broken summary link {relative}")
|
screenshot_checks = []
|
for item in manifest["screenshots"]:
|
path = ROOT / item["path"]
|
matched = path.exists() and sha256(path) == item["sha256"]
|
screenshot_checks.append({"path": item["path"], "hash_matches": matched})
|
if not matched:
|
errors.append(f"screenshot mismatch {item['path']}")
|
for key in (
|
"row_count", "unique_tickers", "formal_reports", "snapshots", "source_manifests",
|
"calculation_results", "run_manifests", "reports_with_16_sections",
|
"reports_without_placeholders", "summary_links_checked",
|
):
|
if checks[key] != 58:
|
errors.append(f"{key}={checks[key]}, expected 58")
|
if checks["qa_error_count"]:
|
errors.append(f"qa_error_count={checks['qa_error_count']}")
|
acceptance = {
|
"batch_id": batch["batch_id"],
|
"as_of": batch["as_of"],
|
"status": "PASS" if not errors else "FAIL",
|
"checks": checks,
|
"screenshots": screenshot_checks,
|
"formal_reports": report_hashes,
|
"errors": errors,
|
}
|
acceptance_path = BATCH_DIR / "batch_acceptance.json"
|
acceptance_path.write_text(json.dumps(acceptance, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
manifest["acceptance"] = {
|
"path": str(acceptance_path.relative_to(ROOT)).replace("\\", "/"),
|
"sha256": sha256(acceptance_path),
|
"status": acceptance["status"],
|
}
|
(CASE_DIR / "batch_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
print(json.dumps(acceptance, ensure_ascii=False))
|
if errors:
|
raise SystemExit(1)
|
|
|
if __name__ == "__main__":
|
main()
|
