From e53d9f2a0bdfe17a42f1e596e50c2c5cacf57741 Mon Sep 17 00:00:00 2001 From: Dariusz L Date: Sat, 27 Jun 2026 03:25:31 +0200 Subject: [PATCH] feat: generalize remote metadata lookup across CivitAI, CivArchive, and HuggingFace Replace CivitAI-only search with a multi-source metadata resolution route using SHA256 exact-match validation, dead-link filtering, and live download URL probing. Unify sidecar metadata saving across all remote sources. Add force-refresh support, local metadata import with preview detection, and frontend source-aware UI controls. --- __init__.py | 674 ++++++++++++++++++++--- core/downloader.py | 26 +- core/sources/civarchive.py | 147 +++-- core/sources/civitai.py | 176 +++++- web/resolver/views/model_info_methods.js | 91 ++- 5 files changed, 964 insertions(+), 150 deletions(-) diff --git a/__init__.py b/__init__.py index bf17526..3e3cd5a 100644 --- a/__init__.py +++ b/__init__.py @@ -169,6 +169,7 @@ class ModelResolverExtension: CivArchiveSearchError, is_civarchive_available, search_civarchive_for_file, + resolve_civarchive_by_hash, resolve_civarchive_model_version, get_civarchive_model_details, clear_search_cache as clear_civarchive_search_cache, @@ -1102,17 +1103,20 @@ class ModelResolverExtension: {"loaded_models": loaded_models, "total": len(loaded_models)} ) - # ==================== CIVITAI SEARCH ROUTE ==================== + # ==================== MODEL METADATA LOOKUP ROUTE ==================== @routes.post("/model_resolver/civitai-search") @json_api_endpoint("civitai-search") async def civitai_search(request): - """Search CivitAI for a model using file hash.""" + """Fetch model metadata from trusted exact-match sources.""" data = await request.json() filename = data.get("filename", "") category = data.get("category", "") resolved_path = data.get("resolved_path", "") local_only = to_bool(data.get("local_only"), False) + force_refresh = to_bool( + data.get("force_refresh") or data.get("force"), False + ) provided_hash = ( data.get("sha256") or data.get("hash") @@ -1125,6 +1129,12 @@ class ModelResolverExtension: and all(ch in "0123456789abcdef" for ch in provided_hash) ): provided_hash = "" + hf_token = data.get("hf_token", "") + brave_search_api_key = data.get("brave_search_api_key", "") + hf_use_brave_fallback = to_bool( + data.get("hf_use_brave_fallback", True), + True, + ) if not filename: return web.json_response( @@ -1222,8 +1232,16 @@ class ModelResolverExtension: or "", "metadata_saved": bool(metadata_saved), "location": result.get("location") or file_location, + "source": result.get("source") or "", + "details_source": result.get("details_source") + or result.get("source") + or "", "url": result.get("url"), "version_url": result.get("version_url"), + "download_url": result.get("download_url"), + "platform_url": result.get("platform_url"), + "repo_id": result.get("repo_id"), + "path": result.get("path"), "model_id": result.get("model_id"), "model_name": result.get("model_name") or clean_name, "model_type": model_type, @@ -1240,11 +1258,364 @@ class ModelResolverExtension: "model_description": result.get("model_description", ""), "from_metadata": bool(result.get("from_metadata")), "local_only": bool(local_payload), + "metadata_checked": bool(civitai_checked), "civitai_checked": bool(civitai_checked), } + def normalize_sha256_value(value): + text = str(value or "").strip().lower() + if text.startswith("sha256:"): + text = text.split(":", 1)[1].strip() + if len(text) == 64 and all( + ch in "0123456789abcdef" for ch in text + ): + return text + return "" + + def extract_result_sha256(result): + if not isinstance(result, dict): + return "" + for key in ("sha256", "hash"): + normalized = normalize_sha256_value(result.get(key)) + if normalized: + return normalized + hashes = result.get("hashes") + if isinstance(hashes, dict): + for key in ("SHA256", "sha256", "Sha256"): + normalized = normalize_sha256_value(hashes.get(key)) + if normalized: + return normalized + return "" + + def result_filename_matches(result): + if not isinstance(result, dict): + return False + expected = _os.path.basename(filename or "").lower() + candidates = [ + result.get("filename"), + result.get("path"), + result.get("file_path"), + ] + for candidate in candidates: + basename = _os.path.basename(str(candidate or "")).lower() + if basename and basename == expected: + return True + return False + + def result_hash_matches(result, *, require_filename=False): + if not provided_hash or not isinstance(result, dict): + return False + result_hash = extract_result_sha256(result) + if result_hash != provided_hash: + return False + if require_filename and not result_filename_matches(result): + return False + return True + + def huggingface_page_url(result): + try: + from urllib.parse import quote as _quote + except Exception: + _quote = None + repo_id = str( + result.get("repo_id") or result.get("repo") or "" + ).strip() + hf_path = str( + result.get("path") or result.get("filename") or "" + ).strip() + if not repo_id or not hf_path: + return "" + if _quote: + hf_path = _quote(hf_path.replace("\\", "/"), safe="/") + return f"https://huggingface.co/{repo_id}/blob/main/{hf_path}" + + def prepare_remote_result(result, source_name): + result = dict(result or {}) + source_name = str(source_name or result.get("source") or "").lower() + if source_name == "huggingface": + download_url = result.get("download_url") or result.get("url") + page_url = result.get("page_url") or huggingface_page_url(result) + if page_url: + result["url"] = page_url + result["version_url"] = page_url + result["download_url"] = download_url + result["model_name"] = ( + result.get("model_name") + or result.get("name") + or result.get("repo_id") + or clean_name + ) + result["model_type"] = ( + result.get("model_type") + or infer_model_type_from_category(category) + ) + else: + result["model_name"] = ( + result.get("model_name") + or result.get("name") + or clean_name + ) + if result.get("url") and not result.get("version_url"): + result["version_url"] = result.get("url") + + result["source"] = source_name or result.get("source") or "" + result["details_source"] = ( + result.get("details_source") + or source_name + or result.get("source") + or "" + ) + result["file_path"] = file_path or result.get("file_path") or "" + result["resolved_path"] = ( + file_path or result.get("resolved_path") or "" + ) + result["location"] = result.get("location") or file_location + result["filename"] = result.get("filename") or filename + result["sha256"] = ( + extract_result_sha256(result) + or provided_hash + or result.get("sha256") + ) + if not result.get("size") and file_path and _os.path.exists(file_path): + try: + result["size"] = _os.path.getsize(file_path) + except Exception: + pass + return result + + def remote_link_is_marked_dead(item): + if not isinstance(item, dict): + return False + status = str(item.get("status") or "").lower() + return bool( + item.get("deletedAt") + or item.get("deleted_at") + or item.get("is_dead") + or item.get("isDead") + or item.get("likelyDead") + or item.get("likely_dead") + or item.get("dead") + or status in {"dead", "deleted", "unavailable", "missing"} + ) + + def result_url_looks_like_model_file(url, expected_filename=""): + text = str(url or "").strip() + if not text.startswith(("http://", "https://")): + return False + try: + from urllib.parse import unquote as _unquote, urlparse as _urlparse + parsed = _urlparse(text) + host = parsed.netloc.lower() + path = _unquote(parsed.path or "") + except Exception: + host = "" + path = text + if host.endswith("huggingface.co") and path.startswith("/spaces/"): + return False + if ( + host.endswith("civitai.com") + or host.endswith("civitai.red") + ) and path.startswith("/api/download/"): + return True + + basename = _os.path.basename(path).lower() + expected = _os.path.basename(str(expected_filename or "")).lower() + if expected and basename == expected: + return True + model_extensions = { + ".safetensors", + ".ckpt", + ".pt", + ".pth", + ".bin", + ".gguf", + ".onnx", + ".pb", + ".pkl", + ".pickle", + } + return _os.path.splitext(basename)[1].lower() in model_extensions + + def collect_result_download_urls(result): + urls = [] + if remote_link_is_marked_dead(result): + return urls + expected_filename = result.get("filename") or filename + dead_urls = set() + mirrors = result.get("mirrors") or [] + if not isinstance(mirrors, list): + mirrors = [mirrors] + for mirror in mirrors: + if not isinstance(mirror, dict): + continue + if remote_link_is_marked_dead(mirror): + dead_url = str(mirror.get("url") or "").strip() + if dead_url.startswith(("http://", "https://")): + dead_urls.add(dead_url) + continue + url = str(mirror.get("url") or "").strip() + mirror_filename = ( + mirror.get("filename") + or mirror.get("name") + or expected_filename + ) + if ( + result_url_looks_like_model_file(url, mirror_filename) + and url not in urls + ): + urls.append(url) + raw_urls = result.get("download_urls") or [] + if not isinstance(raw_urls, list): + raw_urls = [raw_urls] + for raw_url in raw_urls: + url = str(raw_url or "").strip() + if ( + result_url_looks_like_model_file(url, expected_filename) + and url not in dead_urls + and url not in urls + ): + urls.append(url) + for key in ("download_url", "downloadUrl"): + url = str(result.get(key) or "").strip() + if ( + result_url_looks_like_model_file(url, expected_filename) + and url not in dead_urls + and url not in urls + ): + urls.append(url) + return urls + + def remote_download_url_is_alive(url): + try: + import requests + except Exception: + return True + + headers = { + "User-Agent": "ComfyUI-Model-Resolver/1.0", + "Accept": "*/*", + } + try: + response = requests.head( + url, + headers=headers, + allow_redirects=True, + timeout=8, + ) + try: + if response.status_code < 400: + return True + if response.status_code in {401, 403, 404, 410}: + return False + finally: + response.close() + except Exception: + pass + + try: + response = requests.get( + url, + headers={**headers, "Range": "bytes=0-0"}, + allow_redirects=True, + stream=True, + timeout=8, + ) + try: + return response.status_code < 400 + finally: + response.close() + except Exception: + return False + + def civarchive_result_has_live_download(result): + urls = collect_result_download_urls(result) + if not urls: + return False + for url in urls[:3]: + if remote_download_url_is_alive(url): + result["download_url"] = url + result["download_urls"] = [ + url, + *[other for other in urls if other != url], + ] + return True + return False + + def save_remote_metadata(result, source_name): + metadata_path = result.get("metadata_path") or "" + metadata_saved = False + if ( + result.get("from_metadata") + or not file_path + or not _os.path.exists(file_path) + ): + return metadata_path, metadata_saved + try: + metadata_payload = { + "source": source_name, + "details_source": result.get("details_source") + or source_name, + "filename": filename, + "category": category, + "model_name": result.get("model_name", clean_name), + "name": result.get("model_name", clean_name), + "model_type": result.get("model_type", "") + or result.get("type", ""), + "type": result.get("model_type", "") + or result.get("type", ""), + "model_id": result.get("model_id"), + "version_id": result.get("version_id"), + "version_name": result.get("version_name", ""), + "sha256": result.get("sha256") or provided_hash, + "size": result.get("size"), + "base_model": result.get("base_model"), + "tags": result.get("tags", []), + "trained_words": result.get("trained_words", []), + "images": result.get("images", []), + "clip_skip": result.get("clip_skip"), + "description": result.get("description", ""), + "model_description": result.get("model_description", ""), + "download_url": result.get("download_url"), + "source_url": result.get("version_url") or result.get("url"), + "version_url": result.get("version_url") or result.get("url"), + "model_url": result.get("url"), + "url": result.get("version_url") or result.get("url"), + "platform_url": result.get("platform_url"), + "repo_id": result.get("repo_id"), + "path": result.get("path"), + "path_metadata": { + "filename": filename, + "category": category, + "source": source_name, + "model_id": result.get("model_id"), + "version_id": result.get("version_id"), + "repo_id": result.get("repo_id"), + "path": result.get("path"), + }, + } + metadata_path = write_lora_manager_metadata( + file_path, + metadata_payload, + category, + result.get("version_url") + or result.get("url") + or result.get("platform_url") + or result.get("download_url") + or "", + ) or "" + metadata_saved = bool(metadata_path) + if metadata_path: + result["metadata_path"] = metadata_path + except Exception as metadata_error: + self.logger.warning( + f"{source_name} metadata sidecar save failed: {metadata_error}" + ) + return metadata_path, metadata_saved + if local_only: result = None + metadata_path = "" + metadata_saved = False if download_available and file_path and _os.path.exists(file_path): try: from .core.sources.civitai import ( @@ -1255,17 +1626,99 @@ class ModelResolverExtension: file_path, local_only=True, ) + source_metadata_path = ( + result.get("metadata_path") + if isinstance(result, dict) + else "" + ) + canonical_metadata_path = get_metadata_sidecar_path(file_path) + if ( + isinstance(result, dict) + and result.get("from_metadata") + and source_metadata_path + and canonical_metadata_path + and _os.path.abspath(source_metadata_path) + != _os.path.abspath(canonical_metadata_path) + and not _os.path.exists(canonical_metadata_path) + ): + preview_url = result.get("preview_url") or "" + if not preview_url: + model_base_path, _model_ext = _os.path.splitext(file_path) + for preview_ext in ( + ".preview.png", + ".preview.jpg", + ".preview.jpeg", + ".preview.webp", + ): + preview_path = f"{model_base_path}{preview_ext}" + if _os.path.exists(preview_path): + preview_url = preview_path.replace("\\", "/") + break + metadata_payload = { + "source": "metadata_import", + "details_source": result.get("source") or "metadata", + "filename": filename, + "category": category, + "model_name": result.get("model_name", clean_name), + "name": result.get("model_name", clean_name), + "model_type": result.get("model_type", ""), + "type": result.get("model_type", ""), + "model_id": result.get("model_id"), + "version_id": result.get("version_id"), + "version_name": result.get("version_name", ""), + "sha256": result.get("sha256"), + "size": result.get("size"), + "base_model": result.get("base_model"), + "tags": result.get("tags", []), + "trained_words": result.get("trained_words", []), + "images": result.get("images", []), + "clip_skip": result.get("clip_skip"), + "description": result.get("description", ""), + "model_description": result.get("model_description", ""), + "download_url": result.get("download_url"), + "preview_url": preview_url, + "source_url": result.get("version_url") + or result.get("url"), + "version_url": result.get("version_url") + or result.get("url"), + "model_url": result.get("url"), + "url": result.get("version_url") or result.get("url"), + "path_metadata": { + "filename": filename, + "category": category, + "source": "metadata_import", + "model_id": result.get("model_id"), + "version_id": result.get("version_id"), + "imported_from": source_metadata_path, + }, + } + metadata_path = write_lora_manager_metadata( + file_path, + metadata_payload, + category, + result.get("version_url") + or result.get("url") + or result.get("download_url") + or "", + ) or "" + metadata_saved = bool(metadata_path) + if metadata_path: + result["metadata_path"] = metadata_path + result["metadata_imported_from"] = source_metadata_path except Exception as e: self.logger.warning(f"Local model info error: {e}") return web.json_response( build_info_response( result, + metadata_path=metadata_path, + metadata_saved=metadata_saved, local_payload=True, civitai_checked=False, ) ) - # Search CivitAI for the model using hash + # Search remote metadata. CivitAI and CivArchive are hash-only; + # HuggingFace name results must still confirm the exact SHA. if download_available and file_path and _os.path.exists(file_path): try: from .core.sources.civitai import ( @@ -1274,7 +1727,10 @@ class ModelResolverExtension: ) if provided_hash: - result = get_model_info_by_hash(provided_hash) + result = get_model_info_by_hash( + provided_hash, + use_cache=not force_refresh, + ) if result: result["file_path"] = file_path result["resolved_path"] = file_path @@ -1284,66 +1740,28 @@ class ModelResolverExtension: result["size"] = _os.path.getsize(file_path) except Exception: pass - else: + if not extract_result_sha256(result): + result["sha256"] = provided_hash + elif not force_refresh: result = get_model_info_for_file(file_path) + else: + result = None if result and ( result.get("url") or result.get("version_url") or result.get("from_metadata") or result.get("trained_words") + ) and ( + result.get("from_metadata") + or not provided_hash + or result_hash_matches(result) ): - metadata_path = result.get("metadata_path") or "" - metadata_saved = False - if not result.get("from_metadata"): - try: - sidecar_path = get_metadata_sidecar_path(file_path) - if sidecar_path: - metadata_payload = { - "source": "civitai", - "details_source": "civitai", - "filename": filename, - "category": category, - "model_name": result.get("model_name", clean_name), - "name": result.get("model_name", clean_name), - "model_type": result.get("model_type", ""), - "type": result.get("model_type", ""), - "model_id": result.get("model_id"), - "version_id": result.get("version_id"), - "version_name": result.get("version_name", ""), - "sha256": result.get("sha256") or provided_hash, - "size": result.get("size"), - "base_model": result.get("base_model"), - "tags": result.get("tags", []), - "trained_words": result.get("trained_words", []), - "images": result.get("images", []), - "clip_skip": result.get("clip_skip"), - "description": result.get("description", ""), - "model_description": result.get("model_description", ""), - "download_url": result.get("download_url"), - "url": result.get("version_url") or result.get("url"), - "path_metadata": { - "filename": filename, - "category": category, - "source": "civitai", - "model_id": result.get("model_id"), - "version_id": result.get("version_id"), - }, - } - metadata_path = write_lora_manager_metadata( - file_path, - metadata_payload, - category, - result.get("version_url") - or result.get("url") - or result.get("download_url") - or "", - ) or "" - metadata_saved = bool(metadata_path) - except Exception as metadata_error: - self.logger.warning( - f"CivitAI metadata sidecar save failed: {metadata_error}" - ) + result = prepare_remote_result(result, "civitai") + metadata_path, metadata_saved = save_remote_metadata( + result, + "civitai", + ) return web.json_response( build_info_response( @@ -1356,37 +1774,135 @@ class ModelResolverExtension: except Exception as e: self.logger.warning(f"CivitAI search error: {e}") - # No result found - try fallback to filename search - if download_available: - try: - from .core.sources.civitai import ( - search_civitai_for_file, - ) + if provided_hash: + try: + result = resolve_civarchive_by_hash( + provided_hash, + query=filename, + exact_only=False, + model_type=infer_model_type_from_category(category), + ) + if result: + if not extract_result_sha256(result): + result["sha256"] = provided_hash + result = prepare_remote_result(result, "civarchive") + if result_hash_matches(result) and civarchive_result_has_live_download(result): + metadata_path, metadata_saved = save_remote_metadata( + result, + "civarchive", + ) + return web.json_response( + build_info_response( + result, + metadata_path=metadata_path, + metadata_saved=metadata_saved, + civitai_checked=True, + ) + ) + if result_hash_matches(result): + self.logger.info( + f"CivArchive metadata candidate rejected: " + f"no live download link for {filename}" + ) + except Exception as e: + self.logger.warning(f"CivArchive hash lookup error: {e}") - result = search_civitai_for_file( - filename, model_type=category - ) - if result and result.get("url"): - return web.json_response( + if provided_hash: + hf_attempts = [ + { + "label": "HuggingFace", + "use_api_search": True, + "use_comfy_org_fallback": True, + "use_brave_fallback": False, + } + ] + if hf_use_brave_fallback and brave_search_api_key: + hf_attempts.append( { - "filename": filename, - "url": result["url"], - "file_path": file_path, - "resolved_path": file_path, - "location": file_location, - "model_name": result.get("name", clean_name), - "version_id": result.get("version_id"), - "size": result.get("size"), - "tags": result.get("tags", []), - "civitai_checked": True, + "label": "HuggingFace Brave", + "use_api_search": False, + "use_comfy_org_fallback": False, + "use_brave_fallback": True, } ) - except Exception as e: - self.logger.warning(f"CivitAI fallback search error: {e}") + + for hf_attempt in hf_attempts: + try: + result = search_huggingface_for_file( + filename, + token=hf_token or None, + exact_only=True, + brave_api_key=brave_search_api_key or None, + use_api_search=hf_attempt["use_api_search"], + use_comfy_org_fallback=hf_attempt[ + "use_comfy_org_fallback" + ], + use_brave_fallback=hf_attempt[ + "use_brave_fallback" + ], + force_refresh=force_refresh, + ) + if result and result_hash_matches( + result, + require_filename=True, + ): + result = prepare_remote_result( + result, + "huggingface", + ) + metadata_path, metadata_saved = save_remote_metadata( + result, + "huggingface", + ) + return web.json_response( + build_info_response( + result, + metadata_path=metadata_path, + metadata_saved=metadata_saved, + civitai_checked=True, + ) + ) + if result: + self.logger.info( + f"{hf_attempt['label']} metadata candidate rejected: " + f"filename/hash mismatch for {filename}" + ) + except Exception as e: + self.logger.warning( + f"{hf_attempt['label']} metadata lookup error: {e}" + ) # No result found response = build_info_response(None, civitai_checked=True) response["url"] = None + if force_refresh and file_path and _os.path.exists(file_path): + try: + metadata_payload = { + "filename": filename, + "category": category, + "model_name": clean_name, + "name": clean_name, + "model_type": infer_model_type_from_category(category), + "sha256": provided_hash, + "size": _os.path.getsize(file_path), + "civitai_deleted": True, + "civitai_checked": True, + "remote_metadata_missing": True, + "source": "local", + "details_source": "", + } + metadata_path = write_lora_manager_metadata( + file_path, + metadata_payload, + category, + "", + ) or "" + response["metadata_path"] = metadata_path + response["metadata_saved"] = bool(metadata_path) + except Exception as metadata_error: + self.logger.warning( + f"Remote metadata no-match sidecar save failed: {metadata_error}" + ) return web.json_response(response) @routes.post("/model_resolver/model-details") diff --git a/core/downloader.py b/core/downloader.py index 605ba06..3de8f78 100644 --- a/core/downloader.py +++ b/core/downloader.py @@ -395,6 +395,23 @@ def build_lora_manager_metadata( source_url, source.get("url"), ) + source_page_url = _first_present( + source.get("version_url"), + source.get("model_url"), + source.get("page_url"), + source.get("source_url"), + source.get("url"), + source.get("platform_url"), + source_url, + ) + platform_url = _first_present(source.get("platform_url"), source.get("platformUrl")) + preview_url = _first_present( + source.get("preview_url"), + source.get("previewUrl"), + source.get("preview"), + source.get("thumbnail_url"), + source.get("thumbnailUrl"), + ) model_description = _first_present( source.get("modelDescription"), source.get("model_description"), @@ -458,7 +475,7 @@ def build_lora_manager_metadata( "modified": time.time(), "sha256": sha256, "base_model": str(base_model or "Unknown"), - "preview_url": "", + "preview_url": str(preview_url or ""), "preview_nsfw_level": 0, "notes": "", "from_civitai": bool(metadata_source or model_id or version_id), @@ -470,6 +487,13 @@ def build_lora_manager_metadata( "exclude": False, "db_checked": False, "skip_metadata_refresh": False, + "source": source_name, + "details_source": source_name, + "source_url": _strip_sensitive_url_params(str(source_page_url or "")), + "model_url": _strip_sensitive_url_params(str(source_page_url or "")), + "version_url": _strip_sensitive_url_params(str(source_page_url or "")), + "download_url": _strip_sensitive_url_params(str(direct_url or "")), + "platform_url": _strip_sensitive_url_params(str(platform_url or "")), "metadata_source": metadata_source, "last_checked_at": time.time(), "hash_status": "completed" if sha256 else "pending", diff --git a/core/sources/civarchive.py b/core/sources/civarchive.py index 30a2fa0..668f85e 100644 --- a/core/sources/civarchive.py +++ b/core/sources/civarchive.py @@ -10,7 +10,7 @@ import os import re import time from typing import Any, Callable, Dict, List, Optional, Tuple -from urllib.parse import parse_qs, urljoin, urlparse +from urllib.parse import parse_qs, unquote, urljoin, urlparse import requests @@ -416,7 +416,7 @@ def _extract_download_urls_from_html(html_text: str) -> List[str]: ) for match in pattern.finditer(text): url = _normalize_download_url(match.group(0).rstrip(".,;]})")) - if url and url not in urls: + if url and _download_url_looks_like_model_file(url) and url not in urls: urls.append(url) return urls @@ -586,6 +586,45 @@ def _normalize_download_url(url: Any) -> Optional[str]: return None +def _download_url_looks_like_model_file( + url: Any, + expected_filename: str = "", +) -> bool: + normalized = _normalize_download_url(url) + if not normalized: + return False + + parsed = urlparse(normalized) + host = parsed.netloc.lower() + path = unquote(parsed.path or "") + if host.endswith("huggingface.co") and path.startswith("/spaces/"): + return False + if normalized.startswith(CIVITAI_DOWNLOAD_URL_PREFIXES): + return True + + basename = os.path.basename(path).lower() + expected = os.path.basename(str(expected_filename or "")).lower() + if expected and basename == expected: + return True + return _has_known_model_extension(basename) + + +def _archive_link_is_dead(item: Dict[str, Any]) -> bool: + if not isinstance(item, dict): + return False + status = str(item.get("status") or "").lower() + return bool( + item.get("deletedAt") + or item.get("deleted_at") + or item.get("is_dead") + or item.get("isDead") + or item.get("likelyDead") + or item.get("likely_dead") + or item.get("dead") + or status in {"dead", "deleted", "unavailable", "missing"} + ) + + def _prepare_size_probe_url(url: Any) -> Optional[str]: normalized = _normalize_download_url(url) if not normalized: @@ -746,16 +785,18 @@ def _transform_file_entry(file_data: Dict[str, Any]) -> Dict[str, Any]: mirror["url"] = mirror_url transformed_mirrors.append(mirror) - download_url = _normalize_download_url( - file_data.get("downloadUrl") - or file_data.get("download_url") - or file_data.get("url") - ) + download_url = None + if not _archive_link_is_dead(file_data): + download_url = _normalize_download_url( + file_data.get("downloadUrl") + or file_data.get("download_url") + or file_data.get("url") + ) name = file_data.get("name") or file_data.get("filename") if not download_url: for mirror in transformed_mirrors: - if mirror.get("deletedAt") is None and mirror.get("url"): + if not _archive_link_is_dead(mirror) and mirror.get("url"): download_url = mirror["url"] break @@ -877,6 +918,7 @@ def _select_primary_file(files: List[Dict[str, Any]]) -> Optional[Dict[str, Any] def _collect_download_urls(file_info: Dict[str, Any]) -> List[str]: urls: List[str] = [] + expected_filename = file_info.get("filename") or file_info.get("name") or "" mirrors = file_info.get("mirrors") or [] if not isinstance(mirrors, list): mirrors = [mirrors] @@ -884,14 +926,25 @@ def _collect_download_urls(file_info: Dict[str, Any]) -> List[str]: for mirror in mirrors: if not isinstance(mirror, dict): continue - if mirror.get("deletedAt") is not None: + if _archive_link_is_dead(mirror): continue url = _normalize_download_url(mirror.get("url")) - if url and url not in urls: + mirror_filename = mirror.get("filename") or mirror.get("name") or expected_filename + if ( + url + and _download_url_looks_like_model_file(url, mirror_filename) + and url not in urls + ): urls.append(url) - download_url = _normalize_download_url(file_info.get("downloadUrl")) - if download_url and download_url not in urls: + download_url = None + if not _archive_link_is_dead(file_info): + download_url = _normalize_download_url(file_info.get("downloadUrl")) + if ( + download_url + and _download_url_looks_like_model_file(download_url, expected_filename) + and download_url not in urls + ): urls.append(download_url) non_civitai_urls = [ @@ -917,15 +970,7 @@ def _normalize_archive_mirrors(file_info: Dict[str, Any]) -> List[Dict[str, Any] seen_urls.add(url) sha256 = mirror.get("sha256") or mirror.get("hash") deleted_at = mirror.get("deletedAt") or mirror.get("deleted_at") - is_dead = bool( - deleted_at - or mirror.get("is_dead") - or mirror.get("isDead") - or mirror.get("likelyDead") - or mirror.get("likely_dead") - or mirror.get("dead") - or str(mirror.get("status") or "").lower() in {"dead", "deleted", "unavailable", "missing"} - ) + is_dead = _archive_link_is_dead(mirror) normalized.append( { "url": url, @@ -940,7 +985,9 @@ def _normalize_archive_mirrors(file_info: Dict[str, Any]) -> List[Dict[str, Any] } ) - download_url = _normalize_download_url(file_info.get("downloadUrl")) + download_url = None + if not _archive_link_is_dead(file_info): + download_url = _normalize_download_url(file_info.get("downloadUrl")) if download_url and download_url not in seen_urls: normalized.append( { @@ -950,16 +997,7 @@ def _normalize_archive_mirrors(file_info: Dict[str, Any]) -> List[Dict[str, Any] "sha256": file_info.get("sha256"), "hash": file_info.get("sha256"), "deleted_at": file_info.get("deletedAt") or file_info.get("deleted_at"), - "is_dead": bool( - file_info.get("deletedAt") - or file_info.get("deleted_at") - or file_info.get("is_dead") - or file_info.get("isDead") - or file_info.get("likelyDead") - or file_info.get("likely_dead") - or file_info.get("dead") - or str(file_info.get("status") or "").lower() in {"dead", "deleted", "unavailable", "missing"} - ), + "is_dead": _archive_link_is_dead(file_info), "is_gated": file_info.get("is_gated"), "is_paid": file_info.get("is_paid"), } @@ -1208,18 +1246,53 @@ def _version_sort_key(version: Dict[str, Any]) -> tuple: def _collect_normalized_download_urls(file_info: Dict[str, Any]) -> List[str]: urls: List[str] = [] + if _archive_link_is_dead(file_info): + return urls + + expected_filename = file_info.get("filename") or file_info.get("name") or "" + dead_urls = set() + mirrors = file_info.get("mirrors") or [] + if not isinstance(mirrors, list): + mirrors = [mirrors] + for mirror in mirrors: + if not isinstance(mirror, dict): + continue + if _archive_link_is_dead(mirror): + dead_url = _normalize_download_url(mirror.get("url")) + if dead_url: + dead_urls.add(dead_url) + continue + url = _normalize_download_url(mirror.get("url")) + mirror_filename = mirror.get("filename") or mirror.get("name") or expected_filename + if ( + url + and _download_url_looks_like_model_file(url, mirror_filename) + and url not in urls + ): + urls.append(url) + raw_urls = file_info.get("download_urls") or [] if not isinstance(raw_urls, list): raw_urls = [raw_urls] for raw_url in raw_urls: url = _normalize_download_url(raw_url) - if url and url not in urls: + if ( + url + and url not in dead_urls + and _download_url_looks_like_model_file(url, expected_filename) + and url not in urls + ): urls.append(url) for key in ("download_url", "downloadUrl"): url = _normalize_download_url(file_info.get(key)) - if url and url not in urls: + if ( + url + and url not in dead_urls + and _download_url_looks_like_model_file(url, expected_filename) + and url not in urls + ): urls.append(url) return urls @@ -1511,7 +1584,9 @@ def _select_hash_page_file( for file_info in files: if not isinstance(file_info, dict): continue - if not _normalize_download_url(file_info.get("url") or file_info.get("downloadUrl")): + if _archive_link_is_dead(file_info): + continue + if not _collect_normalized_download_urls(file_info): continue filename = file_info.get("filename") or file_info.get("name") or "" @@ -1584,6 +1659,8 @@ def _prefer_query_matching_mirror( for mirror in mirrors: if not isinstance(mirror, dict): continue + if _archive_link_is_dead(mirror): + continue url = _normalize_download_url(mirror.get("url")) filename = mirror.get("filename") or mirror.get("name") or "" if not url or not filename: diff --git a/core/sources/civitai.py b/core/sources/civitai.py index 630089c..da09739 100644 --- a/core/sources/civitai.py +++ b/core/sources/civitai.py @@ -1119,6 +1119,55 @@ def get_civitai_model_details( } +def _enrich_model_info_with_details( + result: Dict[str, Any], + model_id: Optional[int], + version_id: Optional[int], + api_key: Optional[str] = None, +) -> Dict[str, Any]: + """Fill hash lookup results with the fuller /models/{id} payload when useful.""" + if not result or not model_id: + return result + + needs_details = ( + not result.get("images") + or not result.get("description") + or not result.get("model_description") + or not result.get("tags") + ) + if not needs_details: + return result + + try: + details = get_civitai_model_details(model_id, version_id, api_key) + except Exception as exc: + log_debug(f"CivitAI details enrichment failed for model_id={model_id}: {exc}") + return result + + if not details: + return result + + selected_version = _as_metadata_dict(details.get("selected_version")) + result["model_name"] = result.get("model_name") or details.get("name") + result["model_type"] = result.get("model_type") or details.get("type") + result["description"] = ( + result.get("description") + or selected_version.get("description") + or details.get("description") + or "" + ) + result["model_description"] = ( + result.get("model_description") + or details.get("description") + or "" + ) + result["tags"] = result.get("tags") or details.get("tags") or [] + result["images"] = result.get("images") or details.get("images") or [] + result["version_url"] = result.get("version_url") or details.get("version_url") + result["url"] = result.get("url") or details.get("url") + return result + + def search_civitai_by_hash( hash_value: str, api_key: Optional[str] = None ) -> Optional[Dict[str, Any]]: @@ -1344,6 +1393,12 @@ def get_model_info_by_hash( "model_description": model_info.get("description", ""), } + result = _enrich_model_info_with_details( + result, + model_id, + version_info.get("id"), + api_key, + ) _hash_cache[cache_key] = result log_info(f"Found model by hash {file_hash}: {result.get('model_name')}") return result @@ -1411,6 +1466,12 @@ def get_model_info_by_hash( "model_description": model_info.get("description", ""), } + result = _enrich_model_info_with_details( + result, + model_id, + version_info.get("id"), + api_key, + ) _hash_cache[cache_key] = result log_info(f"Found model by hash {file_hash}: {result.get('model_name')}") return result @@ -1540,6 +1601,8 @@ def _get_metadata_file_path(model_path: str) -> str: possible_names = [ base_name + ".metadata.json", filename + ".metadata.json", + base_name + ".civitai.info", + filename + ".civitai.info", base_name + ".json", filename.replace("_", " ").split()[0] + ".metadata.json" if "_" in base_name @@ -1572,8 +1635,34 @@ def _read_model_metadata(metadata_path: str) -> Optional[Dict[str, Any]]: if not isinstance(data, dict): return None - # Check if it has the needed info - if not data.get("sha256") and not data.get("civitai"): + files = _as_metadata_list(data.get("files")) + file_hashes = [ + _as_metadata_dict(file_info).get("hashes") + for file_info in files + if isinstance(file_info, dict) + ] + has_file_sha = any( + isinstance(hashes, dict) + and (hashes.get("SHA256") or hashes.get("sha256")) + for hashes in file_hashes + ) + has_civitai_shape = bool( + data.get("civitai") + or data.get("modelId") + or data.get("model") + or data.get("images") + or data.get("downloadUrl") + ) + + # Check if it has the needed info. Older sidecars use a top-level + # CivitAI model-version shape and are commonly named *.civitai.info. + if ( + not data.get("sha256") + and not data.get("hash") + and not data.get("hashes") + and not has_file_sha + and not has_civitai_shape + ): return None log_info(f"Successfully read metadata from: {metadata_path}") @@ -1644,13 +1733,29 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]: """ Convert metadata file format to model info format used by our extension. """ - civitai_data = _as_metadata_dict(metadata.get("civitai")) + embedded_civitai_data = _as_metadata_dict(metadata.get("civitai")) + looks_like_civitai_version = bool( + metadata.get("modelId") + or metadata.get("files") + or metadata.get("images") + or metadata.get("downloadUrl") + ) + civitai_data = embedded_civitai_data or ( + metadata if looks_like_civitai_version else {} + ) selected_version = _as_metadata_dict(metadata.get("selected_version")) + if not selected_version and looks_like_civitai_version: + selected_version = civitai_data path_metadata = _as_metadata_dict(metadata.get("path_metadata")) # Extract images with metadata images = [] - for img in civitai_data.get("images") or []: + for img in ( + civitai_data.get("images") + or selected_version.get("images") + or metadata.get("images") + or [] + ): if isinstance(img, dict) and img.get("url"): img_meta = img.get("meta") or {} images.append( @@ -1700,22 +1805,60 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]: file_infos = _as_metadata_list(civitai_data.get("files")) file_info = _as_metadata_dict(file_infos[0]) if file_infos else {} metadata_hashes = _as_metadata_dict(metadata.get("hashes")) + file_hashes = _as_metadata_dict(file_info.get("hashes")) # Build model_id from CivitAI data - model_id = civitai_data.get("modelId") or civitai_data.get("id") + model_id = ( + civitai_data.get("modelId") + or metadata.get("model_id") + or metadata.get("modelId") + ) + version_id = ( + civitai_data.get("id") + or metadata.get("version_id") + or metadata.get("versionId") + ) + details_source = ( + metadata.get("details_source") + or metadata.get("source") + or metadata.get("metadata_source") + or "metadata" + ) + stored_page_url = _first_metadata_value( + metadata.get("version_url"), + metadata.get("model_url"), + metadata.get("page_url"), + metadata.get("source_url"), + metadata.get("url"), + metadata.get("platform_url"), + path_metadata.get("version_url"), + path_metadata.get("model_url"), + path_metadata.get("source_url"), + path_metadata.get("url"), + path_metadata.get("platform_url"), + ) + civitai_model_url = f"https://civitai.com/models/{model_id}" if model_id else None + civitai_version_url = ( + f"https://civitai.com/models/{model_id}?modelVersionId={version_id}" + if model_id and version_id + else civitai_model_url + ) + page_url = stored_page_url or civitai_version_url return { - "source": "metadata", + "source": details_source, + "details_source": details_source, "model_id": model_id, "model_name": _first_metadata_value( metadata.get("model_name"), metadata.get("modelName"), model_info.get("name"), + metadata.get("name") if not model_id else None, metadata.get("file_name"), ) or "", "model_type": model_info.get("type", "") or civitai_data.get("type", ""), - "version_id": civitai_data.get("id"), + "version_id": version_id, "version_name": _first_metadata_value( metadata.get("version_name"), metadata.get("versionName"), @@ -1728,6 +1871,8 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]: metadata.get("hash"), metadata_hashes.get("SHA256"), metadata_hashes.get("sha256"), + file_hashes.get("SHA256"), + file_hashes.get("sha256"), ) or "", "size": _metadata_size_to_bytes( @@ -1745,13 +1890,16 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]: else None, ) ), - "url": f"https://civitai.com/models/{civitai_data.get('modelId')}" - if civitai_data.get("modelId") - else None, - "version_url": f"https://civitai.com/models/{civitai_data.get('modelId')}?modelVersionId={civitai_data.get('id')}" - if model_id - else None, - "download_url": civitai_data.get("downloadUrl"), + "url": page_url, + "version_url": page_url, + "model_url": stored_page_url or civitai_model_url, + "source_url": stored_page_url, + "platform_url": metadata.get("platform_url") or path_metadata.get("platform_url"), + "download_url": _first_metadata_value( + metadata.get("download_url"), + civitai_data.get("downloadUrl"), + file_info.get("downloadUrl"), + ), "base_model": _first_metadata_value( metadata.get("base_model"), metadata.get("baseModel"), diff --git a/web/resolver/views/model_info_methods.js b/web/resolver/views/model_info_methods.js index 07d5c15..dba5d84 100644 --- a/web/resolver/views/model_info_methods.js +++ b/web/resolver/views/model_info_methods.js @@ -1140,6 +1140,7 @@ export const modelInfoMethods = { if (!dialog) return; if (button?.dataset.fetchingCivitai === 'true') return; + const forceRefresh = button?.dataset.forceRefresh === 'true'; const lookup = dialog._infoDialogLookup || {}; let data = dialog._infoDialogData || {}; const filename = data.filename || lookup.loraName || ''; @@ -1153,13 +1154,14 @@ export const modelInfoMethods = { } dialog._infoDialogCivitaiFetchStarted = true; - const setFetchingState = (isFetching, label = 'Fetch with CivitAI') => { + const defaultLabel = forceRefresh ? 'Refetch metadata' : 'Fetch metadata'; + const setFetchingState = (isFetching, label = defaultLabel) => { if (!button) return; button.dataset.fetchingCivitai = isFetching ? 'true' : 'false'; button.disabled = isFetching; button.innerHTML = isFetching - ? `${getSvgIcon('civitai', 'currentColor', 'mr-info-external-link-icon')} Fetching...` - : `${getSvgIcon('civitai', 'currentColor', 'mr-info-external-link-icon')} ${this.escapeHtml(label)}`; + ? `${getSvgIcon('globe', 'currentColor', 'mr-info-external-link-icon')} Fetching...` + : `${getSvgIcon('globe', 'currentColor', 'mr-info-external-link-icon')} ${this.escapeHtml(label)}`; }; setFetchingState(true); @@ -1181,15 +1183,20 @@ export const modelInfoMethods = { throw new Error('No SHA256 hash available'); } + const tokens = this.getStoredTokens?.() || {}; const result = await this.fetchJson('/model_resolver/civitai-search', { method: 'POST', body: JSON.stringify({ filename, category, resolved_path: resolvedPath, - sha256: hash + sha256: hash, + force_refresh: forceRefresh, + hf_token: tokens.hf_token || '', + brave_search_api_key: tokens.brave_search_api_key || '', + hf_use_brave_fallback: tokens.hf_use_brave_fallback !== false }) - }, 'Fetch CivitAI model info'); + }, 'Fetch model metadata'); const merged = { ...data, @@ -1199,19 +1206,24 @@ export const modelInfoMethods = { file_path: result.file_path || data.file_path || resolvedPath, resolved_path: result.resolved_path || data.resolved_path || resolvedPath, sha256: result.sha256 || hash, + metadata_checked: true, civitai_checked: true }; dialog._infoDialogData = merged; this.updateInfoDialogWithData(dialog, merged); if (merged.url || merged.version_url) { - this.showNotification?.('CivitAI info loaded.', 'success'); + this.showNotification?.( + forceRefresh ? 'Metadata refreshed.' : 'Metadata loaded.', + 'success' + ); } else { - this.showNotification?.('No exact CivitAI match found.', 'info'); + this.showNotification?.('No exact remote metadata match found.', 'info'); } } catch (error) { - this.showNotification?.(`CivitAI fetch failed: ${error?.message || error}`, 'error'); + this.showNotification?.(`Metadata fetch failed: ${error?.message || error}`, 'error'); this.updateInfoDialogWithData(dialog, { ...(dialog._infoDialogData || data), + metadata_checked: true, civitai_checked: true }); } finally { @@ -1219,6 +1231,34 @@ export const modelInfoMethods = { } }, + getInfoMetadataSourceKey(data = {}) { + const source = String(data.details_source || data.source || '').toLowerCase(); + const url = String(data.version_url || data.url || data.platform_url || '').toLowerCase(); + if (source.includes('huggingface') || url.includes('huggingface.co')) return 'huggingface'; + if (source.includes('civarchive') || url.includes('civarchive.com')) return 'civarchive'; + if (source.includes('civitai') || url.includes('civitai.com') || url.includes('civitai.red')) return 'civitai'; + return 'metadata'; + }, + + getInfoMetadataSourceLabel(data = {}) { + const key = this.getInfoMetadataSourceKey(data); + if (key === 'huggingface') return 'HuggingFace'; + if (key === 'civarchive') return 'CivArchive'; + if (key === 'civitai') return 'CivitAI'; + return 'metadata source'; + }, + + getInfoMetadataSourceIcon(data = {}) { + const key = this.getInfoMetadataSourceKey(data); + if (key === 'huggingface') return getSvgIcon('huggingface', 'currentColor', 'mr-info-external-link-icon'); + if (key === 'civitai') return getSvgIcon('civitai', 'currentColor', 'mr-info-external-link-icon'); + return getSvgIcon('externalLink', 'currentColor', 'mr-info-external-link-icon'); + }, + + hasInfoMetadataBeenChecked(data = {}) { + return Boolean(data.metadata_checked || data.metadataChecked || data.civitai_checked || data.civitaiChecked); + }, + /** * Update the info dialog with data */ @@ -1317,28 +1357,37 @@ export const modelInfoMethods = { // Update CivitAI link const civitaiLinkEl = dialog.querySelector('.mr-info-civitai-link'); if (civitaiLinkEl) { + const hasLocalFile = Boolean(this.getInfoDialogModelFilePath(data)); if (data.url || data.version_url) { const url = data.version_url || data.url; + const sourceLabel = this.getInfoMetadataSourceLabel(data); civitaiLinkEl.innerHTML = ` - - View on Civitai - ${getSvgIcon('externalLink', 'currentColor', 'mr-info-external-link-icon')} + + View on ${this.escapeHtml(sourceLabel)} + ${this.getInfoMetadataSourceIcon(data)} + ${hasLocalFile ? ` + + ` : ''} `; - } else if (data.civitai_checked || data.civitaiChecked) { - const searchName = data.model_name || data.modelName || 'Unknown'; + } else if (this.hasInfoMetadataBeenChecked(data)) { civitaiLinkEl.innerHTML = ` - Model not found - - Search on CivitAI - ${getSvgIcon('externalLink', 'currentColor', 'mr-info-external-link-icon')} - + No exact metadata match + ${hasLocalFile ? ` + + ` : ''} `; - } else if (this.getInfoDialogModelFilePath(data)) { + } else if (hasLocalFile) { civitaiLinkEl.innerHTML = ` `; } else {