From 57bb9dac8ea77ec462a5cbac5d6375ff1409b61e Mon Sep 17 00:00:00 2001 From: DemonGatanjieu <1027066412@qq.com> Date: Sun, 12 Jul 2026 22:00:17 +0800 Subject: [PATCH] Delete api.py --- api.py | 988 --------------------------------------------------------- 1 file changed, 988 deletions(-) delete mode 100644 api.py diff --git a/api.py b/api.py deleted file mode 100644 index 9f6bda4..0000000 --- a/api.py +++ /dev/null @@ -1,988 +0,0 @@ -import os -import sys -import json -import urllib.parse -import subprocess -import threading -from aiohttp import web -import folder_paths - -def get_metadata(file_path): - base_path = os.path.splitext(file_path)[0] - metadata = { - "name": os.path.basename(base_path), - "description": "", - "notes": "", - "trainedWords": [], - "baseModel": "", - "civitai_url": "", - "hash": "" - } - - info_files = [f"{base_path}.civitai.info", f"{base_path}.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", "") - name = f"{model_name} - {name}".strip(' -') - - 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 = data.get("modelId", "") - version_id = data.get("id", "") - civitai_url = "" - if model_id: - # 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: - civitai_url += f"?modelVersionId={version_id}" - - - hash_val = "" - for file_info in data.get("files", []): - if isinstance(file_info, dict): - hashes = file_info.get("hashes", {}) - if isinstance(hashes, dict) and "SHA256" in hashes: - hash_val = hashes["SHA256"] - break - - metadata["name"] = name if name else metadata["name"] - metadata["description"] = description - metadata["notes"] = notes - metadata["trainedWords"] = trained_words - metadata["baseModel"] = base_model - metadata["civitai_url"] = civitai_url - metadata["hash"] = hash_val - break - except Exception: - pass - - return metadata - - -async def api_get_folders(request): - types = ['checkpoints', 'loras', 'unet', 'diffusion_models'] - result = [] - seen_dirs = set() - - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - if not paths: - continue - - for path_idx, base_dir in enumerate(paths): - if not os.path.exists(base_dir): - continue - real_path = os.path.realpath(base_dir) - if real_path in seen_dirs: - continue - seen_dirs.add(real_path) - - tree = {} - for root, dirs, files in os.walk(base_dir): - has_models = any(f.endswith(('.safetensors', '.ckpt', '.pt')) for f in files) - rel = os.path.relpath(root, base_dir) - if rel == '.': - rel = '/' - else: - rel = '/' + rel.replace('\\', '/') - tree[rel] = { - "path": rel, - "name": os.path.basename(root) if rel != '/' else '[Root]', - "has_models": has_models, - "model_count": sum(1 for f in files if f.endswith(('.safetensors', '.ckpt', '.pt'))) - } - - label = t.capitalize() - if path_idx > 0: - label += f" ({os.path.basename(base_dir)})" - - result.append({ - "type": t, - "path_idx": path_idx, - "label": label, - "folders": tree - }) - - return web.json_response({"folders": result}) - - -async def api_get_models(request): - folder_type = request.query.get('type', 'checkpoints') - subfolder = request.query.get('subfolder', '/') - try: - path_idx = int(request.query.get('path_idx', 0)) - except: - path_idx = 0 - - try: - paths = folder_paths.get_folder_paths(folder_type) - except Exception: - return web.json_response({"models": []}) - if not paths or path_idx >= len(paths): - return web.json_response({"models": []}) - - if '..' in subfolder: - return web.Response(status=400, text='Invalid subfolder') - - base_dir = paths[path_idx] - if subfolder == '/': - target_dir = base_dir - rel_subfolder = "" - else: - rel_subfolder = subfolder.strip('/') - target_dir = os.path.join(base_dir, rel_subfolder) - - if not os.path.exists(target_dir): - return web.json_response({"models": []}) - - models = [] - try: - entries = os.listdir(target_dir) - except Exception: - return web.json_response({"models": []}) - - for f in sorted(entries): - if f.endswith('.safetensors') or f.endswith('.ckpt') or f.endswith('.pt'): - file_path = os.path.join(target_dir, f) - if not os.path.isfile(file_path): - continue - - meta = get_metadata(file_path) - - base_name = os.path.splitext(f)[0] - preview_file = None - for ext in ['.preview.png', '.png', '.jpg', '.jpeg', '.webp', '.mp4', '.webm']: - if os.path.exists(os.path.join(target_dir, base_name + ext)): - preview_file = base_name + ext - break - - preview_url = "" - if preview_file: - q_type = urllib.parse.quote(folder_type) - q_idx = str(path_idx) - q_sub = urllib.parse.quote(rel_subfolder) - q_file = urllib.parse.quote(preview_file) - preview_url = f"/anomalous/image?type={q_type}&path_idx={q_idx}&subfolder={q_sub}&filename={q_file}" - - try: - size_bytes = os.path.getsize(file_path) - size_mb = round(size_bytes / (1024 * 1024), 1) - except Exception: - size_mb = 0 - size_bytes = 0 - - models.append({ - "filename": f, - "size_mb": size_mb, - "size_bytes": size_bytes, - "metadata": meta, - "preview_url": preview_url - }) - - return web.json_response({"models": models}) - - -async def api_serve_image(request): - """Dedicated image serving endpoint for model preview images.""" - folder_type = request.query.get('type', 'checkpoints') - try: - path_idx = int(request.query.get('path_idx', 0)) - except: - path_idx = 0 - subfolder = request.query.get('subfolder', '') - filename = request.query.get('filename', '') - - if not filename or '..' in filename or '..' in subfolder: - return web.Response(status=400, text='Invalid request') - - try: - paths = folder_paths.get_folder_paths(folder_type) - except Exception: - return web.Response(status=404, text='Folder type not found') - if not paths or path_idx >= len(paths): - return web.Response(status=404, text='Folder type not found') - - base_dir = paths[path_idx] - if subfolder: - file_path = os.path.join(base_dir, subfolder, filename) - else: - file_path = os.path.join(base_dir, filename) - - if not os.path.exists(file_path) or not os.path.isfile(file_path): - return web.Response(status=404, text='Image not found') - - ext = os.path.splitext(filename)[1].lower() - content_types = { - '.png': 'image/png', - '.jpg': 'image/jpeg', - '.jpeg': 'image/jpeg', - '.webp': 'image/webp', - '.gif': 'image/gif' - } - content_type = content_types.get(ext, 'application/octet-stream') - - return web.FileResponse(file_path, headers={'Content-Type': content_type}) - - - -async def api_scan_status(request): - folder_type = request.query.get('type', 'checkpoints') - subfolder = request.query.get('subfolder', '/') - try: - path_idx = int(request.query.get('path_idx', 0)) - except: - path_idx = 0 - try: - paths = folder_paths.get_folder_paths(folder_type) - except Exception: - return web.json_response({"scanning": False}) - if not paths or path_idx >= len(paths) or '..' in subfolder: - return web.json_response({"scanning": False}) - - base_dir = paths[path_idx] - if subfolder == '/': - target_dir = base_dir - else: - target_dir = os.path.join(base_dir, subfolder.strip('/')) - - marker_file = os.path.join(target_dir, '.scan_in_progress') - result_file = os.path.join(target_dir, '.scan_result.json') - scanning = os.path.exists(marker_file) - - data = {"scanning": scanning} - if not scanning and os.path.exists(result_file): - try: - with open(result_file, 'r', encoding='utf-8') as f: - data["result"] = __import__('json').load(f) - os.remove(result_file) - except: - pass - - return web.json_response(data) - -async def api_scan_folder(request): - """Launches the scraper in the background for a specific directory.""" - folder_type = request.query.get('type', 'checkpoints') - subfolder = request.query.get('subfolder', '/') - try: - path_idx = int(request.query.get('path_idx', 0)) - except: - path_idx = 0 - - if '..' in subfolder: - return web.json_response({"status": "error", "message": "Invalid subfolder"}) - - try: - paths = folder_paths.get_folder_paths(folder_type) - except Exception: - return web.json_response({"status": "error", "message": "Invalid folder type"}) - - if not paths or path_idx >= len(paths): - return web.json_response({"status": "error", "message": "Invalid path index"}) - - base_dir = paths[path_idx] - if subfolder == '/': - target_dir = base_dir - else: - target_dir = os.path.join(base_dir, subfolder.strip('/')) - - if not os.path.exists(target_dir): - return web.json_response({"status": "error", "message": "Directory does not exist"}) - - # Get the scraper path relative to this script - script_dir = os.path.dirname(os.path.abspath(__file__)) - scraper_path = os.path.join(script_dir, "scraper.py") - - if not os.path.exists(scraper_path): - return web.json_response({"status": "error", "message": "scraper.py not found in extension directory"}) - - print(f"[Anomalous Browser] Starting background scan for: {target_dir}") - - try: - marker_file = os.path.join(target_dir, '.scan_in_progress') - with open(marker_file, 'w') as f: f.write('1') - def run_bg(): - try: - subprocess.run( - [sys.executable, scraper_path, target_dir], - cwd=script_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL - ) - finally: - if hasattr(folder_paths, "filename_list_cache"): - folder_paths.filename_list_cache.clear() - if hasattr(folder_paths, "cache_helper") and hasattr(folder_paths.cache_helper, "clear"): - folder_paths.cache_helper.clear() - - if os.path.exists(marker_file): - try: os.remove(marker_file) - except: pass - - threading.Thread(target=run_bg, daemon=True).start() - - return web.json_response({"status": "ok", "message": "Scan started in background. Check console for details."}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - -async def api_scan_all(request): - script_dir = os.path.dirname(os.path.abspath(__file__)) - scraper_path = os.path.join(script_dir, "scraper.py") - marker_file = os.path.join(script_dir, '.global_scan_in_progress') - - if os.path.exists(marker_file): - return web.json_response({"status": "error", "message": "Global scan already in progress"}) - - try: - with open(marker_file, 'w') as f: f.write('1') - - def run_global_bg(): - try: - types = ['checkpoints', 'loras', 'unet', 'diffusion_models'] - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - if not paths: continue - for base_dir in paths: - if not os.path.exists(base_dir): continue - print(f"[Anomalous Browser] Global scan processing: {base_dir}") - subprocess.run( - [sys.executable, scraper_path, base_dir, "--skip-rename", "--skip-media"], - cwd=script_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL - ) - except Exception as e: - print(f"[Anomalous Browser] Global scan error on {t}: {e}") - finally: - if hasattr(folder_paths, "filename_list_cache"): - folder_paths.filename_list_cache.clear() - if hasattr(folder_paths, "cache_helper") and hasattr(folder_paths.cache_helper, "clear"): - folder_paths.cache_helper.clear() - - if os.path.exists(marker_file): - try: os.remove(marker_file) - except: pass - - threading.Thread(target=run_global_bg, daemon=True).start() - return web.json_response({"status": "ok", "message": "Global scan started"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - -async def api_global_scan_status(request): - script_dir = os.path.dirname(os.path.abspath(__file__)) - marker_file = os.path.join(script_dir, '.global_scan_in_progress') - return web.json_response({"scanning": os.path.exists(marker_file)}) -async def api_clear_cache(request): - try: - if hasattr(folder_paths, "filename_list_cache"): - folder_paths.filename_list_cache.clear() - if hasattr(folder_paths, "cache_helper") and hasattr(folder_paths.cache_helper, "clear"): - folder_paths.cache_helper.clear() - return web.json_response({"status": "success"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - -async def api_get_config(request): - script_dir = os.path.dirname(os.path.abspath(__file__)) - config_path = os.path.join(script_dir, "config.json") - has_key = False - if os.path.exists(config_path): - try: - with open(config_path, 'r', encoding='utf-8') as f: - cfg = json.load(f) - has_key = bool(cfg.get("CIVITAI_API_KEY", "").strip()) - except: - pass - return web.json_response({"has_api_key": has_key}) - -async def api_save_config(request): - try: - data = await request.json() - api_key = data.get("api_key", "").strip() - script_dir = os.path.dirname(os.path.abspath(__file__)) - config_path = os.path.join(script_dir, "config.json") - - cfg = {} - if os.path.exists(config_path): - try: - with open(config_path, 'r', encoding='utf-8') as f: - cfg = json.load(f) - except: - pass - - cfg["CIVITAI_API_KEY"] = api_key - with open(config_path, 'w', encoding='utf-8') as f: - json.dump(cfg, f, indent=4) - - return web.json_response({"status": "ok"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - - -async def api_delete_model(request): - try: - data = await request.json() - folder_type = data.get('type', 'checkpoints') - subfolder = data.get('subfolder', '/') - filename = data.get('filename', '') - try: - path_idx = int(data.get('path_idx', 0)) - except: - path_idx = 0 - - if not filename or '..' in filename or '..' in subfolder: - return web.json_response({"status": "error", "message": "Invalid request parameters"}) - - try: - paths = folder_paths.get_folder_paths(folder_type) - except Exception: - return web.json_response({"status": "error", "message": "Invalid folder type"}) - - if not paths or path_idx >= len(paths): - return web.json_response({"status": "error", "message": "Invalid path index"}) - - base_dir = paths[path_idx] - if subfolder == '/': - target_dir = base_dir - else: - target_dir = os.path.join(base_dir, subfolder.strip('/')) - - model_path = os.path.join(target_dir, filename) - if not os.path.exists(model_path): - return web.json_response({"status": "error", "message": "Model file not found"}) - - # 1. 优先尝试删除你点击的主模型文件 - try: - os.remove(model_path) - except Exception as e: - error_msg = str(e) - if "being used" in error_msg or "WinError 32" in error_msg or "Permission" in error_msg: - error_msg = "文件被占用 (正在被 ComfyUI 使用)。请先重启 ComfyUI 或在工作流中卸载该模型后再删除！" - return web.json_response({"status": "error", "message": f"主模型删除失败: {error_msg}"}) - - base_name = os.path.splitext(filename)[0] - - # 2. 主模型成功删除后，再清理配套的垃圾文件 - associated_exts = [ - '.safetensors', '.ckpt', '.pt', '.bin', - '.info', '.civitai.info', - '.png', '.preview.png', '.jpg', '.jpeg', '.webp', '.gif', - '.json', '.txt', '.yaml' - ] - - deleted_files = [filename] - for ext in associated_exts: - file_to_del = os.path.join(target_dir, base_name + ext) - if file_to_del == model_path: - continue - if os.path.exists(file_to_del) and os.path.isfile(file_to_del): - try: - os.remove(file_to_del) - deleted_files.append(base_name + ext) - except Exception as e: - print(f"[Anomalous Browser] Warning: Failed to delete {file_to_del}: {e}") - - # 3. 修正：前端期待的成功状态是 "success" 而不是 "ok" - return web.json_response({"status": "success", "deleted": deleted_files}) - - except Exception as e: - import traceback - traceback.print_exc() - return web.json_response({"status": "error", "message": str(e)}, status=500) - -async def api_clean_civitai_info(request): - try: - types = ['checkpoints', 'loras', 'unet', 'diffusion_models'] - deleted_count = 0 - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - if not paths: continue - - for base_dir in paths: - if not os.path.exists(base_dir): continue - for root, dirs, files in os.walk(base_dir): - for file in files: - if file.endswith('.civitai.info'): - file_path = os.path.join(root, file) - try: - os.remove(file_path) - deleted_count += 1 - except Exception as e: - pass - - return web.json_response({"status": "success", "count": deleted_count}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}, status=500) - -async def api_compatible_models(request): - base_model = request.query.get('base_model', '') - target_type = request.query.get('target_type', 'loras') - - if not base_model: - return web.json_response({"models": []}) - - target_types = [t.strip() for t in target_type.split(',')] - compatible_models = [] - seen_files = set() - - for t in target_types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - - if not paths: - continue - - for path_idx, base_dir in enumerate(paths): - if not os.path.exists(base_dir): - continue - - for root, _, files in os.walk(base_dir): - for f in files: - if f.endswith('.safetensors') or f.endswith('.ckpt') or f.endswith('.pt'): - file_path = os.path.join(root, f) - real_path = os.path.realpath(file_path) - if real_path in seen_files: - continue - seen_files.add(real_path) - - meta = get_metadata(file_path) - m_bm = str(meta.get("baseModel", "")).strip().lower().replace(" ", "") - req_bm = str(base_model).strip().lower().replace(" ", "") - - if req_bm and m_bm and (req_bm in m_bm or m_bm in req_bm): - rel_subfolder = os.path.relpath(root, base_dir) - if rel_subfolder == '.': - rel_subfolder = '/' - else: - rel_subfolder = '/' + rel_subfolder.replace('\\', '/') - - base_name = os.path.splitext(f)[0] - preview_file = None - for ext in ['.preview.png', '.png', '.jpg', '.jpeg', '.webp', '.mp4', '.webm']: - if os.path.exists(os.path.join(root, base_name + ext)): - preview_file = base_name + ext - break - - preview_url = "" - if preview_file: - q_type = urllib.parse.quote(t) - q_idx = str(path_idx) - q_sub = urllib.parse.quote(rel_subfolder.strip('/')) if rel_subfolder != '/' else "" - q_file = urllib.parse.quote(preview_file) - preview_url = f"/anomalous/image?type={q_type}&path_idx={q_idx}&subfolder={q_sub}&filename={q_file}" - - try: - size_bytes = os.path.getsize(file_path) - size_mb = round(size_bytes / (1024 * 1024), 1) - except Exception: - size_mb = 0 - size_bytes = 0 - - compatible_models.append({ - "type": t, - "path_idx": path_idx, - "subfolder": rel_subfolder, - "filename": f, - "size_mb": size_mb, - "size_bytes": size_bytes, - "preview_url": preview_url, - "metadata": meta - }) - - return web.json_response({"models": compatible_models}) - -async def api_get_notebooks(request): - script_dir = os.path.dirname(os.path.abspath(__file__)) - nb_dir = os.path.join(script_dir, "notebooks") - if not os.path.exists(nb_dir): - os.makedirs(nb_dir) - - notebooks = [] - for f in os.listdir(nb_dir): - if f.endswith('.json'): - try: - with open(os.path.join(nb_dir, f), 'r', encoding='utf-8') as file: - data = json.load(file) - notebooks.append({ - "filename": f, - "name": data.get("name", f.replace('.json', '')), - "data": data - }) - except Exception: - pass - return web.json_response({"notebooks": notebooks}) - -async def api_save_notebook(request): - try: - data = await request.json() - filename = data.get("filename", "") - if not filename or '..' in filename: - return web.json_response({"status": "error", "message": "Invalid filename"}) - if not filename.endswith('.json'): - filename += '.json' - - script_dir = os.path.dirname(os.path.abspath(__file__)) - nb_dir = os.path.join(script_dir, "notebooks") - if not os.path.exists(nb_dir): - os.makedirs(nb_dir) - - file_path = os.path.join(nb_dir, filename) - with open(file_path, 'w', encoding='utf-8') as f: - json.dump(data.get("data", {}), f, indent=4, ensure_ascii=False) - - return web.json_response({"status": "success"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - -async def api_delete_notebook(request): - try: - data = await request.json() - filename = data.get("filename", "") - if not filename or '..' in filename: - return web.json_response({"status": "error", "message": "Invalid filename"}) - - script_dir = os.path.dirname(os.path.abspath(__file__)) - nb_dir = os.path.join(script_dir, "notebooks") - file_path = os.path.join(nb_dir, filename) - - if os.path.exists(file_path): - os.remove(file_path) - return web.json_response({"status": "success"}) - return web.json_response({"status": "error", "message": "File not found"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}) - -async def api_translate(request): - try: - data = await request.json() - text = data.get("text", "") - tl = data.get("target_lang", "zh-CN") - if not text: - return web.json_response({"translated": ""}) - - # Try DeepL if API key exists - script_dir = os.path.dirname(os.path.abspath(__file__)) - config_path = os.path.join(script_dir, "config.json") - deepl_key = "" - if os.path.exists(config_path): - try: - with open(config_path, 'r', encoding='utf-8') as f: - cfg = json.load(f) - deepl_key = cfg.get("DEEPL_API_KEY", "").strip() - except: - pass - - if deepl_key: - deepl_map = { "zh-CN": "ZH", "en": "EN", "ja": "JA", "ko": "KO", "fr": "FR", "de": "DE", "es": "ES", "ru": "RU" } - d_tl = deepl_map.get(tl, "EN") - url = "https://api-free.deepl.com/v2/translate" if ":fx" in deepl_key else "https://api.deepl.com/v2/translate" - payload = urllib.parse.urlencode({ - "auth_key": deepl_key, - "text": text, - "target_lang": d_tl - }).encode('utf-8') - req = urllib.request.Request(url, data=payload) - with urllib.request.urlopen(req, timeout=5) as resp: - result = json.loads(resp.read().decode('utf-8')) - return web.json_response({"translated": result["translations"][0]["text"]}) - else: - # Fallback to Google Translate (free, no key) - url = f"https://translate.googleapis.com/translate_a/single?client=gtx&sl=auto&tl={tl}&dt=t&q={urllib.parse.quote(text)}" - req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) - with urllib.request.urlopen(req, timeout=5) as resp: - result = json.loads(resp.read().decode('utf-8')) - # Google Translate returns a list of fragments - translated_text = "".join([part[0] for part in result[0]]) - return web.json_response({"translated": translated_text}) - - except Exception as e: - print(f"Translate Error: {e}") - return web.json_response({"translated": text, "error": str(e)}) - - -async def api_base_models(request): - target_types = ['checkpoints', 'unet', 'diffusion_models', 'loras'] - base_models = set() - seen_files = set() - - for t in target_types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - if not paths: continue - - for base_dir in paths: - if not os.path.exists(base_dir): continue - for root, _, files in os.walk(base_dir): - for f in files: - if f.endswith('.safetensors') or f.endswith('.ckpt') or f.endswith('.pt'): - file_path = os.path.join(root, f) - real_path = os.path.realpath(file_path) - if real_path in seen_files: continue - seen_files.add(real_path) - - meta = get_metadata(file_path) - m_bm = meta.get("baseModel", "") - if m_bm and str(m_bm).strip(): - # Remove typical generic strings that might pollute - clean_bm = str(m_bm).strip() - base_models.add(clean_bm) - - return web.json_response({"base_models": sorted(list(base_models))}) - -async def api_get_gallery_images(request): - try: - output_dir = folder_paths.get_output_directory() - if not os.path.exists(output_dir): - return web.json_response({"images": [], "total": 0, "page": 1, "pages": 0}) - - page = int(request.query.get('page', 1)) - limit = int(request.query.get('limit', 50)) - - valid_exts = {'.png', '.jpg', '.jpeg', '.webp', '.gif'} - images = [] - - for root, dirs, files in os.walk(output_dir): - for f in files: - ext = os.path.splitext(f)[1].lower() - if ext in valid_exts: - rel_path = os.path.relpath(root, output_dir) - subfolder = "" if rel_path == "." else rel_path.replace('\\', '/') - full_path = os.path.join(root, f) - try: - mtime = os.path.getmtime(full_path) - except: - mtime = 0 - images.append({ - "filename": f, - "subfolder": subfolder, - "type": "output", - "mtime": mtime - }) - - # Sort by mtime descending (newest first) - images.sort(key=lambda x: x['mtime'], reverse=True) - - total = len(images) - start_idx = (page - 1) * limit - end_idx = start_idx + limit - paginated_images = images[start_idx:end_idx] - - return web.json_response({ - "images": paginated_images, - "total": total, - "page": page, - "pages": (total + limit - 1) // limit if limit > 0 else 0 - }) - except Exception as e: - return web.json_response({"error": str(e)}, status=500) - -async def api_delete_gallery_image(request): - try: - data = await request.json() - filename = data.get("filename") - subfolder = data.get("subfolder", "") - - if not filename or '..' in filename or '..' in subfolder: - return web.json_response({"status": "error", "message": "Invalid parameters"}) - - output_dir = folder_paths.get_output_directory() - if subfolder: - file_path = os.path.join(output_dir, subfolder, filename) - else: - file_path = os.path.join(output_dir, filename) - - if os.path.exists(file_path): - os.remove(file_path) - return web.json_response({"status": "success"}) - else: - return web.json_response({"status": "error", "message": "File not found"}) - except Exception as e: - return web.json_response({"status": "error", "message": str(e)}, status=500) - -async def api_resolve_hash(request): - target_hash = request.query.get("hash", "").strip().upper() - size_str = request.query.get("size", "").strip() - filename_query = request.query.get("filename", "").strip() - target_size = int(size_str) if size_str.isdigit() else None - - if not target_hash and not target_size and not filename_query: - return web.json_response({"found": False}) - - types = ['checkpoints', 'loras', 'unet', 'diffusion_models'] - - # First pass: try exact hash match in .info files - if target_hash: - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - if not paths: - continue - - for base_dir in paths: - if not os.path.exists(base_dir): - continue - - for root, dirs, files in os.walk(base_dir): - for file in files: - if file.endswith('.safetensors') or file.endswith('.ckpt') or file.endswith('.pt'): - file_path = os.path.join(root, file) - meta = get_metadata(file_path) - if meta.get("hash", "").upper() == target_hash: - # ComfyUI widget expects relative path from folder_paths base_dir - rel_path = os.path.relpath(file_path, base_dir) - if rel_path.startswith('.\\') or rel_path.startswith('./'): - rel_path = rel_path[2:] - rel_path = rel_path.replace('\\', '/') - return web.json_response({ - "found": True, - "type": t, - "filename": rel_path - }) - - # Second pass: Fallback size match - if target_size is not None: - size_matches = [] - seen_realpaths = set() - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - except Exception: - continue - if not paths: - continue - - for base_dir in paths: - if not os.path.exists(base_dir): - continue - - for root, dirs, files in os.walk(base_dir): - for file in files: - if file.endswith('.safetensors') or file.endswith('.ckpt') or file.endswith('.pt'): - file_path = os.path.join(root, file) - try: - if os.path.getsize(file_path) == target_size: - real_p = os.path.realpath(file_path) - if real_p in seen_realpaths: - continue - seen_realpaths.add(real_p) - - rel_path = os.path.relpath(file_path, base_dir) - if rel_path.startswith('.\\') or rel_path.startswith('./'): - rel_path = rel_path[2:] - rel_path = rel_path.replace('\\', '/') - size_matches.append({ - "type": t, - "filename": rel_path - }) - except Exception: - pass - - # If exactly ONE file matches the size, it's a definitive match - if len(size_matches) == 1: - return web.json_response({ - "found": True, - "type": size_matches[0]["type"], - "filename": size_matches[0]["filename"], - "matched_by_size": True - }) - - return web.json_response({"found": False}) - -async def api_get_all_hashes(request): - """ - Returns a dictionary of all scanned models with their hash and size. - Keyed by both relative path and basename for maximum frontend resilience. - """ - import asyncio - - def fetch_all(): - hashes = {} - types = ['checkpoints', 'loras', 'unet', 'diffusion_models'] - for t in types: - try: - paths = folder_paths.get_folder_paths(t) - if not paths: continue - for base_dir in paths: - if not os.path.exists(base_dir): continue - for root, dirs, files in os.walk(base_dir): - for file in files: - if file.endswith('.safetensors') or file.endswith('.ckpt') or file.endswith('.pt'): - file_path = os.path.join(root, file) - try: - size_bytes = os.path.getsize(file_path) - except Exception: - size_bytes = 0 - - meta = get_metadata(file_path) - hash_val = "" - if meta and meta.get("hash"): - hash_val = meta["hash"] - - rel_path = os.path.relpath(file_path, base_dir) - if rel_path.startswith('.\\') or rel_path.startswith('./'): - rel_path = rel_path[2:] - rel_path = rel_path.replace('\\', '/') - basename = os.path.basename(file_path) - - val = {"hash": hash_val, "size": size_bytes} - hashes[rel_path] = val - hashes[basename] = val - except Exception: - pass - return hashes - - hashes = await asyncio.to_thread(fetch_all) - return web.json_response(hashes) - -def setup_routes(app): - app.router.add_get('/anomalous/folders', api_get_folders) - app.router.add_get('/anomalous/models', api_get_models) - app.router.add_get('/anomalous/image', api_serve_image) - app.router.add_post('/anomalous/scan', api_scan_folder) - app.router.add_get('/anomalous/scan_status', api_scan_status) - app.router.add_get('/anomalous/config', api_get_config) - app.router.add_post('/anomalous/save_config', api_save_config) - app.router.add_post('/anomalous/delete_model', api_delete_model) - app.router.add_post('/anomalous/clean_civitai_info', api_clean_civitai_info) - app.router.add_get('/anomalous/compatible_models', api_compatible_models) - app.router.add_get('/anomalous/notebooks', api_get_notebooks) - app.router.add_post('/anomalous/save_notebook', api_save_notebook) - app.router.add_post('/anomalous/delete_notebook', api_delete_notebook) - app.router.add_post('/anomalous/translate', api_translate) - app.router.add_get('/anomalous/base_models', api_base_models) - app.router.add_get('/anomalous/gallery_images', api_get_gallery_images) - app.router.add_post('/anomalous/delete_gallery_image', api_delete_gallery_image) - app.router.add_get('/anomalous/resolve_hash', api_resolve_hash) - app.router.add_get('/anomalous/all_hashes', api_get_all_hashes) - app.router.add_post('/anomalous/scan_all', api_scan_all) - app.router.add_get('/anomalous/global_scan_status', api_global_scan_status)