Spaces:
Sleeping
Sleeping
File size: 4,412 Bytes
f6712ff | 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
def sha256_file(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def verify_model_entries(v3_root: Path, manifest: dict) -> list[str]:
errors: list[str] = []
models = manifest.get("models", {})
for model_id, meta in models.items():
rel = meta.get("file")
sha = meta.get("sha256")
if rel is None:
# virtual ensemble model
continue
p = v3_root / rel
if not p.exists():
errors.append(f"model file missing: {model_id} -> {rel}")
continue
if not sha:
errors.append(f"sha256 missing for model: {model_id}")
continue
if sha != sha256_file(p):
errors.append(f"sha256 mismatch for model: {model_id} -> {rel}")
return errors
def verify_sections(v3_root: Path, checksums: dict, section: str) -> list[str]:
errors: list[str] = []
entries = checksums.get(section, {})
if not isinstance(entries, dict):
return [f"checksums.{section} must be an object"]
for rel, expected in entries.items():
p = v3_root / rel
if not p.exists():
errors.append(f"checksums.{section} missing file: {rel}")
continue
actual = sha256_file(p)
if actual != expected:
errors.append(f"checksums.{section} mismatch: {rel}")
return errors
def verify_scalers(v3_root: Path, manifest: dict) -> list[str]:
errors: list[str] = []
scalers = manifest.get("scalers", {})
for name, rel in scalers.items():
p = v3_root / rel
if not p.exists():
errors.append(f"scaler missing: {name} -> {rel}")
scaler_checksums = manifest.get("scaler_checksums", {})
for name, expected in scaler_checksums.items():
rel = scalers.get(name)
if not rel or expected is None:
continue
p = v3_root / rel
if not p.exists():
continue
actual = sha256_file(p)
if actual != expected:
errors.append(f"scaler checksum mismatch: {name} -> {rel}")
return errors
def verify_auxiliary(v3_root: Path, manifest: dict) -> list[str]:
errors: list[str] = []
aux = manifest.get("auxiliary_artifacts", {})
if not isinstance(aux, dict):
return ["auxiliary_artifacts must be an object"]
for aux_id, aux_meta in aux.items():
if not isinstance(aux_meta, dict):
errors.append(f"auxiliary_artifacts.{aux_id} must be an object")
continue
rel = aux_meta.get("file")
sha = aux_meta.get("sha256")
if not rel:
continue
p = v3_root / rel
if not p.exists():
errors.append(f"auxiliary file missing: {aux_id} -> {rel}")
continue
if sha and sha != sha256_file(p):
errors.append(f"auxiliary sha mismatch: {aux_id} -> {rel}")
return errors
def main() -> int:
parser = argparse.ArgumentParser(description="Verify v3 artifact checksums from models.json")
parser.add_argument("--v3-root", default="artifacts/v3", help="Path to v3 artifact root")
args = parser.parse_args()
v3_root = Path(args.v3_root).resolve()
manifest_path = v3_root / "models.json"
if not manifest_path.exists():
raise SystemExit(f"models.json not found at: {manifest_path}")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
errors: list[str] = []
errors.extend(verify_model_entries(v3_root, manifest))
errors.extend(verify_scalers(v3_root, manifest))
errors.extend(verify_auxiliary(v3_root, manifest))
checksums = manifest.get("checksums", {})
for section in ("models", "scalers", "results", "features"):
errors.extend(verify_sections(v3_root, checksums, section))
if errors:
print("VERIFICATION FAILED")
for e in errors:
print(f" - {e}")
return 1
print("VERIFICATION PASSED")
print(f"manifest: {manifest_path}")
print(f"models: {len(manifest.get('models', {}))}")
print(f"auxiliary: {len(manifest.get('auxiliary_artifacts', {}))}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|