nodes/Azornes-Comfyui-Model-Resolver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: generalize remote metadata lookup across CivitAI, CivArchive, and HuggingFace

Browse Source 
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.
This commit is contained in:
Dariusz L
2026-06-27 03:25:31 +02:00
parent 3cd302d4ee
commit e53d9f2a0b
5 changed files with 964 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/downloader.py
+25 -1
View File
@@ -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",
core/sources/civarchive.py
+112 -35
View File
@@ -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:
core/sources/civitai.py
+162 -14
View File
@@ -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"),