Ben FinkleaandClaude Opus 4.5
9256a1c87c
A custom node extension for ComfyUI that provides: - Model version tracking for HuggingFace and CivitAI - Automatic model updates with configurable scheduling - Multi-GPU synchronization with VRAM-based filtering - Metadata management with hash verification Features: - SQLite database for model metadata storage - REST API endpoints for all operations - Web UI panel integrated into ComfyUI menu - Background scheduler for automatic update checks - rsync-based sync to remote machines - VRAM estimation for model filtering Tests: - 17 unit tests covering all library modules - Playwright integration tests for UI verification Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
95 lines
2.4 KiB
Python
95 lines
2.4 KiB
Python
"""Hash computation utilities for model files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional, Callable, Union
|
|
|
|
|
|
def compute_sha256(file_path: Union[str, Path], chunk_size: int = 8192) -> str:
|
|
"""
|
|
Compute SHA256 hash of a file.
|
|
|
|
Args:
|
|
file_path: Path to the file to hash
|
|
chunk_size: Size of chunks to read (default 8KB)
|
|
|
|
Returns:
|
|
Hexadecimal SHA256 hash string
|
|
"""
|
|
sha256 = hashlib.sha256()
|
|
with open(file_path, "rb") as f:
|
|
while chunk := f.read(chunk_size):
|
|
sha256.update(chunk)
|
|
return sha256.hexdigest()
|
|
|
|
|
|
def compute_sha256_chunked(
|
|
file_path: Union[str, Path],
|
|
progress_callback: Optional[Callable[[int, int], None]] = None,
|
|
chunk_size: int = 1024 * 1024 # 1MB chunks for progress
|
|
) -> str:
|
|
"""
|
|
Compute SHA256 hash with progress reporting.
|
|
|
|
Args:
|
|
file_path: Path to the file to hash
|
|
progress_callback: Callable(bytes_processed, total_bytes) for progress
|
|
chunk_size: Size of chunks to read (default 1MB for better progress)
|
|
|
|
Returns:
|
|
Hexadecimal SHA256 hash string
|
|
"""
|
|
file_path = Path(file_path)
|
|
total_size = file_path.stat().st_size
|
|
processed = 0
|
|
|
|
sha256 = hashlib.sha256()
|
|
with open(file_path, "rb") as f:
|
|
while chunk := f.read(chunk_size):
|
|
sha256.update(chunk)
|
|
processed += len(chunk)
|
|
if progress_callback:
|
|
progress_callback(processed, total_size)
|
|
|
|
return sha256.hexdigest()
|
|
|
|
|
|
def verify_file_hash(file_path: Union[str, Path], expected_hash: str) -> bool:
|
|
"""
|
|
Verify a file matches an expected SHA256 hash.
|
|
|
|
Args:
|
|
file_path: Path to the file to verify
|
|
expected_hash: Expected SHA256 hash (hex string)
|
|
|
|
Returns:
|
|
True if hash matches, False otherwise
|
|
"""
|
|
actual_hash = compute_sha256(file_path)
|
|
return actual_hash.lower() == expected_hash.lower()
|
|
|
|
|
|
def get_file_info(file_path: Union[str, Path]) -> dict:
|
|
"""
|
|
Get file information including size and modification time.
|
|
|
|
Args:
|
|
file_path: Path to the file
|
|
|
|
Returns:
|
|
Dict with size, mtime, and exists keys
|
|
"""
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
return {"exists": False, "size": 0, "mtime": None}
|
|
|
|
stat = path.stat()
|
|
return {
|
|
"exists": True,
|
|
"size": stat.st_size,
|
|
"mtime": stat.st_mtime,
|
|
}
|