| |
| from __future__ import annotations |
|
|
| import ast |
| import csv |
| import hashlib |
| import importlib.util |
| import json |
| import re |
| import sys |
| from pathlib import Path |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def digest(path: Path) -> str: |
| h = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| h.update(chunk) |
| return h.hexdigest() |
|
|
|
|
| def main() -> None: |
| errors = [] |
| checked_python = 0 |
| private_roots = [ |
| "ho" + "me", "projects" + "_vol", "hpc" + "_stor", |
| "apd" + "cephfs", "Wo" + "rk20", "Wo" + "rk21", "CD" + "Share", |
| ] |
| identity = re.compile( |
| r"(?i)(/(?:" + "|".join(private_roots) + r")/|" |
| r"https?://(?:10|172\.(?:1[6-9]|2[0-9]|3[01])|192\.168)(?:\.[0-9]{1,3}){2,3})" |
| ) |
| for path in ROOT.rglob("*.py"): |
| try: |
| ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) |
| checked_python += 1 |
| except SyntaxError as exc: |
| errors.append(f"syntax: {path.relative_to(ROOT)}: {exc}") |
|
|
| for path in ROOT.rglob("*.json"): |
| try: |
| json.loads(path.read_text(encoding="utf-8-sig")) |
| except json.JSONDecodeError as exc: |
| errors.append(f"json: {path.relative_to(ROOT)}: {exc}") |
|
|
| for path in ROOT.rglob("*"): |
| if path.is_file() and path.suffix.lower() in {".py", ".sh", ".sbatch", ".json", ".jsonl", ".yaml", ".yml", ".md", ".txt"}: |
| if identity.search(path.read_text(encoding="utf-8-sig", errors="replace")): |
| errors.append(f"identity: {path.relative_to(ROOT)}") |
|
|
| sys.path.insert(0, str(ROOT)) |
| try: |
| __import__("calibration_agent") |
| if importlib.util.find_spec("calibration_agent.supervisor") is None: |
| errors.append("import: calibration_agent.supervisor package not found") |
| except Exception as exc: |
| errors.append(f"import: calibration_agent: {exc}") |
|
|
| manifest = ROOT / "MANIFEST.tsv" |
| manifest_paths = set() |
| with manifest.open(encoding="utf-8", newline="") as handle: |
| for row in csv.DictReader(handle, delimiter="\t"): |
| manifest_paths.add(row["path"]) |
| path = ROOT / row["path"] |
| if not path.is_file(): |
| errors.append(f"missing: {row['path']}") |
| elif digest(path) != row["sha256"]: |
| errors.append(f"hash: {row['path']}") |
| actual_paths = { |
| path.relative_to(ROOT).as_posix() |
| for path in ROOT.rglob("*") |
| if path.is_file() and path.name != "MANIFEST.tsv" and "__pycache__" not in path.parts and path.suffix != ".pyc" |
| } |
| if actual_paths != manifest_paths: |
| errors.append("manifest: file set differs from archive tree") |
|
|
| if errors: |
| raise SystemExit("Validation failed:\n" + "\n".join(errors)) |
| print(f"OK: {checked_python} Python files parsed and manifest hashes verified") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
