Will Miao
19abf7ed03
Distinguish same-version quantized models (e.g. GGUF Q4_K_M vs Q8_0) on cards, which previously showed only the identical CivitAI version name. Detection is content-based and stored as a three-state sidecar field (absent = unchecked, "" = checked-unavailable, value = canonical name), mirroring the autov3 pipeline: - GGUF: minimal pure-struct header parser reads general.file_type - safetensors: dominant tensor dtype F8_* -> FP8 - CivitAI files[].metadata.fp == "fp8" fills in as a fallback only - A one-time backfill service covers existing libraries; each file costs only a few-hundred-byte header read The footer renders the badge after the version name (suppressed in grouped mode), gated by a new show_quantization_on_card setting that follows the show_version_on_card toggle pattern.
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""
|
|
Cache Entry Validator
|
|
|
|
Validates and repairs cache entries to prevent runtime errors from
|
|
missing or invalid critical fields.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
import logging
|
|
import os
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
"""Result of validating a single cache entry."""
|
|
is_valid: bool
|
|
repaired: bool
|
|
errors: List[str] = field(default_factory=list)
|
|
entry: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class CacheEntryValidator:
|
|
"""
|
|
Validates and repairs cache entry core fields.
|
|
|
|
Critical fields that cause runtime errors when missing:
|
|
- file_path: KeyError in multiple locations
|
|
- sha256: KeyError/AttributeError in hash operations
|
|
|
|
Medium severity fields that may cause sorting/display issues:
|
|
- size: KeyError during sorting
|
|
- modified: KeyError during sorting
|
|
- model_name: AttributeError on .lower() calls
|
|
|
|
Low severity fields:
|
|
- tags: KeyError/TypeError in recipe operations
|
|
"""
|
|
|
|
# Field definitions: (default_value, is_required)
|
|
CORE_FIELDS: Dict[str, Tuple[Any, bool]] = {
|
|
'file_path': ('', True),
|
|
'sha256': ('', True),
|
|
'file_name': ('', False),
|
|
'model_name': ('', False),
|
|
'folder': ('', False),
|
|
'size': (0, False),
|
|
'modified': (0.0, False),
|
|
'tags': ([], False),
|
|
'preview_url': ('', False),
|
|
'base_model': ('', False),
|
|
'from_civitai': (True, False),
|
|
'favorite': (False, False),
|
|
'exclude': (False, False),
|
|
'db_checked': (False, False),
|
|
'preview_nsfw_level': (0, False),
|
|
'notes': ('', False),
|
|
'usage_tips': ('', False),
|
|
'hash_status': ('completed', False),
|
|
'autov3': (None, False),
|
|
'quantization': (None, False),
|
|
}
|
|
|
|
@classmethod
|
|
def validate(cls, entry: Dict[str, Any], *, auto_repair: bool = True) -> ValidationResult:
|
|
"""
|
|
Validate a single cache entry.
|
|
|
|
Args:
|
|
entry: The cache entry dictionary to validate
|
|
auto_repair: If True, attempt to repair missing/invalid fields
|
|
|
|
Returns:
|
|
ValidationResult with validation status and optionally repaired entry
|
|
"""
|
|
if entry is None:
|
|
return ValidationResult(
|
|
is_valid=False,
|
|
repaired=False,
|
|
errors=['Entry is None'],
|
|
entry=None
|
|
)
|
|
|
|
if not isinstance(entry, dict):
|
|
return ValidationResult(
|
|
is_valid=False,
|
|
repaired=False,
|
|
errors=[f'Entry is not a dict: {type(entry).__name__}'],
|
|
entry=None
|
|
)
|
|
|
|
errors: List[str] = []
|
|
repaired = False
|
|
|
|
# If auto_repair is on, we work on a copy. If not, we still need a safe way to check fields.
|
|
working_entry = dict(entry) if auto_repair else entry
|
|
|
|
# Determine effective hash_status for validation logic
|
|
hash_status = entry.get('hash_status')
|
|
if hash_status is None:
|
|
if auto_repair:
|
|
working_entry['hash_status'] = 'completed'
|
|
repaired = True
|
|
hash_status = 'completed'
|
|
|
|
for field_name, (default_value, is_required) in cls.CORE_FIELDS.items():
|
|
# Get current value from the original entry to avoid side effects during validation
|
|
value = entry.get(field_name)
|
|
|
|
# Check if field is missing or None
|
|
if value is None:
|
|
# Special case: sha256 can be None/empty if hash_status is pending
|
|
if field_name == 'sha256' and hash_status == 'pending':
|
|
if auto_repair:
|
|
working_entry[field_name] = ''
|
|
repaired = True
|
|
continue
|
|
|
|
if is_required:
|
|
errors.append(f"Required field '{field_name}' is missing or None")
|
|
if auto_repair:
|
|
# A missing optional field whose default is None is already
|
|
# semantically equal to its default (e.g. autov3: absent
|
|
# means "not checked") — writing None back is a no-op, not
|
|
# a repair.
|
|
if default_value is not None:
|
|
working_entry[field_name] = cls._get_default_copy(default_value)
|
|
repaired = True
|
|
continue
|
|
|
|
# Validate field type and value
|
|
field_error = cls._validate_field(field_name, value, default_value)
|
|
if field_error:
|
|
# Special case: allow empty string for sha256 if pending
|
|
if field_name == 'sha256' and hash_status == 'pending' and value == '':
|
|
continue
|
|
|
|
errors.append(field_error)
|
|
if auto_repair:
|
|
working_entry[field_name] = cls._get_default_copy(default_value)
|
|
repaired = True
|
|
|
|
# Special validation: file_path must not be empty for required field
|
|
file_path = working_entry.get('file_path', '')
|
|
if not file_path or (isinstance(file_path, str) and not file_path.strip()):
|
|
errors.append("Required field 'file_path' is empty")
|
|
# Cannot repair empty file_path - entry is invalid
|
|
return ValidationResult(
|
|
is_valid=False,
|
|
repaired=repaired,
|
|
errors=errors,
|
|
entry=working_entry if auto_repair else None
|
|
)
|
|
|
|
# Special validation: sha256 must not be empty for required field
|
|
# BUT allow empty sha256 when hash_status is pending (lazy hash calculation)
|
|
sha256 = working_entry.get('sha256', '')
|
|
# Use the effective hash_status we determined earlier
|
|
if not sha256 or (isinstance(sha256, str) and not sha256.strip()):
|
|
# Allow empty sha256 for lazy hash calculation (checkpoints)
|
|
if hash_status != 'pending':
|
|
errors.append("Required field 'sha256' is empty")
|
|
# Cannot repair empty sha256 - entry is invalid
|
|
return ValidationResult(
|
|
is_valid=False,
|
|
repaired=repaired,
|
|
errors=errors,
|
|
entry=working_entry if auto_repair else None
|
|
)
|
|
|
|
# Normalize sha256 to lowercase if needed
|
|
if isinstance(sha256, str):
|
|
normalized_sha = sha256.lower().strip()
|
|
if normalized_sha != sha256:
|
|
if auto_repair:
|
|
working_entry['sha256'] = normalized_sha
|
|
repaired = True
|
|
else:
|
|
# If not auto-repairing, we don't consider case difference as a "critical error"
|
|
# that invalidates the entry, but we also don't mark it repaired.
|
|
pass
|
|
|
|
# Normalize autov3 to lowercase if needed (optional field, never stripped).
|
|
autov3 = working_entry.get('autov3')
|
|
if isinstance(autov3, str) and autov3:
|
|
normalized_autov3 = autov3.lower()
|
|
if normalized_autov3 != autov3:
|
|
if auto_repair:
|
|
working_entry['autov3'] = normalized_autov3
|
|
repaired = True
|
|
|
|
# Determine if entry is valid
|
|
# Entry is valid if no critical required field errors remain after repair
|
|
# Critical fields are file_path and sha256
|
|
CRITICAL_REQUIRED_FIELDS = {'file_path', 'sha256'}
|
|
has_critical_errors = any(
|
|
"Required field" in error and
|
|
any(f"'{field}'" in error for field in CRITICAL_REQUIRED_FIELDS)
|
|
for error in errors
|
|
)
|
|
|
|
is_valid = not has_critical_errors
|
|
|
|
return ValidationResult(
|
|
is_valid=is_valid,
|
|
repaired=repaired,
|
|
errors=errors,
|
|
entry=working_entry if auto_repair else entry
|
|
)
|
|
|
|
@classmethod
|
|
def validate_batch(
|
|
cls,
|
|
entries: List[Dict[str, Any]],
|
|
*,
|
|
auto_repair: bool = True
|
|
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
|
|
"""
|
|
Validate a batch of cache entries.
|
|
|
|
Args:
|
|
entries: List of cache entry dictionaries to validate
|
|
auto_repair: If True, attempt to repair missing/invalid fields
|
|
|
|
Returns:
|
|
Tuple of (valid_entries, invalid_entries)
|
|
"""
|
|
if not entries:
|
|
return [], []
|
|
|
|
valid_entries: List[Dict[str, Any]] = []
|
|
invalid_entries: List[Dict[str, Any]] = []
|
|
|
|
for entry in entries:
|
|
result = cls.validate(entry, auto_repair=auto_repair)
|
|
|
|
if result.is_valid:
|
|
# Use repaired entry if available, otherwise original
|
|
valid_entries.append(result.entry if result.entry else entry)
|
|
else:
|
|
invalid_entries.append(entry)
|
|
# Log invalid entries for debugging
|
|
file_path = entry.get('file_path', '<unknown>') if isinstance(entry, dict) else '<not a dict>'
|
|
logger.warning(
|
|
f"Invalid cache entry for '{file_path}': {', '.join(result.errors)}"
|
|
)
|
|
|
|
return valid_entries, invalid_entries
|
|
|
|
@classmethod
|
|
def _validate_field(cls, field_name: str, value: Any, default_value: Any) -> Optional[str]:
|
|
"""
|
|
Validate a specific field value.
|
|
|
|
Returns an error message if invalid, None if valid.
|
|
"""
|
|
expected_type = type(default_value)
|
|
|
|
# Special case: autov3 is optional with a three-state contract.
|
|
# None = not checked, "" = checked but unavailable, otherwise a
|
|
# 12-character hex string (case-insensitive here; normalized to
|
|
# lowercase separately).
|
|
if field_name == 'autov3':
|
|
if value is None or value == "":
|
|
return None
|
|
if not isinstance(value, str):
|
|
return f"Field 'autov3' should be string or None, got {type(value).__name__}"
|
|
if len(value) != 12 or any(c not in '0123456789abcdefABCDEF' for c in value):
|
|
return "Field 'autov3' should be a 12-character hex string"
|
|
return None
|
|
|
|
# Special case: quantization shares autov3's three-state contract
|
|
# (None = not checked, "" = checked but unavailable) but the value is
|
|
# a canonical name ("Q4_K_M", "FP8"), not a hex string, so only the
|
|
# type is constrained.
|
|
if field_name == 'quantization':
|
|
if value is None or value == "":
|
|
return None
|
|
if not isinstance(value, str):
|
|
return f"Field 'quantization' should be string or None, got {type(value).__name__}"
|
|
return None
|
|
|
|
# Special handling for numeric types
|
|
if expected_type == int:
|
|
if not isinstance(value, (int, float)):
|
|
return f"Field '{field_name}' should be numeric, got {type(value).__name__}"
|
|
elif expected_type == float:
|
|
if not isinstance(value, (int, float)):
|
|
return f"Field '{field_name}' should be numeric, got {type(value).__name__}"
|
|
elif expected_type == bool:
|
|
# Be lenient with boolean fields - accept truthy/falsy values
|
|
pass
|
|
elif expected_type == str:
|
|
if not isinstance(value, str):
|
|
return f"Field '{field_name}' should be string, got {type(value).__name__}"
|
|
elif expected_type == list:
|
|
if not isinstance(value, (list, tuple)):
|
|
return f"Field '{field_name}' should be list, got {type(value).__name__}"
|
|
|
|
return None
|
|
|
|
@classmethod
|
|
def _get_default_copy(cls, default_value: Any) -> Any:
|
|
"""Get a copy of the default value to avoid shared mutable state."""
|
|
if isinstance(default_value, list):
|
|
return list(default_value)
|
|
if isinstance(default_value, dict):
|
|
return dict(default_value)
|
|
return default_value
|
|
|
|
@classmethod
|
|
def get_file_path_safe(cls, entry: Dict[str, Any], default: str = '') -> str:
|
|
"""Safely get file_path from an entry."""
|
|
if not isinstance(entry, dict):
|
|
return default
|
|
value = entry.get('file_path')
|
|
if isinstance(value, str):
|
|
return value
|
|
return default
|
|
|
|
@classmethod
|
|
def get_sha256_safe(cls, entry: Dict[str, Any], default: str = '') -> str:
|
|
"""Safely get sha256 from an entry."""
|
|
if not isinstance(entry, dict):
|
|
return default
|
|
value = entry.get('sha256')
|
|
if isinstance(value, str):
|
|
return value.lower()
|
|
return default
|