| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576 |
- """Validate V5-M4 dead config archive manifests."""
- from __future__ import annotations
- import json
- import sys
- from pathlib import Path
- from typing import Any
- ROOT = Path(__file__).resolve().parents[1]
- ARCHIVE_ROOT = ROOT / "archive/v5_m4_dead_config_archive"
- REQUIRED_FIELDS = {
- "archive_id",
- "source_path",
- "archive_path",
- "archived_at",
- "archived_by",
- "proof_no_runtime_read",
- "restore_instruction",
- "approval_status",
- }
- FORBIDDEN_SOURCE_PREFIXES = {
- "sql/",
- "tests/fixtures/cases/v4_douyin_57663",
- "tests/fixtures/cases/v4_kuaishou_57758",
- }
- FORBIDDEN_SOURCE_PATHS = {
- "tech_documents/数据接口与来源/score_thresholds.json",
- "tech_documents/数据接口与来源/walk_policy.json",
- }
- def validate_archive(root: Path = ARCHIVE_ROOT) -> list[dict[str, Any]]:
- findings: list[dict[str, Any]] = []
- if not root.exists():
- return findings
- for manifest_path in sorted(root.glob("*/manifest.json")):
- manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
- for index, item in enumerate(manifest.get("items", [])):
- _check_item(findings, manifest_path, index, item)
- return findings
- def _check_item(findings: list[dict[str, Any]], manifest_path: Path, index: int, item: dict[str, Any]) -> None:
- missing = sorted(field for field in REQUIRED_FIELDS if not item.get(field))
- if missing:
- _fail(findings, "archive_manifest_required_field", f"{manifest_path}:{index} missing {missing}")
- source_path = str(item.get("source_path") or "")
- archive_path = str(item.get("archive_path") or "")
- source_file = source_path.split("#", 1)[0]
- if source_file in FORBIDDEN_SOURCE_PATHS or any(source_file.startswith(prefix) for prefix in FORBIDDEN_SOURCE_PREFIXES):
- _fail(findings, "archive_forbidden_source", f"{manifest_path}:{index} forbidden source_path {source_path}")
- if archive_path:
- path = ROOT / archive_path
- if not path.exists():
- _fail(findings, "archive_file_missing", f"{manifest_path}:{index} archive_path missing: {archive_path}")
- if not path.is_relative_to(ARCHIVE_ROOT):
- _fail(findings, "archive_path_outside_root", f"{manifest_path}:{index} archive_path outside M4 archive: {archive_path}")
- if item.get("approval_status") != "approved":
- _fail(findings, "archive_approval_status", f"{manifest_path}:{index} approval_status must be approved")
- def _fail(findings: list[dict[str, Any]], check_id: str, message: str) -> None:
- findings.append({"level": "fail", "check_id": check_id, "message": message})
- def main() -> int:
- findings = validate_archive()
- result = {"status": "fail" if findings else "pass", "findings": findings}
- print(json.dumps(result, ensure_ascii=False, indent=2))
- return 1 if findings else 0
- if __name__ == "__main__":
- sys.exit(main())
|