MB-X Bilibili Pipeline
7 days ago 5c75bc6b3fe4eda9b1210574230d22249ecf949f
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#!/usr/bin/env python3
"""Read-only verifier for the NEWENERGY BATCH-002 acceptance sync."""
 
from __future__ import annotations
 
import csv
import hashlib
import json
from pathlib import Path
 
 
ROOT = Path(__file__).resolve().parents[2]
INDUSTRY = ROOT / "ana-data/cases/新能源案例"
CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002"
CASE = INDUSTRY / CASE_ID
RESULT = ROOT / "ana-data/result/新能源案例" / CASE_ID
ACCEPTED = "ACCEPTED_BY_INDEPENDENT_REVIEW"
FINAL_ARTIFACT = "FINAL_ACCEPTED_BY_INDEPENDENT_REVIEW"
FINAL_AUDIT = "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-EXECUTION-OUTPUT-REPAIR004-REREVIEW-20260807-001"
 
 
def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest().upper()
 
 
def rows(path: Path) -> list[dict[str, str]]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))
 
 
output_manifest = CASE / "manifest/output_manifest.csv"
outputs = rows(output_manifest)
assert len(outputs) == 42
assert sum(row["output_status"] == "FINAL_ACCEPTED" for row in outputs) == 40
assert sum(row["output_status"] == "NOT_APPLICABLE" for row in outputs) == 2
assert all(row["review_status"] == ACCEPTED for row in outputs)
for row in outputs:
    if row["applicability"] == "APPLICABLE_FILE_OUTPUT":
        path = ROOT / row["output_path"]
        assert path.is_file()
        assert path.stat().st_size == int(row["file_size"])
        assert sha256(path) == row["sha256"].upper()
 
artifact_manifest = INDUSTRY / "manifest/artifact_manifest_BATCH002.csv"
artifacts = rows(artifact_manifest)
assert len(artifacts) == 85
assert len({row["relative_path"] for row in artifacts}) == 85
for row in artifacts:
    path = ROOT / row["relative_path"]
    assert path.is_file()
    assert path.stat().st_size == int(row["file_size"])
    assert sha256(path) == row["sha256"].upper()
    assert row["artifact_status"] == FINAL_ARTIFACT
    assert row["review_status"] == ACCEPTED
 
for path in (
    CASE / "outputs",
    CASE / "evidence",
    CASE / "manifest",
    RESULT,
):
    for item in path.rglob("*"):
        if item.is_file() and item.suffix.lower() in {".md", ".csv", ".json"}:
            text = item.read_text(encoding="utf-8-sig")
            assert "\ufffd" not in text
            assert "DRAFT_FOR_REVIEW" not in text
 
registry = rows(INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv")
assert len(registry) == 33
for row in registry:
    shard = ROOT / row["shard_path"]
    assert sha256(shard) == row["sha256"].upper()
    assert len(rows(shard)) == int(row["row_count"])
    assert row["review_status"] == ACCEPTED
    if row["batch_id"] == "BATCH-002":
        assert row["immutable"] == "YES_ACCEPTED_IMMUTABLE"
 
current = rows(INDUSTRY / "manifest/current_output_manifest.csv")
assert len(current) == 25
assert sha256(INDUSTRY / "manifest/current_output_manifest.csv") == "A2BEAF2CD0D3955ACF9DE8C6769D4BC9A3A7A1094FDD28C8447B66FD09372EB7"
assert not any("BATCH-002" in row.get("source_path", "") or CASE_ID in row.get("source_path", "") for row in current)
 
acceptance = json.loads((CASE / "manifest/acceptance_validation_receipt.json").read_text(encoding="utf-8"))
assert acceptance["status"] == ACCEPTED
assert acceptance["final_audit"] == FINAL_AUDIT
assert acceptance["validation_status"] == "PASS"
record = (RESULT / "acceptance_record.md").read_text(encoding="utf-8")
assert f"final_audit={FINAL_AUDIT}" in record
 
print(json.dumps({
    "status": "PASS_READ_ONLY_ACCEPTANCE_VERIFIER",
    "outputs": "40_FINAL_ACCEPTED+2_NOT_APPLICABLE",
    "output_manifest_rows": len(outputs),
    "artifact_exact_set": len(artifacts),
    "registry": "33/33_HASH_AND_ROW_COUNT_MATCH",
    "batch002_current_release_paths": 0,
    "output_manifest_sha256": sha256(output_manifest),
    "artifact_manifest_sha256": sha256(artifact_manifest),
    "acceptance_record_sha256": sha256(RESULT / "acceptance_record.md"),
    "acceptance_validation_sha256": sha256(CASE / "manifest/acceptance_validation_receipt.json"),
}, ensure_ascii=False, sort_keys=True))