#!/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()
|
