DemonGatanjieuandClaude Opus 5.5
851a8cb2d5
On an AutoDL server every model in the models folder is a link into the platform's shared, read-only model disk, named like 6b4679... without an extension. Scans wrote the .info beside the link, but get_metadata resolved the path and read beside the disk's file: a scanned model showed as not scanned, under the disk's file name. Delete, edit, open folder, MCP and saved-workflow lookups resolved the file too, so they refused it as outside the models folder (or, for a link inside it, would have acted on the file behind it). path_utils.model_file_within joins a resolved folder with a bare name (require_filename, which now refuses "." and ".." too); model_reference_within resolves the subfolders of a relative path inside the root, through links as before, and takes the file as named. get_metadata reads beside the path it is given (abspath). Edits, renames and deletes act on the link; nothing writes into a model file. Checked with a link stood in by patching realpath (this computer cannot make links): the info beside the link is read and the name is the link's; "..", ".", "a/b" and a relative path leaving the root are refused. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
222 lines
9.2 KiB
Python
222 lines
9.2 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
|
|
|
|
# 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": "",
|
|
# "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": [],
|
|
}
|
|
|
|
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 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)
|
|
# 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 {}
|
|
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):
|
|
# 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()
|
|
|
|
|