validate_v5_m4_dead_config_archive.py 2.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576
  1. """Validate V5-M4 dead config archive manifests."""
  2. from __future__ import annotations
  3. import json
  4. import sys
  5. from pathlib import Path
  6. from typing import Any
  7. ROOT = Path(__file__).resolve().parents[1]
  8. ARCHIVE_ROOT = ROOT / "archive/v5_m4_dead_config_archive"
  9. REQUIRED_FIELDS = {
  10. "archive_id",
  11. "source_path",
  12. "archive_path",
  13. "archived_at",
  14. "archived_by",
  15. "proof_no_runtime_read",
  16. "restore_instruction",
  17. "approval_status",
  18. }
  19. FORBIDDEN_SOURCE_PREFIXES = {
  20. "sql/",
  21. "tests/fixtures/cases/v4_douyin_57663",
  22. "tests/fixtures/cases/v4_kuaishou_57758",
  23. }
  24. FORBIDDEN_SOURCE_PATHS = {
  25. "tech_documents/数据接口与来源/score_thresholds.json",
  26. "tech_documents/数据接口与来源/walk_policy.json",
  27. }
  28. def validate_archive(root: Path = ARCHIVE_ROOT) -> list[dict[str, Any]]:
  29. findings: list[dict[str, Any]] = []
  30. if not root.exists():
  31. return findings
  32. for manifest_path in sorted(root.glob("*/manifest.json")):
  33. manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
  34. for index, item in enumerate(manifest.get("items", [])):
  35. _check_item(findings, manifest_path, index, item)
  36. return findings
  37. def _check_item(findings: list[dict[str, Any]], manifest_path: Path, index: int, item: dict[str, Any]) -> None:
  38. missing = sorted(field for field in REQUIRED_FIELDS if not item.get(field))
  39. if missing:
  40. _fail(findings, "archive_manifest_required_field", f"{manifest_path}:{index} missing {missing}")
  41. source_path = str(item.get("source_path") or "")
  42. archive_path = str(item.get("archive_path") or "")
  43. source_file = source_path.split("#", 1)[0]
  44. if source_file in FORBIDDEN_SOURCE_PATHS or any(source_file.startswith(prefix) for prefix in FORBIDDEN_SOURCE_PREFIXES):
  45. _fail(findings, "archive_forbidden_source", f"{manifest_path}:{index} forbidden source_path {source_path}")
  46. if archive_path:
  47. path = ROOT / archive_path
  48. if not path.exists():
  49. _fail(findings, "archive_file_missing", f"{manifest_path}:{index} archive_path missing: {archive_path}")
  50. if not path.is_relative_to(ARCHIVE_ROOT):
  51. _fail(findings, "archive_path_outside_root", f"{manifest_path}:{index} archive_path outside M4 archive: {archive_path}")
  52. if item.get("approval_status") != "approved":
  53. _fail(findings, "archive_approval_status", f"{manifest_path}:{index} approval_status must be approved")
  54. def _fail(findings: list[dict[str, Any]], check_id: str, message: str) -> None:
  55. findings.append({"level": "fail", "check_id": check_id, "message": message})
  56. def main() -> int:
  57. findings = validate_archive()
  58. result = {"status": "fail" if findings else "pass", "findings": findings}
  59. print(json.dumps(result, ensure_ascii=False, indent=2))
  60. return 1 if findings else 0
  61. if __name__ == "__main__":
  62. sys.exit(main())