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 from ..model_sidecars import find except ImportError: from model_identity import USER_INFO_SUFFIX, is_unmatched, read_json, sidecar_file_hash, sidecar_info, unmatched_reason from model_sidecars import find from .model_heads import inspect as inspect_file # the header, read once per file version # Fields the model editor sets: .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 _info_base(file_path, base_path): """Where the model's info files are read: beside the model as named; for a link with none beside it, beside the file it links to, where another tool (A1111's Civitai Helper) may have left them. Only read there: covers, edits and scans stay beside the link.""" if os.path.islink(file_path) and not (find(base_path, ".info") or os.path.exists(base_path + ".civitai.info")): return os.path.splitext(os.path.realpath(file_path))[0] return base_path def _read_metadata(file_path): base_path = os.path.splitext(file_path)[0] info_base = _info_base(file_path, base_path) 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 = [path for path in (find(info_base, ".info"), f"{info_base}.civitai.info") if path] 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(info_base) 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_file = find(base_path, USER_INFO_SUFFIX) user = (read_json(user_file) if user_file else None) 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(find(base_path, ".info") or f"{base_path}.info"), _file_signature(f"{base_path}.civitai.info"), _file_signature(find(base_path, USER_INFO_SUFFIX) or 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): # As named, not resolved: a model that is a link has its covers and info beside the link. normalized_path = os.path.abspath(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()