DemonGatanjieuandClaude Opus 5.5
908ad72ec2
- The scan summary marks new models whose safetensors header reads but names no drawing model
kind (a voice model such as F5-TTS) as not_drawing; the scan page asks: don't scan them,
scan them, or see each. The choice lives in <model>.anomalous.json ("scan": "skip" / "scan",
model_identity.scan_choice), saved by POST /anomalous/scan_choice (api/scan_choice.py, this
computer only, at most 2000 models, only readable .safetensors layers).
- "skip": a scan of a whole folder leaves the model out (picked by hand it is still scanned),
the summary counts it apart (excluded) with its own list to change it back, the card and the
details say "不扫描"; the details have the same button.
- The unmatched count warns only for damaged files or lookups still to do; not-found rows say
the fingerprint is kept for repairing workflows.
- model_kind.inspect_file now gives format None when the header cannot be read, as its text said.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
239 lines
10 KiB
Python
239 lines
10 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import urllib.parse
|
|
import subprocess
|
|
import threading
|
|
import asyncio
|
|
import copy
|
|
from functools import lru_cache
|
|
from aiohttp import web
|
|
import folder_paths
|
|
import struct
|
|
try:
|
|
from ..model_identity import USER_INFO_SUFFIX, is_unmatched, read_json, sidecar_file_hash, sidecar_info, unmatched_reason
|
|
except ImportError:
|
|
from model_identity import USER_INFO_SUFFIX, is_unmatched, read_json, sidecar_file_hash, sidecar_info, unmatched_reason
|
|
try:
|
|
from ..model_kind import inspect_file
|
|
except ImportError:
|
|
from model_kind import inspect_file
|
|
|
|
# Fields the model editor sets: <model>.anomalous.json key -> metadata key.
|
|
USER_FIELDS = {"custom_name": "custom_name", "custom_notes": "custom_notes", "source_url": "source_url"}
|
|
|
|
def _select_info_file(data, file_path):
|
|
"""Select the Civitai file entry that actually describes ``file_path``.
|
|
|
|
A model version may contain a checkpoint, text encoder, and VAE in the same
|
|
``files`` array. Taking the first SHA256 silently assigns the dependency's
|
|
hash to every local sidecar copied from that version. Physical byte size is
|
|
the strongest cheap discriminator available here and remains valid after a
|
|
local rename. Filenames are intentionally excluded: local renames are the
|
|
reason provenance recovery exists and cannot be identity evidence.
|
|
"""
|
|
entries = []
|
|
for file_info in data.get("files", []):
|
|
if not isinstance(file_info, dict):
|
|
continue
|
|
hashes = file_info.get("hashes", {})
|
|
sha256 = hashes.get("SHA256") if isinstance(hashes, dict) else None
|
|
if not sha256:
|
|
continue
|
|
|
|
size_bytes = None
|
|
try:
|
|
if file_info.get("sizeKB") is not None:
|
|
size_bytes = int(round(float(file_info["sizeKB"]) * 1024))
|
|
except (TypeError, ValueError, OverflowError):
|
|
pass
|
|
entries.append((file_info, str(sha256), size_bytes))
|
|
|
|
if not entries:
|
|
return None
|
|
|
|
try:
|
|
physical_size = os.path.getsize(file_path)
|
|
except OSError:
|
|
physical_size = None
|
|
|
|
if physical_size is not None:
|
|
size_matches = [entry for entry in entries if entry[2] == physical_size]
|
|
if len(size_matches) == 1:
|
|
return size_matches[0][0]
|
|
if len(size_matches) > 1:
|
|
return None
|
|
|
|
# Offline-generated metadata commonly contains one injected hash without
|
|
# Civitai size/name data. It is safe only when there is a single candidate.
|
|
if len(entries) == 1:
|
|
return entries[0][0]
|
|
return None
|
|
|
|
|
|
def _select_info_hash(data, file_path):
|
|
selected = _select_info_file(data, file_path)
|
|
if not selected:
|
|
return ""
|
|
hashes = selected.get("hashes", {})
|
|
return str(hashes.get("SHA256", "")) if isinstance(hashes, dict) else ""
|
|
|
|
|
|
def _positive_civitai_id(value):
|
|
"""Return a real Civitai identifier, excluding offline sentinel values."""
|
|
if isinstance(value, bool):
|
|
return None
|
|
try:
|
|
parsed = int(value)
|
|
except (TypeError, ValueError, OverflowError):
|
|
return None
|
|
return parsed if parsed > 0 else None
|
|
|
|
def _read_metadata(file_path):
|
|
base_path = os.path.splitext(file_path)[0]
|
|
metadata = {
|
|
"name": os.path.basename(base_path),
|
|
"description": "",
|
|
"notes": "",
|
|
"trainedWords": [],
|
|
"baseModel": "",
|
|
"civitai_url": "",
|
|
"source_url": "",
|
|
"hash": "",
|
|
"hash_algorithm": "sha256",
|
|
"hash_scope": "file",
|
|
"hash_source": "",
|
|
"custom_name": "",
|
|
"custom_notes": "",
|
|
"version_name": "", # the Civitai version ("v2.0", "Illustrious"…), told apart on the card
|
|
"base_guess": "", # no base model known yet (not scanned): what the file's header tells
|
|
# "civitai" (matched), "local" (only what the file tells: no match, or offline) or "" (not scanned)
|
|
"info_source": "",
|
|
"unmatched_reason": "", # for "local": not_found | network | offline | "" (scanned before this was kept)
|
|
"user_fields": [],
|
|
"scan_choice": "", # the user's "skip" / "scan" (model_identity.scan_choice)
|
|
"not_drawing": False, # not scanned yet, and its header looks like no drawing model
|
|
}
|
|
|
|
info_files = [f"{base_path}.info", f"{base_path}.civitai.info"]
|
|
for info_file in info_files:
|
|
if os.path.exists(info_file):
|
|
try:
|
|
with open(info_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
name = data.get("name", "")
|
|
if "model" in data and isinstance(data["model"], dict):
|
|
model_name = data["model"].get("name", "")
|
|
# A model and version of one name (what the file alone tells) read once.
|
|
name = f"{model_name} - {name}" if model_name and name and model_name != name else model_name or name
|
|
|
|
description = data.get("description", "") or ""
|
|
if not description and "model" in data and isinstance(data["model"], dict):
|
|
description = data["model"].get("description", "")
|
|
notes = data.get("notes", "")
|
|
trained_words = data.get("trainedWords", [])
|
|
base_model = data.get("baseModel", "")
|
|
|
|
model_id = _positive_civitai_id(data.get("modelId"))
|
|
version_id = _positive_civitai_id(data.get("id"))
|
|
civitai_url = ""
|
|
if model_id is not None:
|
|
# Handle Civitai's new mature content policy
|
|
nsfw_level = data.get("nsfwLevel", 1)
|
|
is_nsfw = False
|
|
if "model" in data and isinstance(data["model"], dict):
|
|
is_nsfw = data["model"].get("nsfw", False)
|
|
|
|
domain = "civitai.red" if (nsfw_level > 1 or is_nsfw) else "civitai.com"
|
|
civitai_url = f"https://{domain}/models/{model_id}"
|
|
if version_id is not None:
|
|
civitai_url += f"?modelVersionId={version_id}"
|
|
|
|
|
|
selected_file = _select_info_file(data, file_path)
|
|
hash_val, hash_source = sidecar_file_hash(data, file_path, selected_file)
|
|
if hash_val:
|
|
metadata["hash_source"] = hash_source
|
|
|
|
if name: metadata["name"] = name
|
|
if description: metadata["description"] = description
|
|
if notes: metadata["notes"] = notes
|
|
if trained_words: metadata["trainedWords"] = trained_words
|
|
if base_model: metadata["baseModel"] = base_model
|
|
if civitai_url: metadata["civitai_url"] = civitai_url
|
|
if model_id is not None: metadata["model_id"] = model_id
|
|
if version_id is not None: metadata["version_id"] = version_id
|
|
if version_id is not None and isinstance(data.get("model"), dict) and isinstance(data.get("name"), str):
|
|
metadata["version_name"] = data["name"].strip()
|
|
if hash_val: metadata["hash"] = hash_val
|
|
if "anomalous_custom_name" in data and data["anomalous_custom_name"]: metadata["custom_name"] = data["anomalous_custom_name"]
|
|
if "anomalous_custom_notes" in data and data["anomalous_custom_notes"]: metadata["custom_notes"] = data["anomalous_custom_notes"]
|
|
if "anomalous_source_url" in data and data["anomalous_source_url"]: metadata["source_url"] = data["anomalous_source_url"]
|
|
except Exception:
|
|
pass
|
|
|
|
scan = sidecar_info(base_path)
|
|
if scan is not None:
|
|
metadata["info_source"] = "local" if is_unmatched(scan) else "civitai"
|
|
metadata["unmatched_reason"] = unmatched_reason(scan)
|
|
if scan is None:
|
|
found = inspect_file(file_path)
|
|
if not metadata["baseModel"]:
|
|
metadata["base_guess"] = found.get("base") or ""
|
|
# A header that reads but names no drawing model kind (a voice model, say).
|
|
metadata["not_drawing"] = found["format"] == "safetensors" and not found["kind"]
|
|
# The user's own layer comes last and wins, also when a field was cleared on purpose ("").
|
|
user = read_json(base_path + USER_INFO_SUFFIX) or {}
|
|
if user.get("scan") in ("skip", "scan"):
|
|
metadata["scan_choice"] = user["scan"]
|
|
for key, field in USER_FIELDS.items():
|
|
if isinstance(user.get(key), str):
|
|
metadata[field] = user[key]
|
|
metadata["user_fields"].append(field)
|
|
return metadata
|
|
|
|
|
|
def _file_signature(file_path):
|
|
try:
|
|
stat = os.stat(file_path)
|
|
return (stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns)
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _metadata_signature(file_path):
|
|
base_path = os.path.splitext(file_path)[0]
|
|
return (
|
|
_file_signature(file_path),
|
|
_file_signature(f"{base_path}.info"),
|
|
_file_signature(f"{base_path}.civitai.info"),
|
|
_file_signature(base_path + USER_INFO_SUFFIX),
|
|
)
|
|
|
|
|
|
@lru_cache(maxsize=4096)
|
|
def _get_metadata_cached(file_path, signature):
|
|
# ``signature`` is deliberately unused by the parser itself; it is part of
|
|
# the cache key and changes whenever the model or either metadata sidecar
|
|
# changes. The cache is bounded so removed/renamed models cannot grow it
|
|
# without limit.
|
|
return _read_metadata(file_path)
|
|
|
|
|
|
def get_metadata(file_path):
|
|
normalized_path = os.path.realpath(file_path)
|
|
signature = _metadata_signature(normalized_path)
|
|
# Metadata contains mutable lists. Never expose the cached object directly.
|
|
return copy.deepcopy(_get_metadata_cached(normalized_path, signature))
|
|
|
|
|
|
def clear_metadata_cache():
|
|
_get_metadata_cached.cache_clear()
|
|
|
|
|
|
def get_metadata_cache_info():
|
|
return _get_metadata_cached.cache_info()
|
|
|
|
|