DemonGatanjieuandClaude Opus 5.5
e466e9dfcf
The scan page now says which models are unmatched or not scanned yet (type, folder, why Civitai had nothing) and what each scan did, model by model; rows open the model, scan it, look it up again or search Civitai. The progress box shows the counts and, when floating, a way to the result. The activity log records a scan when it ends with its result, and model edits (from -> to) and deletions (files moved) in detail. - scraper.py reports one JSON line per changed model (--report-file); a failing model no longer ends the scan. Unmatched info keeps why (not_found / network / offline); online scans look network/offline ones up again. After two models in a row without an answer the rest of the scan infers from the files (--civitai-down for the next folders) instead of waiting out every timeout. - civitai_client.py: the network side moved out of scraper.py; tells a 404 from Civitai being unreachable (CivitaiUnreachable). - api/scan_report.py: ScanJob turns a scan's reports into one result, kept as the last scan (/anomalous/last_scan) and logged; /anomalous/scan_model opens a listed model. - /anomalous/scan_all takes picked or listed models as `targets`, so one scan has one result; scan_summary lists unmatched and new models with their reason. - Activity middleware takes details a handler attaches (request[ACTIVITY_DETAIL]). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
496 lines
25 KiB
Python
496 lines
25 KiB
Python
import os
|
|
import sys
|
|
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
if _SCRIPT_DIR not in sys.path:
|
|
sys.path.insert(0, _SCRIPT_DIR)
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except:
|
|
pass
|
|
try:
|
|
sys.stderr.reconfigure(encoding="utf-8")
|
|
except:
|
|
pass
|
|
import time
|
|
import json
|
|
import hashlib
|
|
import re
|
|
import urllib.request
|
|
import argparse
|
|
import shutil
|
|
from model_policies import is_physical_rename_protected
|
|
from recycle_bin import move_to_trash
|
|
from model_identity import USER_INFO_SUFFIX, computed_file_identity, is_unmatched, scan_info_path, sidecar_file_hash, sidecar_info as read_local_info, unmatched_reason
|
|
from civitai_client import CIVITAI_API_KEY, CivitaiUnreachable, download_media, fetch_civitai_info, fetch_model_description
|
|
|
|
|
|
# Fixed tuples avoid rebuilding long extension lists for every scanned model.
|
|
# Exact-path checks are intentionally used instead of directory-wide globbing.
|
|
MEDIA_EXTENSIONS = (".png", ".jpg", ".jpeg", ".webp", ".gif", ".avif", ".mp4", ".webm", ".mov", ".avi")
|
|
PREVIEW_SUFFIXES = tuple(f".preview{ext}" for ext in MEDIA_EXTENSIONS)
|
|
CIVITAI_BACKUP_SUFFIXES = tuple(f".civitai_bak{ext}" for ext in MEDIA_EXTENSIONS)
|
|
COVER_SUFFIXES = MEDIA_EXTENSIONS + PREVIEW_SUFFIXES
|
|
ACTIVE_COVER_SUFFIXES = PREVIEW_SUFFIXES + MEDIA_EXTENSIONS
|
|
SIDECAR_SUFFIXES = (
|
|
".info", ".civitai.info", USER_INFO_SUFFIX, ".json", ".txt", ".yaml",
|
|
*MEDIA_EXTENSIONS,
|
|
*PREVIEW_SUFFIXES,
|
|
*CIVITAI_BACKUP_SUFFIXES,
|
|
)
|
|
|
|
|
|
def same_file(a, b):
|
|
"""Byte-for-byte equal (covers are small; size is checked first)."""
|
|
try:
|
|
return os.path.getsize(a) == os.path.getsize(b) and calculate_sha256(a) == calculate_sha256(b)
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def cover_owners(base):
|
|
"""(Civitai's, the user's) active covers. A cover is Civitai's only when it is a copy of the
|
|
Civitai image saved last time (<model>.civitai_bak.*); any other cover is the user's."""
|
|
backups = [base + ext for ext in CIVITAI_BACKUP_SUFFIXES if os.path.exists(base + ext)]
|
|
covers = [base + ext for ext in ACTIVE_COVER_SUFFIXES if os.path.exists(base + ext)]
|
|
civitai = [c for c in covers if any(same_file(c, b) for b in backups)]
|
|
return civitai, [c for c in covers if c not in civitai]
|
|
|
|
|
|
def write_scan_progress(progress_file, phase, total=0, current=0, filename=""):
|
|
if not progress_file:
|
|
return
|
|
temp_file = f"{progress_file}.{os.getpid()}.tmp"
|
|
try:
|
|
with open(temp_file, 'w', encoding='utf-8') as f:
|
|
json.dump({
|
|
"phase": phase,
|
|
"total": total,
|
|
"current": current,
|
|
"filename": filename,
|
|
}, f, ensure_ascii=False)
|
|
os.replace(temp_file, progress_file)
|
|
except OSError:
|
|
try:
|
|
os.remove(temp_file)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def count_scan_files(target_folder, target_files_basenames, unmatched_only=False):
|
|
total = 0
|
|
selected = set(target_files_basenames)
|
|
for root, _, files in os.walk(target_folder):
|
|
total += sum(
|
|
1 for filename in files
|
|
if filename.endswith(".safetensors") and (not selected or filename in selected)
|
|
and (not unmatched_only or is_unmatched(read_local_info(os.path.join(root, os.path.splitext(filename)[0]))))
|
|
)
|
|
return total
|
|
|
|
def calculate_sha256(file_path: str) -> str:
|
|
"""计算文件的 SHA256 哈希值 (用于 Civitai 匹配)"""
|
|
sha256_hash = hashlib.sha256()
|
|
print(f"[*] 正在计算 Hash (大文件可能需要几分钟): {os.path.basename(file_path)}")
|
|
with open(file_path, "rb") as f:
|
|
for byte_block in iter(lambda: f.read(4096 * 1024), b""):
|
|
sha256_hash.update(byte_block)
|
|
return sha256_hash.hexdigest()
|
|
|
|
import struct
|
|
|
|
def infer_base_model_from_header(file_path: str) -> str:
|
|
"""从 safetensors 头文件的张量键名推断底层 Base Model (用于脱机/HuggingFace 兼容)"""
|
|
try:
|
|
with open(file_path, "rb") as f:
|
|
header_size_bytes = f.read(8)
|
|
if len(header_size_bytes) < 8: return 'Unknown'
|
|
header_size = struct.unpack('<Q', header_size_bytes)[0]
|
|
if header_size > 100 * 1024 * 1024: return 'Unknown'
|
|
|
|
header_json = json.loads(f.read(header_size).decode('utf-8'))
|
|
|
|
# 1. 尝试从 __metadata__ 提取
|
|
metadata = header_json.get('__metadata__', {})
|
|
arch = metadata.get('modelspec.architecture', '')
|
|
if 'stable-diffusion-xl' in arch.lower(): return 'SDXL'
|
|
if 'stable-diffusion-v1' in arch.lower() or 'runwayml/stable-diffusion-v1-5' in arch.lower(): return 'SD 1.5'
|
|
if 'flux' in arch.lower(): return 'Flux.1 D'
|
|
if 'sd3' in arch.lower(): return 'SD3'
|
|
|
|
# 2. 暴力张量键名指纹匹配 (Tensor Fingerprinting)
|
|
# 把前 500 个键拼接成字符串以提高检索效率，大部分核心键都在前面
|
|
keys_str = " ".join(list(header_json.keys())[:500])
|
|
|
|
# Flux 指纹
|
|
if 'double_blocks.0.img_attn' in keys_str or 'img_in.weight' in keys_str: return 'Flux.1 D'
|
|
# SD3 指纹
|
|
if 'joint_blocks.0.x_block' in keys_str: return 'SD3'
|
|
# SDXL 指纹 (包含两套 text encoder)
|
|
if 'conditioner.embedders.1.model' in keys_str or 'label_emb.0.0.weight' in keys_str: return 'SDXL'
|
|
# SD 1.5 指纹
|
|
if 'cond_stage_model.transformer.text_model' in keys_str or 'model.diffusion_model.input_blocks.0.0.weight' in keys_str: return 'SD 1.5'
|
|
|
|
return 'Unknown'
|
|
except Exception as e:
|
|
print(f"[-] 离线底模推断失败: {e}")
|
|
return 'Unknown'
|
|
|
|
def sanitize_filename(name: str) -> str:
|
|
"""清理文件名中的非法字符"""
|
|
name = re.sub(r'[\r\n\t]+', ' ', name)
|
|
name = re.sub(r'[\\/*?:"<>|#]', "", name)
|
|
return name.strip(' .')
|
|
|
|
CIVITAI_DOWN_AFTER = 2 # models in a row Civitai did not answer for before the rest of the run stays offline
|
|
|
|
|
|
def write_report(args, **entry):
|
|
"""One line of the scan report (--report-file): a model's outcome, or an event of the run."""
|
|
if not args.report_file:
|
|
return
|
|
try:
|
|
with open(args.report_file, 'a', encoding='utf-8') as f:
|
|
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
|
except OSError as e:
|
|
print(f"[-] 扫描记录写入失败: {e}")
|
|
|
|
|
|
def local_record(file_path, root, file_hash, reason):
|
|
"""What the file alone tells (id -1), with why Civitai's record is missing."""
|
|
inferred_base = infer_base_model_from_header(file_path)
|
|
stem = os.path.splitext(os.path.basename(file_path))[0]
|
|
print(f"[*] 使用本地哈希重建基础元数据 ({os.path.basename(file_path)})")
|
|
return {
|
|
"id": -1,
|
|
"modelId": -1,
|
|
"name": stem,
|
|
"baseModel": "" if inferred_base == 'Unknown' else inferred_base,
|
|
"description": "<p>Automatically inferred by Anomalous Local Engine.</p>",
|
|
"model": {"name": stem, "type": "LORA" if "lora" in root.lower() else "Checkpoint"},
|
|
"files": [{"hashes": {"SHA256": file_hash}}],
|
|
"anomalous_unmatched_reason": reason,
|
|
}
|
|
|
|
|
|
def ask_civitai(args, state, file_hash):
|
|
"""(Civitai's record or None, why it is missing: "not_found" | "network" | "offline" | "")."""
|
|
if args.offline_only:
|
|
return None, "offline"
|
|
if state["civitai_down"]:
|
|
return None, "network"
|
|
try:
|
|
data = fetch_civitai_info(file_hash)
|
|
except CivitaiUnreachable:
|
|
state["network_failures"] += 1
|
|
if state["network_failures"] >= CIVITAI_DOWN_AFTER:
|
|
state["civitai_down"] = True
|
|
print("\033[93m[!] 连续几个模型都连不上 Civitai，本次其余模型改为从文件推断，之后可以“重新联网查找”。\033[0m")
|
|
write_report(args, event="civitai_down")
|
|
return None, "network"
|
|
state["network_failures"] = 0
|
|
return data, ("" if data else "not_found")
|
|
|
|
|
|
def scan_one(args, state, rename_log, root, filename, previous):
|
|
"""Scans one model; returns its report entry, or None when nothing about it changed."""
|
|
file_path = os.path.join(root, filename)
|
|
old_base = os.path.splitext(file_path)[0]
|
|
online = not args.offline_only and not state["civitai_down"]
|
|
|
|
# A .civitai.info holding only edits from older versions is not a scan result.
|
|
info_exists = scan_info_path(old_base) is not None
|
|
# Models Civitai was never asked about (an offline scan, no connection then) are looked up
|
|
# by every online scan; models Civitai did not know wait for "look up again".
|
|
unasked = is_unmatched(previous) and unmatched_reason(previous) in ("offline", "network")
|
|
if args.force_overwrite or args.retry_unmatched or (online and unasked):
|
|
info_exists = False
|
|
|
|
preview_exists = args.skip_media or any(os.path.exists(old_base + ext) for ext in COVER_SUFFIXES)
|
|
needs_rename = bool(args.virtual_rename or (not args.skip_rename and args.physical_rename))
|
|
if info_exists and preview_exists and not needs_rename:
|
|
print(f"[*] 已跳过 (信息满足要求): {filename}")
|
|
return None
|
|
|
|
print(f"\n---> 处理文件: {filename} (位于 {root})")
|
|
civitai_data = None
|
|
file_hash = None
|
|
reason = ""
|
|
if info_exists: # a rename, or a missing cover fetched from the saved record: no new lookup
|
|
civitai_data = read_local_info(old_base)
|
|
if civitai_data:
|
|
print(f"[*] 使用本地已有的信息")
|
|
|
|
if not civitai_data:
|
|
if is_unmatched(previous): # the hash from last time, if the file is unchanged
|
|
file_hash = sidecar_file_hash(previous, file_path, (previous.get("files") or [{}])[0])[0]
|
|
file_hash = file_hash or calculate_sha256(file_path)
|
|
civitai_data, reason = ask_civitai(args, state, file_hash)
|
|
|
|
# What was read from disk is not fetched again: only fresh Civitai data needs the model
|
|
# page's description and the cover.
|
|
fresh = civitai_data is not None and file_hash is not None
|
|
if not civitai_data:
|
|
if args.skip_local_metadata:
|
|
print(f"[*] Civitai 获取失败，且禁用了本地元数据解析。跳过 {filename}")
|
|
return {"status": "failed", "error": "no_local_metadata", "reason": reason}
|
|
civitai_data = local_record(file_path, root, file_hash, reason)
|
|
entry = {"status": "matched" if fresh else "inferred" if reason else "unchanged"}
|
|
if reason:
|
|
entry["reason"] = reason
|
|
|
|
model_id = civitai_data.get("modelId")
|
|
if fresh and model_id and model_id != -1 and not state["civitai_down"]:
|
|
description = fetch_model_description(model_id)
|
|
if description:
|
|
civitai_data["description"] = description
|
|
if not isinstance(civitai_data.get("model"), dict):
|
|
civitai_data["model"] = {}
|
|
civitai_data["model"]["description"] = description
|
|
|
|
model_name = sanitize_filename(civitai_data.get("model", {}).get("name", "UnknownModel"))
|
|
version_name = sanitize_filename(civitai_data.get("name", "UnknownVersion"))
|
|
if fresh:
|
|
entry["name"] = f"{civitai_data.get('model', {}).get('name', '')} · {civitai_data.get('name', '')}".strip(" ·")
|
|
entry["base"] = civitai_data.get("baseModel") or ""
|
|
new_filename = f"{model_name}_{version_name}.safetensors"
|
|
new_file_path = os.path.join(root, new_filename)
|
|
new_base = os.path.splitext(new_file_path)[0]
|
|
|
|
info_data = civitai_data
|
|
if file_hash:
|
|
info_data["anomalous_file_identity"] = computed_file_identity(file_path, file_hash)
|
|
# Civitai's name for the card; a name set in the model editor lives in
|
|
# <model>.anomalous.json and is shown instead.
|
|
if args.virtual_rename:
|
|
info_data["anomalous_custom_name"] = f"{model_name}_{version_name}"
|
|
if not args.dry_run:
|
|
with open(old_base + ".info", 'w', encoding='utf-8') as f:
|
|
json.dump(info_data, f, ensure_ascii=True, indent=4)
|
|
print(f"[+] 写入纯净版 Civitai 描述 -> .info")
|
|
else:
|
|
print(f"[Dry-Run] 拟生成标准描述信息 -> .info")
|
|
|
|
media_url = None
|
|
if not args.skip_media:
|
|
media_url = next((img.get("url") for img in civitai_data.get("images") or [] if img.get("url")), None)
|
|
civitai_covers, user_covers = cover_owners(old_base)
|
|
# Download when the data is fresh, or when the model has no cover at all.
|
|
if media_url and not state["civitai_down"] and (fresh or not (civitai_covers or user_covers)):
|
|
if not args.dry_run:
|
|
print(f"[*] 正在下载预览媒体...")
|
|
saved_path = download_media(media_url, old_base + ".civitai_bak")
|
|
if saved_path:
|
|
print(f"[+] 媒体下载成功 -> {os.path.basename(saved_path)}")
|
|
entry["cover"] = "kept" if user_covers else "new"
|
|
if user_covers:
|
|
# The user's cover always stays; Civitai's image is kept as the backup.
|
|
print(f"[*] 保留你自己的封面，Civitai 封面存为备份")
|
|
else:
|
|
ext = os.path.splitext(saved_path)[1]
|
|
preview_path = old_base + (ext if ext.startswith('.preview.') else f".preview{ext}")
|
|
# Earlier copies of Civitai's image under another name give way to the new one.
|
|
older = [c for c in civitai_covers if c != preview_path]
|
|
try:
|
|
if older:
|
|
move_to_trash(*older)
|
|
except Exception as e:
|
|
print(f"[-] 旧的 Civitai 封面没有移走: {e}")
|
|
shutil.copy2(saved_path, preview_path)
|
|
else:
|
|
print(f"[Dry-Run] 拟下载预览媒体...")
|
|
|
|
if args.skip_rename or not args.physical_rename:
|
|
print(f"[*] 物理重命名已跳过。仅保存 .info 及其可能包含的虚拟重命名。")
|
|
elif file_path != new_file_path and new_filename != filename:
|
|
if os.path.exists(new_file_path):
|
|
print(f"[*] 目标文件名已存在，正在验证内容是否完全相同: {filename}")
|
|
try:
|
|
files_identical = os.path.getsize(file_path) == os.path.getsize(new_file_path)
|
|
if files_identical:
|
|
files_identical = calculate_sha256(file_path) == calculate_sha256(new_file_path)
|
|
except OSError:
|
|
files_identical = False
|
|
|
|
if not files_identical:
|
|
print(f"\033[91m[-] 目标名称冲突但文件内容不同，已保留两个模型并跳过重命名: {filename}\033[0m")
|
|
return {**entry, "status": "failed", "error": "name_conflict", "renamed_to": new_file_path}
|
|
if not args.dry_run:
|
|
try:
|
|
print(f"[*] Hash 一致，把已确认的重复副本移到回收站: {filename}")
|
|
move_to_trash(file_path, *[old_base + ext for ext in SIDECAR_SUFFIXES])
|
|
entry["duplicate_of"] = new_file_path
|
|
except Exception as e:
|
|
print(f"[-] 重复副本没有移走，两份都保留 (可能文件被占用或这个盘没有回收站): {e}")
|
|
return {**entry, "status": "failed", "error": "duplicate_kept"}
|
|
else:
|
|
print(f"[Dry-Run] Hash 一致，拟删除重复副本及其附属文件: {filename}")
|
|
elif not args.dry_run:
|
|
os.rename(file_path, new_file_path)
|
|
rename_log[file_path] = new_file_path
|
|
for ext in SIDECAR_SUFFIXES:
|
|
if os.path.exists(old_base + ext):
|
|
os.replace(old_base + ext, new_base + ext)
|
|
print(f"[+] 物理重命名完成: {filename} ==> {new_filename}")
|
|
entry["renamed_to"] = new_file_path
|
|
else:
|
|
print(f"[Dry-Run] 拟物理重命名文件: {filename} ==> {new_filename}")
|
|
print(f"[Dry-Run] 拟连带重命名附属文件 (.info / .png 等)")
|
|
else:
|
|
print("[*] 文件名已符合规范，无需重命名。")
|
|
changed = entry["status"] != "unchanged" or set(entry) & {"cover", "renamed_to", "duplicate_of"}
|
|
return entry if changed else None
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="ComfyUI 模型 Civitai 嗅探与重命名工具")
|
|
parser.add_argument("folder", help="要扫描的文件夹路径 (例如: models/checkpoints)")
|
|
parser.add_argument("--dry-run", action="store_true", help="空跑模式，仅打印将要执行的操作，不修改任何文件")
|
|
parser.add_argument("--undo", action="store_true", help="根据 backup_rename_log.json 恢复文件名")
|
|
parser.add_argument("--skip-rename", action="store_true", help="只下载信息文件，不重命名主文件")
|
|
parser.add_argument("--virtual-rename", action="store_true", help="虚拟重命名：修改 JSON 注入标准名称，不修改底层物理文件名")
|
|
parser.add_argument("--physical-rename", action="store_true", help="物理重命名：真实修改底层的 safetensors 及其附属文件名")
|
|
parser.add_argument("--skip-media", action="store_true", help="不下载预览图或视频")
|
|
parser.add_argument("--offline-only", action="store_true", help="跳过 Civitai 联网获取，强制使用本地脱机张量推断提取 Base Model")
|
|
parser.add_argument("--force-overwrite", action="store_true", help="强制覆盖已存在的信息文件")
|
|
parser.add_argument("--retry-unmatched", action="store_true", help="只处理之前没在 Civitai 匹配到的模型，重新联网查找（文件没变时复用哈希，不替换封面）")
|
|
parser.add_argument("--skip-local-metadata", action="store_true", help="忽略本地已有的.info / .json文件")
|
|
parser.add_argument("--target-files", type=str, default="", help="仅扫描逗号分隔的具体文件(相对路径)")
|
|
parser.add_argument("--folder-type", default="", help="由 ComfyUI 传入的模型目录类型，用于执行安全策略")
|
|
parser.add_argument("--progress-file", default="", help=argparse.SUPPRESS)
|
|
parser.add_argument("--targets-file", default="", help=argparse.SUPPRESS) # JSON list of file names
|
|
parser.add_argument("--report-file", default="", help=argparse.SUPPRESS) # one JSON line per changed model
|
|
parser.add_argument("--civitai-down", action="store_true", help=argparse.SUPPRESS) # an earlier folder could not reach it
|
|
args = parser.parse_args()
|
|
if args.offline_only:
|
|
# Offline means no network at all: previews and model pages come from Civitai.
|
|
args.skip_media = True
|
|
|
|
target_folder = args.folder
|
|
if not os.path.isdir(target_folder):
|
|
print(f"[-] 错误: 文件夹不存在 -> {target_folder}")
|
|
sys.exit(1)
|
|
|
|
if args.physical_rename and is_physical_rename_protected(args.folder_type, target_folder):
|
|
print("[*] 基础组件目录受保护：VAE、CLIP/Text Encoder 与 CLIP Vision 不执行物理重命名。")
|
|
args.physical_rename = False
|
|
|
|
backup_log_path = os.path.join(target_folder, "backup_rename_log.json")
|
|
|
|
# ==========================
|
|
# 模式一：Undo 回滚模式
|
|
# ==========================
|
|
if args.undo:
|
|
if not os.path.exists(backup_log_path):
|
|
print("[-] 未找到备份日志 backup_rename_log.json，无法撤销。")
|
|
sys.exit(1)
|
|
|
|
with open(backup_log_path, 'r', encoding='utf-8') as f:
|
|
log_data = json.load(f)
|
|
|
|
print("[*] 开始回滚文件名...")
|
|
for old_path, new_path in log_data.items():
|
|
if os.path.exists(new_path):
|
|
print(f"[*] 恢复主文件: {os.path.basename(new_path)} -> {os.path.basename(old_path)}")
|
|
if not args.dry_run:
|
|
os.replace(new_path, old_path)
|
|
else:
|
|
print(f"[-] 找不到被重命名的文件: {new_path}")
|
|
|
|
old_base = os.path.splitext(old_path)[0]
|
|
new_base = os.path.splitext(new_path)[0]
|
|
|
|
for ext in SIDECAR_SUFFIXES:
|
|
new_ext_path = new_base + ext
|
|
old_ext_path = old_base + ext
|
|
if os.path.exists(new_ext_path):
|
|
print(f"[*] 恢复配套文件: {os.path.basename(new_ext_path)} -> {os.path.basename(old_ext_path)}")
|
|
if not args.dry_run:
|
|
os.replace(new_ext_path, old_ext_path)
|
|
|
|
print("[+] 回滚完成！")
|
|
sys.exit(0)
|
|
|
|
# ==========================
|
|
# 模式二：正常嗅探与重命名
|
|
# ==========================
|
|
rename_log = {}
|
|
if os.path.exists(backup_log_path):
|
|
with open(backup_log_path, 'r', encoding='utf-8') as f:
|
|
rename_log = json.load(f)
|
|
|
|
target_files_basenames = []
|
|
for targets_path in (os.path.join(target_folder, '.scan_targets.json'), args.targets_file):
|
|
if not targets_path or not os.path.exists(targets_path):
|
|
continue
|
|
try:
|
|
with open(targets_path, 'r', encoding='utf-8') as f:
|
|
target_files_basenames += [os.path.basename(t.strip()) for t in json.load(f)]
|
|
if targets_path != args.targets_file: # the scanner removes its own file
|
|
os.remove(targets_path)
|
|
except (OSError, ValueError, AttributeError):
|
|
pass
|
|
if args.target_files:
|
|
target_files_basenames.extend([os.path.basename(t.strip()) for t in args.target_files.split(',')])
|
|
|
|
print(f"[*] 开始扫描文件夹: {target_folder}")
|
|
if args.dry_run:
|
|
print("==================================================")
|
|
print("[警告]: 当前处于 Dry-Run (空跑) 模式，不会修改系统中的任何文件！")
|
|
print("==================================================")
|
|
if not CIVITAI_API_KEY and not args.offline_only:
|
|
print("[!] 未配置 Civitai API Key。部分限制级模型或将无法获取图片。")
|
|
|
|
write_scan_progress(args.progress_file, "enumerating")
|
|
total_files = count_scan_files(target_folder, target_files_basenames, args.retry_unmatched)
|
|
write_scan_progress(args.progress_file, "scanning", total_files)
|
|
current_file = 0
|
|
last_progress_write = 0.0
|
|
state = {"network_failures": 0, "civitai_down": args.civitai_down}
|
|
counts = {"success": 0, "fail": 0, "unchanged": 0}
|
|
|
|
for root, _, files in os.walk(target_folder):
|
|
for filename in files:
|
|
if not filename.endswith(".safetensors"):
|
|
continue
|
|
if target_files_basenames and filename not in target_files_basenames:
|
|
continue
|
|
previous = read_local_info(os.path.join(root, os.path.splitext(filename)[0]))
|
|
if args.retry_unmatched and not is_unmatched(previous):
|
|
continue
|
|
|
|
current_file += 1
|
|
now = time.monotonic()
|
|
if current_file == 1 or current_file == total_files or now - last_progress_write >= 0.2:
|
|
write_scan_progress(args.progress_file, "scanning", total_files, current_file, filename)
|
|
last_progress_write = now
|
|
|
|
try:
|
|
entry = scan_one(args, state, rename_log, root, filename, previous)
|
|
except Exception as e: # one model's trouble does not end the scan of the others
|
|
print(f"\033[91m[-] 处理失败，已跳过: {filename}: {e}\033[0m")
|
|
entry = {"status": "failed", "error": str(e)[:300]}
|
|
if entry is None:
|
|
counts["unchanged"] += 1
|
|
continue
|
|
counts["fail" if entry["status"] == "failed" else "success"] += 1
|
|
write_report(args, file=os.path.join(root, filename), **entry)
|
|
|
|
# Save scan results
|
|
write_scan_progress(args.progress_file, "complete", total_files, total_files)
|
|
write_report(args, event="done", unchanged=counts["unchanged"], civitai_down=state["civitai_down"])
|
|
if not args.dry_run:
|
|
result_path = os.path.join(target_folder, ".scan_result.json")
|
|
try:
|
|
with open(result_path, 'w', encoding='utf-8') as f:
|
|
json.dump({"success": counts["success"], "fail": counts["fail"]}, f)
|
|
except Exception as e:
|
|
print(f"[-] 保存统计结果失败: {e}")
|
|
|
|
if not args.dry_run and rename_log:
|
|
with open(backup_log_path, 'w', encoding='utf-8') as f:
|
|
json.dump(rename_log, f, ensure_ascii=True, indent=4)
|
|
print(f"\n[+] 重命名映射日志已保存至: {backup_log_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|