File size: 2,983 Bytes
d8bfe4a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
#!/usr/bin/env python3
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()