fix(civitai): stop check-local from reporting deleted model files

A hash cache hit now resolves through its loader's registry and counts
only while the file exists; entries whose file is gone are pruned, on
the request that finds them and once per process for the rest.
filename is always the local path.
This commit is contained in:
CalamitousFelicitousness
2026-09-14 00:01:14 +01:00
parent 229db02b14
commit 7554389f2e
2 changed files with 112 additions and 37 deletions
+42 -37
View File
@@ -567,7 +567,6 @@ def buildsidecar_index():
continue
# Match the companion file to a JSON entry by size (sizeKB)
companion_size_kb = os.path.getsize(companion) / 1024.0
companion_name = os.path.basename(base)
best_sha = None
best_diff = float('inf')
for v in data.get('modelVersions', []):
@@ -580,7 +579,7 @@ def buildsidecar_index():
best_sha = sha
best_diff = diff
if best_sha:
sidecar_index[best_sha.lower()] = {"filename": companion_name, "type": model_type}
sidecar_index[best_sha.lower()] = {"filename": companion, "type": model_type}
except Exception:
continue
log.debug(f'CivitAI sidecar index: {len(sidecar_index)} hashes from sidecar files')
@@ -597,54 +596,60 @@ def invalidatesidecar_index():
# ---------------------------------------------------------------------------
def post_check_local(request: dict):
"""Check which SHA256 hashes correspond to locally downloaded files."""
"""Check which SHA256 hashes correspond to local model files, dropping hash cache entries whose files are gone."""
from modules import hashes as hash_module
input_hashes = request.get('hashes', [])
if not input_hashes:
from modules.civitai.filemanage_civitai import hash_cache_path, prune_hash_cache
requested = [str(h) for h in request.get('hashes', []) if h]
if not requested:
return {"found": {}}
# Build reverse lookup: lowercase sha256 -> {filename, type}
prune_hash_cache()
wanted = {h.lower() for h in requested}
titles_by_sha: dict[str, list[str]] = {}
for title, entry in list(hash_module.cache().items()):
sha = (entry.get("sha256") or "").lower()
if sha in wanted:
titles_by_sha.setdefault(sha, []).append(title)
found = {}
for title, entry in hash_module.cache().items():
sha = entry["sha256"]
if not sha:
continue
parts = title.split("/", 1)
file_type = parts[0] if len(parts) > 1 else "unknown"
found[sha.lower()] = {"filename": title, "type": file_type}
# Supplement from in-memory checkpoint registry
gone = []
for sha, titles in titles_by_sha.items():
for title in titles:
path = hash_cache_path(title)
if path is None: # no loaded registry names the file
continue
if os.path.exists(path):
found[sha] = {"filename": path, "type": title.split("/", 1)[0]}
break
gone.append(title)
if gone:
for title in gone:
hash_module.cache().pop(title, None)
hash_module.save_cache()
log.debug(f'CivitAI check local: pruned={len(gone)} hash cache entries without files')
try:
from modules.sd_checkpoint import checkpoints_list
for _title, cp in checkpoints_list.items():
if cp.sha256:
key = cp.sha256.lower()
if key not in found:
found[key] = {"filename": cp.filename, "type": "checkpoint"}
for cp in checkpoints_list.values():
key = (cp.sha256 or "").lower()
if key in wanted and key not in found and os.path.exists(cp.filename):
found[key] = {"filename": cp.filename, "type": "checkpoint"}
except Exception:
pass
# Supplement from in-memory LoRA registry
try:
from modules.lora.lora_load import available_networks
for _name, net in available_networks.items():
if net.hash:
key = net.hash.lower()
if key not in found:
found[key] = {"filename": net.filename, "type": "lora"}
for net in available_networks.values():
key = (net.hash or "").lower()
if key in wanted and key not in found and os.path.isfile(net.filename):
found[key] = {"filename": net.filename, "type": "lora"}
except Exception:
pass
# Supplement from sidecar index (covers files never hashed locally)
sidecar = buildsidecar_index()
for h in input_hashes:
if not h:
continue
key = h.lower()
if key not in found and key in sidecar:
found[key] = sidecar[key]
# Match requested hashes
result = {}
for h in input_hashes:
if not h:
continue
match = found.get(h.lower())
for h in requested:
key = h.lower()
match = found.get(key)
if match is None:
entry = sidecar.get(key)
if entry and os.path.isfile(entry["filename"]):
match = entry
if match:
result[h] = match
return {"found": result}
+70
View File
@@ -142,6 +142,76 @@ def hash_cache_title(kind: str | None, filename: str, name: str | None = None) -
return None
hash_cache_pruned = False
def loader_root(kind: str) -> str | None:
"""Folder the loader of kind lists, or None for a kind no loader owns."""
from modules import shared, paths
if kind == 'vae':
return getattr(shared.opts, 'vae_dir', '') or os.path.join(paths.models_path, 'VAE')
if kind == 'unet':
return getattr(shared.opts, 'unet_dir', '')
if kind == 'lora':
return getattr(shared.cmd_opts, 'lora_dir', '')
if kind == 'checkpoint':
return getattr(shared.opts, 'ckpt_dir', '') or os.path.join(paths.models_path, 'Stable-diffusion')
return None
def loader_registry(kind: str) -> dict[str, str] | None:
"""Name to path map of the loader that reads the kind's hash cache keys, or None for a kind no loader owns."""
if kind == 'vae':
from modules.sd_vae import vae_dict
return vae_dict
if kind == 'unet':
from modules.sd_unet import unet_dict
return unet_dict
if kind == 'lora':
from modules.lora.lora_load import available_networks
return {name: entry.filename for name, entry in available_networks.items()}
if kind == 'checkpoint':
from modules.sd_checkpoint import checkpoints_list
return {entry.name: entry.filename for entry in checkpoints_list.values()}
return None
def hash_cache_path(title: str) -> str | None:
"""Path the loader registry holds for a hash cache key, or None when no loaded registry names it."""
kind, _, name = title.partition('/')
if kind == 'vae' and os.path.isabs(name):
return name
registry = loader_registry(kind)
return registry.get(name) if registry else None
def hash_cache_stale(title: str) -> bool:
"""True when the key's loader folder is reachable but the registry no longer lists the file, or lists a path that is gone."""
kind, _, name = title.partition('/')
if kind == 'vae' and os.path.isabs(name):
return os.path.isdir(os.path.dirname(name)) and not os.path.exists(name)
root = loader_root(kind)
if not root or not os.path.isdir(root):
return False
path = (loader_registry(kind) or {}).get(name)
return path is None or not os.path.exists(path)
def prune_hash_cache():
"""Drop hash cache entries for files that are gone, once per process."""
global hash_cache_pruned # pylint: disable=global-statement
if hash_cache_pruned:
return
hash_cache_pruned = True
from modules import hashes
gone = [title for title in list(hashes.cache()) if hash_cache_stale(title)]
for title in gone:
hashes.cache().pop(title, None)
if gone:
hashes.save_cache()
log.info(f'CivitAI hash cache: pruned={len(gone)} entries without files')
def resolve_save_path(model_type: str, model_name: str = "", base_model: str = "",
nsfw: bool = False, creator: str = "", model_id: int = 0,
version_id: int = 0, version_name: str = "") -> Path: