#!/usr/bin/env python3
|
"""Apply the independently approved L0 acceptance sync for NEWENERGY BATCH-002.
|
|
The script is deliberately scoped to the frozen BATCH-002 artifact set. It
|
updates review/package state, recalculates hashes affected by those state-only
|
changes, and creates acceptance receipts. It does not alter research facts,
|
evidence wording, candidate selection, the accepted BATCH-001 package, or the
|
industry-root current release.
|
"""
|
|
from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import re
|
from copy import deepcopy
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
from zoneinfo import ZoneInfo
|
|
|
TASK_ID = "TASK-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-20260806-001"
|
CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002"
|
BATCH_ID = "BATCH-002"
|
RUN_ID = "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002-BATCH-002-001"
|
FINAL_AUDIT = "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-EXECUTION-OUTPUT-REPAIR004-REREVIEW-20260807-001"
|
ACCEPTED = "ACCEPTED_BY_INDEPENDENT_REVIEW"
|
FINAL_OUTPUT = "FINAL_ACCEPTED"
|
FINAL_ARTIFACT = "FINAL_ACCEPTED_BY_INDEPENDENT_REVIEW"
|
PARENT_STATUS = "PENDING_PROJECT_ADMIN_APPEND_ONLY_TERMINAL_SYNC"
|
|
ROOT = Path(__file__).resolve().parents[2]
|
INDUSTRY = ROOT / "ana-data/cases/新能源案例"
|
CASE = INDUSTRY / CASE_ID
|
RESULT = ROOT / "ana-data/result/新能源案例" / CASE_ID
|
ARTIFACT_MANIFEST = INDUSTRY / "manifest/artifact_manifest_BATCH002.csv"
|
OUTPUT_MANIFEST = CASE / "manifest/output_manifest.csv"
|
EXECUTION_RECEIPT = CASE / "manifest/execution_validation_receipt.json"
|
ACCEPTANCE_RECEIPT = CASE / "manifest/acceptance_validation_receipt.json"
|
ACCEPTANCE_RECORD = RESULT / "acceptance_record.md"
|
CURRENT_STABLE_MANIFEST = INDUSTRY / "manifest/current_output_manifest.csv"
|
|
AUDITED_HASHES = {
|
ROOT / "ana-data/tools/newenergy_batch002_build.py": "537C8826CE2F49DEFA188E1E308643E816E5D3B3C8E34A6BCB1B02883FB5D389",
|
INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv": "57F232EB9FF61893FE6D6130684A0BED01E9D2B6B6E76719691A9E5523DAB954",
|
INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv": "A332F0C8F11AF8DDA9E5D77D389EF34502D2ABDB9D4B3663F6CA36D8A2D98055",
|
INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv": "A25D32490C7DEFA14024B1FA46C6DA3CDE323A5CDBEF36D3C84D1DF6ECA2C45D",
|
INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv": "D656AB6C5B5AF4FE2632FAC8F54C6239C4894D5DA81D150588FE67A3B52DAFAE",
|
ARTIFACT_MANIFEST: "A3D9933AFBB4D2066290ACD5679A9AACE986985148BB2668FA430E38871C2B25",
|
OUTPUT_MANIFEST: "B074F15BFF21EB7B3872369B26A0E20E245782BB8C852BAA2D2A0994660174DE",
|
CASE / "evidence/case_evidence_map.csv": "27666694FA5A38E29AB518C19A970E20F5D7CE091DCAD15C0EB29CFA9CC7EF70",
|
EXECUTION_RECEIPT: "41D663B73702CFAAD0C22AF59514062484EFB22FA3AE59DB61CDBE11373F2299",
|
}
|
|
IMMUTABLE_B001 = {
|
INDUSTRY / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001/manifest/output_manifest.csv": "D0DC978D558CA56C9D32B9DC0F76489EDB0F643DB1B482585EF2D96DEE3DF6C4",
|
ROOT / "ana-data/result/新能源案例/ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001/acceptance_record.md": "448E67471C53E1E59011A1ADBB44D2ED0BDA0068084D843251E2CCCD4A407CFA",
|
CURRENT_STABLE_MANIFEST: "A2BEAF2CD0D3955ACF9DE8C6769D4BC9A3A7A1094FDD28C8447B66FD09372EB7",
|
}
|
|
|
def sha256(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as handle:
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest().upper()
|
|
|
def project_relative(path: Path) -> str:
|
return path.resolve().relative_to(ROOT.resolve()).as_posix()
|
|
|
def read_csv(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 write_csv(path: Path, headers: list[str], rows: list[dict[str, Any]]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=headers, extrasaction="ignore", lineterminator="\n")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def write_json(path: Path, value: Any) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8", newline="\n")
|
|
|
def accepted_at() -> str:
|
if ACCEPTANCE_RECORD.exists():
|
match = re.search(r"^- `accepted_at=([^`]+)`$", ACCEPTANCE_RECORD.read_text(encoding="utf-8"), flags=re.M)
|
if match:
|
return match.group(1)
|
return datetime.now(ZoneInfo("Asia/Shanghai")).replace(microsecond=0).isoformat()
|
|
|
def audit_gate() -> None:
|
report = (ROOT / "ana-doc/新能源案例/案例审计报告.md").read_text(encoding="utf-8")
|
marker = f"## 2026-08-07 `{FINAL_AUDIT}`"
|
start = report.find(marker)
|
if start < 0:
|
raise RuntimeError("final audit section missing")
|
tail = report[start:]
|
end = tail.find("\n## ", 1)
|
section = tail if end < 0 else tail[:end]
|
required = (
|
"总体结论：`PASS / REPAIR004 聚焦复审通过`",
|
"允许分析员按项目规范把本批 outputs、canonical shard、manifest 和 result 入口",
|
)
|
if any(token not in section for token in required):
|
raise RuntimeError("final audit does not release the acceptance sync")
|
|
|
def preflight() -> tuple[list[str], list[dict[str, str]]]:
|
audit_gate()
|
for path, expected in {**AUDITED_HASHES, **IMMUTABLE_B001}.items():
|
if not path.is_file() or sha256(path) != expected:
|
raise RuntimeError(f"preflight hash drift: {project_relative(path)}")
|
headers, artifacts = read_csv(ARTIFACT_MANIFEST)
|
if len(artifacts) != 83 or len({row["relative_path"] for row in artifacts}) != 83:
|
raise RuntimeError("audited artifact exact-set is not 83 unique paths")
|
for row in artifacts:
|
path = ROOT / row["relative_path"]
|
if not path.is_file() or path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper():
|
raise RuntimeError(f"audited artifact mismatch: {row['relative_path']}")
|
stable_rows = read_csv(CURRENT_STABLE_MANIFEST)[1]
|
if any("BATCH-002" in row.get("source_path", "") or CASE_ID in row.get("source_path", "") for row in stable_rows):
|
raise RuntimeError("BATCH-002 already entered current industry release")
|
return headers, artifacts
|
|
|
def replace_required(path: Path, old: str, new: str) -> int:
|
text = path.read_text(encoding="utf-8")
|
count = text.count(old)
|
if count:
|
path.write_text(text.replace(old, new), encoding="utf-8", newline="\n")
|
elif new not in text:
|
raise RuntimeError(f"required marker missing: {project_relative(path)} :: {old}")
|
return count
|
|
|
def sync_markdown(artifacts: list[dict[str, str]]) -> dict[str, int]:
|
md_paths = [ROOT / row["relative_path"] for row in artifacts if row["file_ext"] == ".md"]
|
if len(md_paths) != 42:
|
raise RuntimeError(f"audited markdown count changed: {len(md_paths)}/42")
|
counts = {"markdown_files": len(md_paths), "review_markers": 0, "terminal_sentences": 0}
|
for path in md_paths:
|
counts["review_markers"] += replace_required(
|
path, "- review_status: `DRAFT_FOR_REVIEW`", f"- review_status: `{ACCEPTED}`"
|
)
|
|
terminal = {
|
CASE / "outputs/summary.md": (
|
"- 输出状态仍为 `DRAFT_FOR_REVIEW`，等待独立执行/输出审核。",
|
f"- 本批执行与输出已由独立审核 `{FINAL_AUDIT}` 以 `PASS / 0 / 0` 接受；接受不扩展研究边界或完整覆盖声明。",
|
),
|
CASE / "outputs/新能源报告索引.md": (
|
"当前全部产物为 `DRAFT_FOR_REVIEW`，独立执行/输出审核通过前不得标记完成或对外交付。",
|
f"当前全部产物已由独立审核 `{FINAL_AUDIT}` 接受；本入口保留逐批审计血缘，行业默认阅读入口仍由行业根原子 promotion 单独控制。",
|
),
|
RESULT / "result_index.md": (
|
"当前状态：`DRAFT_FOR_REVIEW`；独立执行/输出审核通过前不构成正式交付。",
|
f"当前状态：`{ACCEPTED}`；接受审计：`{FINAL_AUDIT}=PASS`。本逐批入口仅保留审计血缘，不替代行业根当前成果入口。",
|
),
|
CASE / "manifest/batch_summary.md": (
|
"执行/输出独立审核通过前不得回写父级终态或对外交付。",
|
f"执行/输出独立审核 `{FINAL_AUDIT}` 已通过；父级 append-only 终态由项目管理员同步，行业根发布仍走独立 candidate/promotion。",
|
),
|
}
|
for path, (old, new) in terminal.items():
|
counts["terminal_sentences"] += replace_required(path, old, new)
|
|
package = CASE / "manifest/package_exact_set_receipt.md"
|
counts["package_count"] = replace_required(
|
package, "- expected_artifact_exact_set_count: `83`", "- expected_artifact_exact_set_count: `85`"
|
)
|
counts["package_audit"] = replace_required(
|
package,
|
"exact-set 包括本批新行业 shard、新 raw/converted、supplement 查询回执、案例 outputs/manifest/evidence、result 入口和获批 replay tool；不包括空 img/tmp 目录，也不复制前批 artifacts。",
|
f"exact-set 包括本批新行业 shard、新 raw/converted、supplement 查询回执、案例 outputs/manifest/evidence、result 入口、接受记录、接受验证回执和获批 replay tool；独立接受审计为 `{FINAL_AUDIT}=PASS`，不包括空 img/tmp 目录，也不复制前批 artifacts。",
|
)
|
leftovers = [(project_relative(path), path.read_text(encoding="utf-8").count("DRAFT_FOR_REVIEW")) for path in md_paths]
|
leftovers = [item for item in leftovers if item[1]]
|
if leftovers:
|
raise RuntimeError(f"draft markdown markers remain: {leftovers[:3]}")
|
return counts
|
|
|
def sync_json_review_status(value: Any) -> Any:
|
if isinstance(value, dict):
|
return {
|
key: ACCEPTED if key == "review_status" and item == "DRAFT_FOR_REVIEW" else sync_json_review_status(item)
|
for key, item in value.items()
|
}
|
if isinstance(value, list):
|
return [sync_json_review_status(item) for item in value]
|
return value
|
|
|
def sync_structured_statuses(artifacts: list[dict[str, str]]) -> dict[str, int]:
|
counts = {"csv_files": 0, "csv_rows": 0, "json_files": 0, "review_rows": 0, "sample_rows": 0}
|
for artifact in artifacts:
|
path = ROOT / artifact["relative_path"]
|
if path == ARTIFACT_MANIFEST:
|
continue
|
if artifact["file_ext"] == ".csv":
|
headers, rows = read_csv(path)
|
changed = False
|
for row in rows:
|
if "review_status" in headers and row.get("review_status") == "DRAFT_FOR_REVIEW":
|
row["review_status"] = ACCEPTED
|
counts["review_rows"] += 1
|
changed = True
|
if "sample_status" in headers and row.get("sample_status") == "READY_FOR_INDEPENDENT_REVIEW":
|
row["sample_status"] = ACCEPTED
|
counts["sample_rows"] += 1
|
changed = True
|
if changed:
|
write_csv(path, headers, rows)
|
counts["csv_files"] += 1
|
counts["csv_rows"] += len(rows)
|
elif artifact["file_ext"] == ".json":
|
value = json.loads(path.read_text(encoding="utf-8"))
|
updated = sync_json_review_status(value)
|
if updated != value:
|
write_json(path, updated)
|
counts["json_files"] += 1
|
return counts
|
|
|
def update_output_manifest() -> list[dict[str, str]]:
|
headers, rows = read_csv(OUTPUT_MANIFEST)
|
if len(rows) != 42:
|
raise RuntimeError("output manifest is not 42 rows")
|
applicable = 0
|
not_applicable = 0
|
for row in rows:
|
row["review_status"] = ACCEPTED
|
if row["applicability"] == "APPLICABLE_FILE_OUTPUT":
|
output = ROOT / row["output_path"]
|
if not output.is_file():
|
raise FileNotFoundError(output)
|
row["file_size"] = str(output.stat().st_size)
|
row["sha256"] = sha256(output)
|
row["output_status"] = FINAL_OUTPUT
|
applicable += 1
|
elif row["output_status"] == "NOT_APPLICABLE":
|
not_applicable += 1
|
else:
|
raise RuntimeError(f"unexpected output applicability/status: {row['output_item_id']}")
|
if (applicable, not_applicable) != (40, 2):
|
raise RuntimeError(f"output distribution changed: {applicable}/{not_applicable}")
|
write_csv(OUTPUT_MANIFEST, headers, rows)
|
return rows
|
|
|
def update_projection_receipts() -> None:
|
adapter = INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv"
|
period = INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv"
|
projection = INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv"
|
headers, rows = read_csv(projection)
|
for row in rows:
|
shard = ROOT / row["batch002_shard_path"]
|
row["batch002_shard_sha256"] = sha256(shard)
|
row["adapter_contract_sha256"] = sha256(adapter)
|
if row.get("period_projection_receipt_path") not in ("", "NOT_APPLICABLE"):
|
row["period_projection_receipt_sha256"] = sha256(period)
|
row["review_status"] = ACCEPTED
|
write_csv(projection, headers, rows)
|
|
|
def update_registry() -> None:
|
path = INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv"
|
headers, rows = read_csv(path)
|
for row in rows:
|
shard = ROOT / row["shard_path"]
|
actual_rows = read_csv(shard)[1]
|
row["row_count"] = str(len(actual_rows))
|
row["sha256"] = sha256(shard)
|
row["review_status"] = ACCEPTED
|
if row["batch_id"] == BATCH_ID:
|
row["immutable"] = "YES_ACCEPTED_IMMUTABLE"
|
write_csv(path, headers, rows)
|
|
|
def update_execution_receipt(at: str) -> None:
|
receipt = json.loads(EXECUTION_RECEIPT.read_text(encoding="utf-8"))
|
receipt["outputs_status"] = FINAL_OUTPUT
|
receipt["review_status"] = ACCEPTED
|
receipt["final_audit"] = FINAL_AUDIT
|
receipt["acceptance_transition"] = "L0_STATUS_AND_AFFECTED_HASH_SYNC_ONLY"
|
receipt["accepted_at"] = at
|
receipt["parent_terminal_status"] = PARENT_STATUS
|
receipt["legacy_adapter_contract_sha256"] = sha256(INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv")
|
receipt["legacy_period_projection_receipt_sha256"] = sha256(INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv")
|
receipt["legacy_projection_validation_sha256"] = sha256(INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv")
|
write_json(EXECUTION_RECEIPT, receipt)
|
|
|
def write_acceptance_record(at: str, output_rows: list[dict[str, str]]) -> None:
|
text = f"""# {CASE_ID} 正式接受记录
|
|
- `task_id={TASK_ID}`
|
- `case_id={CASE_ID}`
|
- `batch_id={BATCH_ID}`
|
- `run_id={RUN_ID}`
|
- `status={ACCEPTED}`
|
- `accepted_at={at}`
|
- `final_audit={FINAL_AUDIT}`
|
- `audit_result=PASS/0/0`
|
- `parent_status={PARENT_STATUS}`
|
- `accepted_output_manifest=ana-data/cases/新能源案例/{CASE_ID}/manifest/output_manifest.csv`
|
- `accepted_output_manifest_sha256={sha256(OUTPUT_MANIFEST)}`
|
- `applicable_outputs={sum(row['output_status'] == FINAL_OUTPUT for row in output_rows)}`
|
- `conditional_not_applicable_outputs={sum(row['output_status'] == 'NOT_APPLICABLE' for row in output_rows)}`
|
|
## 接受范围
|
|
本记录接受本 case/BATCH/run 的 16 个新能源子行业深化节点、32 份节点正文、三类顶层视图、9 条增量公司—赛道映射、六组 legacy-to-V1 只读投影、期间语义回执和 40 份实物输出。7 个证据不足节点继续保留 GAP，2 个条件逻辑数据集继续 `NOT_APPLICABLE`。
|
|
独立复审 `{FINAL_AUDIT}` 已给出 `PASS / blocker 0 / non-blocking 0`。审核后只执行状态与受影响 hash 的 L0 同步；没有改变来源、证据原文、候选选择、结论强度、BATCH-001 接受包或行业根当前 release。
|
|
## 边界
|
|
本接受不放行下一批、完整覆盖、估值、行情、交易、收益、市场反向补漏、核电敏感内容或数据库写入。逐批 result 入口只保留审计血缘；若要进入行业默认阅读入口，必须另建行业根 candidate，并通过既有新能源审核与原子 promotion。
|
"""
|
ACCEPTANCE_RECORD.parent.mkdir(parents=True, exist_ok=True)
|
ACCEPTANCE_RECORD.write_text(text, encoding="utf-8", newline="\n")
|
|
|
def write_acceptance_validation(at: str, output_rows: list[dict[str, str]], sync_counts: dict[str, Any]) -> None:
|
validation = {
|
"schema_version": "NEWENERGY_BATCH002_ACCEPTANCE_VALIDATION_V1",
|
"task_id": TASK_ID,
|
"case_id": CASE_ID,
|
"batch_id": BATCH_ID,
|
"run_id": RUN_ID,
|
"status": ACCEPTED,
|
"accepted_at": at,
|
"final_audit": FINAL_AUDIT,
|
"transition_type": "L0_REVIEW_STATUS_AND_AFFECTED_HASH_SYNC_ONLY",
|
"outputs": {
|
"manifest_rows": len(output_rows),
|
"final_accepted": sum(row["output_status"] == FINAL_OUTPUT for row in output_rows),
|
"not_applicable": sum(row["output_status"] == "NOT_APPLICABLE" for row in output_rows),
|
"manifest_sha256": sha256(OUTPUT_MANIFEST),
|
},
|
"accepted_scope": {"subindustry_nodes": 16, "node_documents": 32, "incremental_company_track_mappings": 9, "gap_nodes": 7},
|
"sync_counts": sync_counts,
|
"core_hashes": {
|
"case_evidence_map": sha256(CASE / "evidence/case_evidence_map.csv"),
|
"canonical_registry": sha256(INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv"),
|
"projection_validation": sha256(INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv"),
|
"period_projection_receipt": sha256(INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv"),
|
"execution_validation": sha256(EXECUTION_RECEIPT),
|
"acceptance_record": sha256(ACCEPTANCE_RECORD),
|
},
|
"immutable_b001_hash_match": len(IMMUTABLE_B001),
|
"industry_current_release_changed": False,
|
"parent_terminal_status": PARENT_STATUS,
|
"validation_status": "PASS",
|
}
|
write_json(ACCEPTANCE_RECEIPT, validation)
|
|
|
def rebuild_artifact_manifest(headers: list[str], old_rows: list[dict[str, str]], at: str) -> list[dict[str, str]]:
|
by_path = {row["relative_path"]: deepcopy(row) for row in old_rows}
|
additions = [ACCEPTANCE_RECEIPT, ACCEPTANCE_RECORD]
|
for index, path in enumerate(additions, start=84):
|
relative = project_relative(path)
|
by_path[relative] = {
|
"artifact_id": f"NEB2-ART-{index:04d}",
|
"task_id": TASK_ID,
|
"case_id": CASE_ID,
|
"batch_id": BATCH_ID,
|
"run_id": RUN_ID,
|
"artifact_type": "RECEIPT_OR_VALIDATION" if path.suffix == ".json" else "HUMAN_READABLE_OUTPUT_OR_RECEIPT",
|
"industry_case": "新能源案例",
|
"industry_id": "IND-NEWENERGY",
|
"subindustry_id": "",
|
"company_id": "",
|
"logical_path": relative,
|
"relative_path": relative,
|
"absolute_path": str(path.resolve()),
|
"file_name": path.name,
|
"file_ext": path.suffix.lower(),
|
"file_size": "",
|
"sha256": "",
|
"source_doc_id": "",
|
"source_url": "",
|
"source_collected_at": "",
|
"raw_pool_path": "",
|
"source_file_name": "",
|
"detected_type": path.suffix.lower().lstrip("."),
|
"archive_file_name": path.name,
|
"extension_added_by_archive_flag": "NO",
|
"extension_mismatch_flag": "NO",
|
"created_at": at,
|
"created_by": "case_analysis.analyst.new_energy",
|
"tool_or_method": "newenergy_batch002_accept.py",
|
"tool_version": "BATCH002-ACCEPT-V1",
|
"parameters_summary": "independent PASS terminal L0 status/hash sync only",
|
"source_snapshot_id": "BATCH002_ACCEPTANCE",
|
"artifact_status": FINAL_ARTIFACT,
|
"sensitivity_screen": "LEGAL_PUBLIC_CIVIL_NUCLEAR_HIGH_LEVEL_ONLY",
|
"schema_version": "NEWENERGY_EXTENSION_V1",
|
"review_status": ACCEPTED,
|
}
|
if len(by_path) != 85:
|
raise RuntimeError(f"post-acceptance artifact exact-set is not 85: {len(by_path)}")
|
rows: list[dict[str, str]] = []
|
for relative in sorted(by_path):
|
row = by_path[relative]
|
path = ROOT / relative
|
if not path.is_file():
|
raise FileNotFoundError(path)
|
row["file_size"] = str(path.stat().st_size)
|
row["sha256"] = sha256(path)
|
row["artifact_status"] = FINAL_ARTIFACT
|
row["review_status"] = ACCEPTED
|
if row.get("parameters_summary") == "exact released batch/run; immutable B001 references; DRAFT outputs":
|
row["parameters_summary"] = "exact released batch/run; immutable B001 references; independently accepted outputs"
|
rows.append(row)
|
write_csv(ARTIFACT_MANIFEST, headers, rows)
|
return rows
|
|
|
def verify_accepted() -> dict[str, Any]:
|
for path, expected in IMMUTABLE_B001.items():
|
if sha256(path) != expected:
|
raise RuntimeError(f"immutable BATCH-001/current release drift: {project_relative(path)}")
|
output_rows = read_csv(OUTPUT_MANIFEST)[1]
|
if len(output_rows) != 42:
|
raise RuntimeError("accepted output manifest row count mismatch")
|
if sum(row["output_status"] == FINAL_OUTPUT for row in output_rows) != 40:
|
raise RuntimeError("accepted applicable output count mismatch")
|
if sum(row["output_status"] == "NOT_APPLICABLE" for row in output_rows) != 2:
|
raise RuntimeError("accepted N/A output count mismatch")
|
if any(row["review_status"] != ACCEPTED for row in output_rows):
|
raise RuntimeError("output review status mismatch")
|
for row in output_rows:
|
if row["applicability"] == "APPLICABLE_FILE_OUTPUT":
|
path = ROOT / row["output_path"]
|
if path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper():
|
raise RuntimeError(f"accepted output hash mismatch: {row['output_path']}")
|
|
artifact_rows = read_csv(ARTIFACT_MANIFEST)[1]
|
if len(artifact_rows) != 85 or len({row["relative_path"] for row in artifact_rows}) != 85:
|
raise RuntimeError("accepted artifact exact-set mismatch")
|
for row in artifact_rows:
|
path = ROOT / row["relative_path"]
|
if not path.is_file() or path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper():
|
raise RuntimeError(f"accepted artifact hash mismatch: {row['relative_path']}")
|
if row["artifact_status"] != FINAL_ARTIFACT or row["review_status"] != ACCEPTED:
|
raise RuntimeError(f"accepted artifact status mismatch: {row['relative_path']}")
|
|
stable_rows = read_csv(CURRENT_STABLE_MANIFEST)[1]
|
batch2_current = sum("BATCH-002" in row.get("source_path", "") or CASE_ID in row.get("source_path", "") for row in stable_rows)
|
if batch2_current:
|
raise RuntimeError("BATCH-002 polluted the current BATCH-001 industry release")
|
if not ACCEPTANCE_RECORD.is_file() or not ACCEPTANCE_RECEIPT.is_file():
|
raise RuntimeError("acceptance records missing")
|
return {
|
"status": ACCEPTED,
|
"final_audit": FINAL_AUDIT,
|
"output_manifest_rows": len(output_rows),
|
"final_accepted_outputs": 40,
|
"not_applicable_outputs": 2,
|
"artifact_exact_set": len(artifact_rows),
|
"output_manifest_sha256": sha256(OUTPUT_MANIFEST),
|
"artifact_manifest_sha256": sha256(ARTIFACT_MANIFEST),
|
"acceptance_record_sha256": sha256(ACCEPTANCE_RECORD),
|
"acceptance_validation_sha256": sha256(ACCEPTANCE_RECEIPT),
|
"immutable_b001_and_current_release_hash_match": len(IMMUTABLE_B001),
|
"batch002_current_release_paths": batch2_current,
|
"parent_status": PARENT_STATUS,
|
}
|
|
|
def run() -> dict[str, Any]:
|
if ACCEPTANCE_RECORD.exists():
|
result = verify_accepted()
|
result["idempotent"] = True
|
return result
|
headers, artifacts = preflight()
|
at = accepted_at()
|
markdown_counts = sync_markdown(artifacts)
|
structured_counts = sync_structured_statuses(artifacts)
|
output_rows = update_output_manifest()
|
update_projection_receipts()
|
update_registry()
|
update_execution_receipt(at)
|
write_acceptance_record(at, output_rows)
|
write_acceptance_validation(at, output_rows, {"markdown": markdown_counts, "structured": structured_counts})
|
rebuild_artifact_manifest(headers, artifacts, at)
|
result = verify_accepted()
|
result["idempotent"] = False
|
result["accepted_at"] = at
|
return result
|
|
|
if __name__ == "__main__":
|
print(json.dumps(run(), ensure_ascii=False, sort_keys=True, indent=2))
|
