#!/usr/bin/env python3 """Verify the local FLUX.1-dev ConvRot Hugging Face release package.""" from __future__ import annotations import argparse import hashlib import json import math import os from datetime import datetime, timezone from pathlib import Path from typing import Any CHUNK_BYTES = 16 * 1024 * 1024 ROOT = Path(__file__).resolve().parents[1] MANIFEST_PATH = ROOT / "release_manifest.json" OUTPUT_PATH = ROOT / "release_verification.json" class VerificationError(RuntimeError): pass def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--model-root", type=Path, default=Path(os.environ.get("COMFYUI_MODELS_ROOT", "/mnt/d/comfyui/models")), ) parser.add_argument( "--models", action="store_true", help="Hash all four model artifacts; required before upload.", ) return parser.parse_args() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(CHUNK_BYTES), b""): digest.update(chunk) return digest.hexdigest() def load_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise VerificationError(f"expected JSON object: {path}") return value def require(condition: bool, message: str) -> None: if not condition: raise VerificationError(message) def verify_regular_file(path: Path) -> None: require(path.exists(), f"missing file: {path}") require(path.is_file(), f"not a regular file: {path}") require(not path.is_symlink(), f"symlink is not allowed: {path}") def main() -> int: args = parse_args() manifest = load_json(MANIFEST_PATH) require( manifest.get("status") in {"ready_for_local_verification", "verified_private_uploaded"}, "release status drift", ) require(manifest.get("remote_status") in {"not_created", "private"}, "remote-status drift") required = manifest.get("required_repository_files") artifacts = manifest.get("artifacts") if not isinstance(required, list): raise VerificationError("missing required_repository_files") if not isinstance(artifacts, list) or len(artifacts) != 4: raise VerificationError("expected four artifacts") support_records: list[dict[str, Any]] = [] for relative in required: path = ROOT / str(relative) verify_regular_file(path) support_records.append( { "path": str(relative), "bytes": path.stat().st_size, "sha256": sha256_file(path), } ) unexpected_models = sorted(str(path.relative_to(ROOT)) for path in ROOT.rglob("*.safetensors")) require(not unexpected_models, f"model files must not be duplicated into staging: {unexpected_models}") readme = (ROOT / "README.md").read_text(encoding="utf-8") for marker in ( "license: other", "license_name: flux-1-dev-non-commercial-license", "base_model: black-forest-labs/FLUX.1-dev", "base_model_relation: quantized", "pipeline_tag: text-to-image", "FLUX.1-dev ConvRot for ComfyUI", ): require(marker in readme, f"model-card marker missing: {marker}") workflow = load_json(ROOT / "workflows/FLUX1_Dev_ConvRot_API.json") require(workflow["1"]["class_type"] == "UNETLoader", "workflow loader drift") require( workflow["1"]["inputs"]["unet_name"] == "FLUX.1-dev-w8a8-convrot.safetensors", "workflow DiT default drift", ) require( workflow["2"]["inputs"]["clip_name2"] == "t5xxl_flux1_int8_convrot.safetensors", "workflow T5 default drift", ) evidence = manifest["evidence"] overview = ROOT / evidence["overview_contact_sheet"]["repo_path"] require(overview.stat().st_size == evidence["overview_contact_sheet"]["bytes"], "overview byte drift") require(sha256_file(overview) == evidence["overview_contact_sheet"]["sha256"], "overview hash drift") quality = load_json(ROOT / "assets/quality_metrics.json") require(quality.get("status") == "verified_complete", "quality evidence status drift") require( quality.get("source_grid_manifest_sha256") == evidence["source_grid_manifest_sha256"], "quality source-manifest drift", ) results = quality.get("results") if not isinstance(results, list) or len(results) != 4: raise VerificationError("quality result inventory drift") measured = {str(item["model_id"]): float(item["global_psnr_db"]) for item in results} for model_id, expected in evidence["quality_global_psnr_db"].items(): require(model_id in measured, f"missing quality result: {model_id}") require(math.isclose(measured[model_id], float(expected), rel_tol=0.0, abs_tol=1e-12), f"quality drift: {model_id}") model_root = args.model_root.expanduser().resolve(strict=True) artifact_records: list[dict[str, Any]] = [] total_bytes = 0 repo_paths: set[str] = set() for artifact in artifacts: repo_path = str(artifact["repo_path"]) require(repo_path not in repo_paths, f"duplicate repo path: {repo_path}") repo_paths.add(repo_path) source = (model_root / str(artifact["source_relative_path"])).resolve(strict=True) require(source.is_file(), f"artifact is not a file: {source}") actual_bytes = source.stat().st_size require(actual_bytes == int(artifact["bytes"]), f"artifact byte drift: {source}") total_bytes += actual_bytes actual_hash = None if args.models: actual_hash = sha256_file(source) require(actual_hash == str(artifact["sha256"]), f"artifact hash drift: {source}") source_manifest_relative = artifact.get("source_manifest_relative_path") if source_manifest_relative is not None: source_manifest = (model_root / str(source_manifest_relative)).resolve(strict=True) verify_regular_file(source_manifest) require( sha256_file(source_manifest) == str(artifact["source_manifest_sha256"]), f"source conversion-manifest drift: {source_manifest}", ) artifact_records.append( { "repo_path": repo_path, "source_relative_path": str(artifact["source_relative_path"]), "bytes": actual_bytes, "sha256": actual_hash if actual_hash is not None else str(artifact["sha256"]), "hash_verified": bool(args.models), } ) require(total_bytes == int(manifest["artifact_total_bytes"]), "artifact total-byte drift") result = { "schema_version": 1, "status": "PASS", "verified_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "suggested_repo_id": manifest["suggested_repo_id"], "remote_status": manifest["remote_status"], "model_hashes_verified": bool(args.models), "artifact_count": len(artifact_records), "artifact_total_bytes": total_bytes, "artifacts": artifact_records, "support_file_count": len(support_records), "support_files": support_records, "overview_sha256": evidence["overview_contact_sheet"]["sha256"], "quality_source_grid_manifest_sha256": evidence["source_grid_manifest_sha256"], } temporary = OUTPUT_PATH.with_name(f".{OUTPUT_PATH.name}.{os.getpid()}.tmp") temporary.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.replace(temporary, OUTPUT_PATH) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())