#!/usr/bin/env python3 """Apply the independently approved terminal status sync for NEWENERGY BATCH-001. This is an L0 state-only transition. It changes review/document/package state, recomputes receipt/output/artifact hashes affected by that state transition, and creates a terminal acceptance receipt. It does not change source content, evidence text, eligibility, ranking, mappings, or research conclusions. """ from __future__ import annotations import csv import json import re from collections import Counter from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import newenergy_batch001_repair as base import newenergy_batch001_repair004 as r4 TASK_ID = "TASK-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" BATCH_ID = "BATCH-001" RUN_ID = "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001-BATCH-001-001" FINAL_AUDIT = "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH001-EXECUTION-OUTPUT-REPAIR005-REREVIEW-20260806-001" PARENT_STATUS = "COMPLETED_ACCEPTED_AND_CLOSED" ACCEPTED = "ACCEPTED_BY_INDEPENDENT_REVIEW" ARTIFACT_STATUS = "FINAL_ACCEPTED_BY_INDEPENDENT_REVIEW" TOOL_VERSION = "REPAIR-005" PROJECT_ROOT = Path(__file__).resolve().parents[2] INDUSTRY_ROOT = PROJECT_ROOT / "ana-data/cases/新能源案例" CASE_ROOT = INDUSTRY_ROOT / CASE_ID CASE_OUTPUTS = CASE_ROOT / "outputs" CASE_MANIFEST = CASE_ROOT / "manifest" RESULT_ROOT = PROJECT_ROOT / "ana-data/result/新能源案例" / CASE_ID ACCEPTANCE_RECORD = RESULT_ROOT / "acceptance_record.md" ACCEPTANCE_VALIDATION = CASE_MANIFEST / "acceptance_validation_receipt.json" def sha256_file(path: Path) -> str: return r4.sha256_file(path) def stable_accepted_at() -> str: if ACCEPTANCE_RECORD.exists(): text = ACCEPTANCE_RECORD.read_text(encoding="utf-8") match = re.search(r"^- `accepted_at=([^`]+)`$", text, flags=re.M) if match: return match.group(1) return datetime.now(ZoneInfo("Asia/Shanghai")).replace(microsecond=0).isoformat() def read_csv_with_header(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) return list(reader.fieldnames or []), list(reader) def canonical_receipt_hash(row: dict[str, str]) -> str: core = {key: value for key, value in row.items() if key != "receipt_sha256"} payload = json.dumps(core, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return r4.sha256_bytes(payload) def sync_csv_review_statuses() -> dict[str, int]: counters: Counter[str] = Counter() artifact_manifest = INDUSTRY_ROOT / "manifest" / "artifact_manifest.csv" for path in sorted(INDUSTRY_ROOT.rglob("*.csv")): if path.resolve() == artifact_manifest.resolve(): continue headers, rows = read_csv_with_header(path) if "review_status" not in headers: continue for row in rows: row["review_status"] = ACCEPTED if path.name == "human_doc_validation_receipt.csv": row["document_status"] = ACCEPTED if path.name == "output_manifest.csv": row["data_status"] = "FINAL_ACCEPTED" if "receipt_sha256" in headers: row["receipt_sha256"] = canonical_receipt_hash(row) base.write_csv(path, headers, rows) counters["csv_files"] += 1 counters["csv_rows"] += len(rows) counters[path.name] += len(rows) return dict(counters) def replace_required(path: Path, old: str, new: str, *, minimum: int = 1) -> int: text = path.read_text(encoding="utf-8") count = text.count(old) if count < minimum and new not in text: raise RuntimeError(f"required acceptance transition marker missing in {path}: {old!r}") if count: path.write_text(text.replace(old, new), encoding="utf-8", newline="\n") return count def sync_markdown_statuses() -> dict[str, int]: counts: Counter[str] = Counter() for path in sorted(CASE_OUTPUTS.rglob("*.md")): counts["output_status_markers"] += replace_required( path, "> 状态：`DRAFT_FOR_REVIEW`", f"> 状态：`{ACCEPTED}`" ) counts["output_files"] += 1 summary = CASE_OUTPUTS / "summary.md" old = ( "输出仍为 `DRAFT_FOR_REVIEW`。在独立执行/输出审核 PASS 前，不标记完成、" "不回写父级终态、不对外宣称体系成果已经验收。" ) new = ( f"本批执行/输出已由独立审核 `{FINAL_AUDIT}` 以 `PASS / 0 / 0` 接受，" f"父级状态已回写为 `{PARENT_STATUS}`。接受仅覆盖本 case/BATCH/run 的现有公开资料成果，" "不放行下一批、范围扩展、完整覆盖声明、估值、行情、交易、收益、市场反向补漏或核电敏感内容。" ) counts["summary_terminal_paragraph"] += replace_required(summary, old, new) batch_summary = CASE_MANIFEST / "batch_summary.md" counts["candidate_count_fix"] += replace_required( batch_summary, "candidate_state_count=INCLUDED_T1:16,INCLUDED_T2:16,HELD_BY_EVIDENCE_GAP:4479", "candidate_state_count=INCLUDED_T1:16,INCLUDED_T2:16,ELIGIBLE_NOT_SELECTED_BATCH001:40,HELD_BY_EVIDENCE_GAP:4479", ) counts["batch_output_status"] += replace_required( batch_summary, "output_status=DRAFT_FOR_REVIEW", "output_status=FINAL_ACCEPTED" ) counts["batch_review_status"] += replace_required( batch_summary, "review_status=PENDING_REPAIR005_FOCUSED_REREVIEW", f"review_status={ACCEPTED}" ) result_index = RESULT_ROOT / "result_index.md" counts["result_header"] += replace_required( result_index, "> 状态：`DRAFT_FOR_REVIEW`", f"> 状态：`{ACCEPTED}`" ) counts["result_current"] += replace_required( result_index, "当前状态：`DRAFT_FOR_REVIEW`", f"当前状态：`{ACCEPTED}`" ) counts["result_review"] += replace_required( result_index, "执行/输出独立审核：`PENDING`", f"执行/输出独立审核：`PASS`（`{FINAL_AUDIT}`）" ) counts["result_parent"] += replace_required( result_index, "父级终态回写：`NOT_ALLOWED_BEFORE_REVIEW_PASS`", f"父级终态回写：`{PARENT_STATUS}`" ) return dict(counts) def rebuild_output_manifest() -> list[dict[str, str]]: path = CASE_MANIFEST / "output_manifest.csv" headers, rows = read_csv_with_header(path) for row in rows: output = PROJECT_ROOT / row["output_path"] if not output.exists(): raise FileNotFoundError(output) row["output_sha256"] = sha256_file(output) row["data_status"] = "FINAL_ACCEPTED" row["review_status"] = ACCEPTED base.write_csv(path, headers, rows) return rows def write_acceptance_record(accepted_at: str, output_manifest_hash: str) -> None: RESULT_ROOT.mkdir(parents=True, exist_ok=True) text = f"""# {CASE_ID} 正式接受记录 - `task_id={TASK_ID}` - `case_id={CASE_ID}` - `batch_id={BATCH_ID}` - `run_id={RUN_ID}` - `status={ACCEPTED}` - `accepted_at={accepted_at}` - `final_audit={FINAL_AUDIT}` - `audit_result=PASS/0/0/0` - `closed_blockers=B1-R004-A,B1-R004-B` - `parent_status={PARENT_STATUS}` - `accepted_output_manifest=ana-data/cases/新能源案例/{CASE_ID}/manifest/output_manifest.csv` - `accepted_output_manifest_sha256={output_manifest_hash.upper()}` - `governance=OK_PROJECTS1_WARNINGS0` ## 接受范围 本记录接受本 case/BATCH/run 的锂电、光伏、风电、核电四赛道基础行业图谱、16 份子行业正文、三类顶层视图、4,551 条唯一候选映射账本、32 条正式 selected 公司—赛道映射、786/786 份 source/conversion、82 条 evidence fact、110 条结论映射和 24 项 output manifest。候选终态为 `INCLUDED_T1=16 / INCLUDED_T2=16 / ELIGIBLE_NOT_SELECTED_BATCH001=40 / HELD_BY_EVIDENCE_GAP=4479`。 执行/输出独立复审 `{FINAL_AUDIT}` 已关闭 B1-R004-A/B，并给出 `PASS / blocker 0 / non-blocking 0`；此前设计及执行审核的所有 HOLD/PASS 记录和冻结哈希继续保留。审核后仅执行获准的 L0 状态同步，未修改来源内容、证据原文、资格判断、机械排序、结论强度或研究边界。 ## 父级关闭与边界 项目管理员已在 `ana-doc/案例总纲.md` append-only 回写 `{PARENT_STATUS}`。本接受不放行下一批或扩域；继续保持 `primary_region=MAINLAND_CHINA`、全球/境外分账、`coverage_claim=NONE_INITIAL_CANDIDATE_POOL_ONLY`，不形成完整覆盖、估值、行情、交易、收益、市场反向补漏或核电敏感结论。 """ ACCEPTANCE_RECORD.write_text(text, encoding="utf-8", newline="\n") def count_csv(path: Path) -> tuple[int, list[dict[str, str]]]: _, rows = read_csv_with_header(path) return len(rows), rows def write_acceptance_validation(accepted_at: str, output_rows: list[dict[str, str]]) -> dict[str, Any]: candidate_path = INDUSTRY_ROOT / "extracted" / "company_track_candidate_ledger.csv" evidence_path = INDUSTRY_ROOT / "evidence" / "evidence_fact_table.csv" case_map_path = CASE_ROOT / "evidence" / "case_evidence_map.csv" role_path = INDUSTRY_ROOT / "extracted" / "candidate_role_adjudication_receipt.csv" page_path = INDUSTRY_ROOT / "extracted" / "candidate_page_qualification_receipt.csv" source_path = INDUSTRY_ROOT / "manifest" / "source_document.csv" conversion_path = INDUSTRY_ROOT / "manifest" / "conversion_status.csv" _, candidates = count_csv(candidate_path) source_count, _ = count_csv(source_path) conversion_count, _ = count_csv(conversion_path) evidence_count, _ = count_csv(evidence_path) map_count, _ = count_csv(case_map_path) _, roles = count_csv(role_path) _, pages = count_csv(page_path) candidate_states = dict(sorted(Counter(row["candidate_state"] for row in candidates).items())) errors: list[str] = [] expected_states = { "ELIGIBLE_NOT_SELECTED_BATCH001": 40, "HELD_BY_EVIDENCE_GAP": 4479, "INCLUDED_T1": 16, "INCLUDED_T2": 16, } if candidate_states != expected_states: errors.append(f"candidate state mismatch: {candidate_states}") if (source_count, conversion_count, evidence_count, map_count, len(output_rows)) != (786, 786, 82, 110, 24): errors.append( f"accepted package counts mismatch: source={source_count} conversion={conversion_count} " f"evidence={evidence_count} map={map_count} output={len(output_rows)}" ) for label, rows in (("role", roles), ("page", pages)): invalid = sum(row.get("receipt_sha256") != canonical_receipt_hash(row) for row in rows) if invalid: errors.append(f"{label} receipt hash mismatch={invalid}") if any(row["review_status"] != ACCEPTED or row["data_status"] != "FINAL_ACCEPTED" for row in output_rows): errors.append("output manifest terminal status mismatch") output_hash_errors = sum( row["output_sha256"] != sha256_file(PROJECT_ROOT / row["output_path"]) for row in output_rows ) if output_hash_errors: errors.append(f"output hash mismatch={output_hash_errors}") receipt = { "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "accepted_at": accepted_at, "status": ACCEPTED, "final_audit": FINAL_AUDIT, "parent_status": PARENT_STATUS, "transition_type": "L0_REVIEW_STATUS_AND_MANIFEST_HASH_SYNC_ONLY", "candidate_state_distribution": candidate_states, "source_count": source_count, "conversion_count": conversion_count, "evidence_fact_count": evidence_count, "case_map_count": map_count, "output_count": len(output_rows), "role_receipt_count": len(roles), "page_receipt_count": len(pages), "core_hashes": { "candidate_ledger": sha256_file(candidate_path), "evidence_fact_table": sha256_file(evidence_path), "case_evidence_map": sha256_file(case_map_path), "role_adjudication_receipt": sha256_file(role_path), "candidate_page_qualification_receipt": sha256_file(page_path), "output_manifest": sha256_file(CASE_MANIFEST / "output_manifest.csv"), "acceptance_record": sha256_file(ACCEPTANCE_RECORD), }, "historical_frozen_receipt": "manifest/repair005_validation_receipt.json (pre-acceptance audit snapshot; intentionally unchanged)", "validation_errors": errors, "validation_status": "PASS" if not errors else "FAIL", } ACCEPTANCE_VALIDATION.write_text( json.dumps(receipt, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8", newline="\n", ) if errors: raise RuntimeError("acceptance validation failed: " + "; ".join(errors)) return receipt def rebuild_artifact_manifest(output_rows: list[dict[str, str]]) -> list[dict[str, str]]: source_rows = base.read_csv(INDUSTRY_ROOT / "manifest" / "source_document.csv") collected = [row.get("collected_at", "") for row in source_rows if row.get("collected_at")] base.COLLECTED_AT = max(collected, default="2026-08-06T00:00:00+08:00") base.ARTIFACT_TOOL_VERSION = TOOL_VERSION base.ARTIFACT_PARAMETERS_SUMMARY = ( "B1 REPAIR005 independently accepted; terminal L0 review/document/package status sync; " "no source, evidence, eligibility, ranking, mapping, or conclusion change" ) base.ARTIFACT_STATUS = ARTIFACT_STATUS audited_replay_tool = Path(__file__).with_name("newenergy_batch001_repair005.py") return base.build_artifact_manifest(source_rows, output_rows, audited_replay_tool) def validate_artifact_manifest(rows: list[dict[str, str]]) -> dict[str, Any]: audited_replay_tool = Path(__file__).with_name("newenergy_batch001_repair005.py") formal = base.formal_files(audited_replay_tool) by_path = {row["relative_path"]: row for row in rows} errors: list[str] = [] if len(by_path) != len(rows): errors.append("duplicate artifact relative path") if set(by_path) != {base.rel(path) for path in formal}: errors.append("artifact/formal exact-set mismatch") for path in formal: row = by_path.get(base.rel(path)) if row is None: continue if row["file_size"] != str(path.stat().st_size) or row["sha256"] != sha256_file(path): errors.append(f"artifact hash mismatch: {base.rel(path)}") break if any(row["artifact_status"] != ARTIFACT_STATUS or row["tool_version"] != TOOL_VERSION for row in rows): errors.append("artifact terminal status/tool version mismatch") if errors: raise RuntimeError("artifact validation failed: " + "; ".join(errors)) return { "artifact_exact_set": len(rows), "artifact_manifest_sha256": sha256_file(INDUSTRY_ROOT / "manifest" / "artifact_manifest.csv"), "package_exact_set_receipt_sha256": sha256_file(CASE_MANIFEST / "package_exact_set_receipt.md"), } def assert_no_pending_markers() -> None: forbidden = ("DRAFT_FOR_REVIEW", "PENDING_REPAIR005_FOCUSED_REREVIEW", "NOT_ALLOWED_BEFORE_REVIEW_PASS") files = list(CASE_OUTPUTS.rglob("*.md")) + [CASE_MANIFEST / "batch_summary.md", RESULT_ROOT / "result_index.md"] hits = [(path, marker) for path in files for marker in forbidden if marker in path.read_text(encoding="utf-8")] if hits: raise RuntimeError(f"terminal Markdown still contains pending marker: {hits[:3]}") def run() -> dict[str, Any]: accepted_at = stable_accepted_at() csv_sync = sync_csv_review_statuses() markdown_sync = sync_markdown_statuses() output_rows = rebuild_output_manifest() output_manifest_hash = sha256_file(CASE_MANIFEST / "output_manifest.csv") write_acceptance_record(accepted_at, output_manifest_hash) acceptance = write_acceptance_validation(accepted_at, output_rows) artifacts = rebuild_artifact_manifest(output_rows) artifact_validation = validate_artifact_manifest(artifacts) assert_no_pending_markers() return { "status": "ACCEPTED_BY_INDEPENDENT_REVIEW_AND_PARENT_CLOSED", "final_audit": FINAL_AUDIT, "parent_status": PARENT_STATUS, "accepted_at": accepted_at, "candidate_state_distribution": acceptance["candidate_state_distribution"], "csv_sync": csv_sync, "markdown_sync": markdown_sync, "output_count": len(output_rows), "output_manifest_sha256": output_manifest_hash, "acceptance_record_sha256": sha256_file(ACCEPTANCE_RECORD), "acceptance_validation_sha256": sha256_file(ACCEPTANCE_VALIDATION), **artifact_validation, } def main() -> None: print(json.dumps(run(), ensure_ascii=False, sort_keys=True, indent=2)) if __name__ == "__main__": main()