DemonGatanjieuandClaude Fable 5.2
99ee111c81
Privacy mode left the AI blind twice over: the whole prompt was a token, so it could not help with light, lens or style, and model and LoRA file names still went out as written. - The prompt box has a first field, the gist: the picture in words the user writes for the AI to read. seal_answer hands it on as written beside the tokens of the private prompts; either alone can be handed in. - ai_sealed_models.py: while the Gates window's "Model names in a private task" is on (the default), each model of the library is one token ([M1], sealed as its model_id) however an answer writes it: model_id, loader name, file name in a text, card name. A model's own entry drops name, file, folder, links and hash and keeps type, base model, the author's settings and the user's note. expand writes a model token back as the model_id, as the loader's name among a node's values, as the card's name inside the AI's words. The task's menu lists what each token stands for. - The model editor has "Note for the AI" (ai_note, note_for_ai in the tools, searched by search_models): what a model is for, in words the AI may read. - A combo's name is sealed too: unless renamed it is made of its model's name and the start of its prompt. - The privacy skill says how to write around a gist and how to pick and pass on model tokens. Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
255 lines
12 KiB
Python
255 lines
12 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
|
|
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: <model>.anomalous.json key -> metadata key.
|
|
USER_FIELDS = {"custom_name": "custom_name", "custom_notes": "custom_notes", "source_url": "source_url", "ai_note": "ai_note"}
|
|
|
|
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": [],
|
|
"headerWords": [], # no trigger words from Civitai: what the LoRA's header suggests (header_words.py), a guess
|
|
"baseModel": "",
|
|
"civitai_url": "",
|
|
"source_url": "",
|
|
"hash": "",
|
|
"hash_algorithm": "sha256",
|
|
"hash_scope": "file",
|
|
"hash_source": "",
|
|
"custom_name": "",
|
|
"custom_notes": "",
|
|
"ai_note": "", # the user's words for an AI on what it is for; a private task hiding model names gives it too
|
|
"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
|
|
kept_words = data.get("anomalous_header_words")
|
|
if not trained_words and isinstance(kept_words, list):
|
|
metadata["headerWords"] = [str(word) for word in kept_words if str(word).strip()]
|
|
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()
|
|
|
|
|