refactor: extract downloader metadata helpers
This commit is contained in:
Dariusz L
@@ -0,0 +1,495 @@
|
||||
"""Metadata normalization and sidecar payload helpers for downloads."""
|
||||
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Dict, Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from ..log_system import create_module_logger
|
||||
from ..network_utils import host_matches_domain
|
||||
from ..path_utils import (
|
||||
MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
MODEL_RESOLVER_METADATA_SCHEMA_VERSION,
|
||||
find_metadata_sidecar_path,
|
||||
get_filename_from_path,
|
||||
read_merged_model_metadata,
|
||||
)
|
||||
from ..resolver import normalize_sha256
|
||||
from ..settings import normalize_download_category
|
||||
from ..type_utils import (
|
||||
as_dict,
|
||||
as_list,
|
||||
first_non_empty,
|
||||
normalize_category_to_model_type,
|
||||
)
|
||||
from .validation import _is_sensitive_metadata_key, _strip_sensitive_url_params
|
||||
|
||||
log = create_module_logger("core.downloader")
|
||||
|
||||
_as_dict = as_dict
|
||||
_as_list = as_list
|
||||
_first_present = first_non_empty
|
||||
|
||||
|
||||
def _json_safe_metadata(value: Any, depth: int = 0) -> Any:
|
||||
if depth > 10:
|
||||
return str(value)
|
||||
|
||||
if isinstance(value, dict):
|
||||
cleaned = {}
|
||||
for key, item_value in value.items():
|
||||
if _is_sensitive_metadata_key(key):
|
||||
continue
|
||||
cleaned[str(key)] = _json_safe_metadata(item_value, depth + 1)
|
||||
return cleaned
|
||||
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
return [_json_safe_metadata(item, depth + 1) for item in value]
|
||||
|
||||
if isinstance(value, str):
|
||||
return _strip_sensitive_url_params(value)
|
||||
|
||||
if value is None or isinstance(value, (bool, int, float)):
|
||||
return value
|
||||
|
||||
return str(value)
|
||||
|
||||
|
||||
def _coerce_int_or_value(value: Any) -> Any:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return value
|
||||
|
||||
|
||||
def _coerce_size(value: Any) -> int:
|
||||
try:
|
||||
return int(float(value or 0))
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def _normalise_metadata_file_path(path_value: str) -> str:
|
||||
return str(path_value or "").replace(os.sep, "/")
|
||||
|
||||
|
||||
def _resolve_lora_manager_model_type(category: str, source_type: Any = "") -> str:
|
||||
result = normalize_category_to_model_type(category)
|
||||
if result in ("checkpoint", "diffusion_model", "embedding"):
|
||||
return result
|
||||
|
||||
source_token = (
|
||||
str(source_type or "")
|
||||
.strip()
|
||||
.lower()
|
||||
.replace(" ", "_")
|
||||
.replace("-", "_")
|
||||
)
|
||||
if "checkpoint" in source_token:
|
||||
return "checkpoint"
|
||||
if "diffusion" in source_token or source_token == "unet":
|
||||
return "diffusion_model"
|
||||
if "textual" in source_token or "embedding" in source_token:
|
||||
return "embedding"
|
||||
return ""
|
||||
|
||||
|
||||
def _metadata_source_value(source_name: str, existing: Any = None) -> Optional[str]:
|
||||
if existing:
|
||||
return str(existing)
|
||||
|
||||
source_token = str(source_name or "").strip().lower()
|
||||
if source_token == "civitai":
|
||||
return "civitai_api"
|
||||
if source_token == "civarchive":
|
||||
return "civarchive"
|
||||
if source_token == "lora_manager_archive":
|
||||
return "archive_db"
|
||||
return None
|
||||
|
||||
|
||||
def _find_metadata_file_info(
|
||||
source: Dict[str, Any],
|
||||
selected_version: Dict[str, Any],
|
||||
filename: str,
|
||||
) -> Dict[str, Any]:
|
||||
# A selected file is the most specific provenance available. It must be
|
||||
# checked before the source-level hash because the latter can still belong
|
||||
# to the workflow's original model after the user chooses another variant.
|
||||
for key in ("selected_file", "selectedFile", "file_info", "file"):
|
||||
value = source.get(key)
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
|
||||
filename_lower = get_filename_from_path(str(filename or "")).lower()
|
||||
file_lists = [
|
||||
source.get("files"),
|
||||
selected_version.get("files"),
|
||||
]
|
||||
for file_list in file_lists:
|
||||
if not isinstance(file_list, list):
|
||||
continue
|
||||
first_file = None
|
||||
for file_info in file_list:
|
||||
if not isinstance(file_info, dict):
|
||||
continue
|
||||
if first_file is None:
|
||||
first_file = file_info
|
||||
candidate = str(
|
||||
file_info.get("name")
|
||||
or file_info.get("filename")
|
||||
or file_info.get("fileName")
|
||||
or ""
|
||||
).lower()
|
||||
if filename_lower and candidate == filename_lower:
|
||||
return file_info
|
||||
if first_file:
|
||||
return first_file
|
||||
return {}
|
||||
|
||||
|
||||
def _extract_expected_sha256(metadata: Optional[Dict[str, Any]]) -> str:
|
||||
source = metadata if isinstance(metadata, dict) else {}
|
||||
details = _as_dict(source.get("civitai_details") or source.get("details"))
|
||||
selected_version = _as_dict(
|
||||
source.get("selected_version") or details.get("selected_version")
|
||||
)
|
||||
path_metadata = _as_dict(source.get("path_metadata"))
|
||||
filename = _first_present(
|
||||
source.get("filename"),
|
||||
path_metadata.get("filename"),
|
||||
)
|
||||
file_info = _find_metadata_file_info(source, selected_version, str(filename))
|
||||
file_hashes = _as_dict(file_info.get("hashes"))
|
||||
file_sha256 = normalize_sha256(
|
||||
_first_present(
|
||||
file_info.get("sha256"),
|
||||
file_info.get("hash"),
|
||||
file_hashes.get("SHA256"),
|
||||
file_hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
if file_info:
|
||||
# An explicit file entry without a hash means this provider did not
|
||||
# declare a checksum for the selected URL. Do not fall back to a
|
||||
# stale source/workflow hash and reject a user-selected download.
|
||||
return file_sha256
|
||||
|
||||
# Keep the source-level fallback for providers that expose one hash but no
|
||||
# file object.
|
||||
hashes = _as_dict(source.get("hashes"))
|
||||
return normalize_sha256(
|
||||
_first_present(
|
||||
source.get("sha256"),
|
||||
source.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def read_completed_metadata_sha256(file_path: str) -> str:
|
||||
"""Read a trusted SHA256 from merged local metadata when available."""
|
||||
metadata_path = find_metadata_sidecar_path(file_path)
|
||||
payload = read_merged_model_metadata(file_path, {})
|
||||
if not isinstance(payload, dict) or not payload:
|
||||
return ""
|
||||
|
||||
hash_status = str(payload.get("hash_status") or "completed").strip().lower()
|
||||
if hash_status != "completed":
|
||||
log.debug(
|
||||
f"Skipping metadata SHA256 for {metadata_path}: hash_status={hash_status}"
|
||||
)
|
||||
return ""
|
||||
|
||||
return normalize_sha256(payload.get("sha256") or payload.get("hash"))
|
||||
|
||||
|
||||
def build_model_resolver_metadata(
|
||||
dest_path: str,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
category: str = "",
|
||||
source_url: str = "",
|
||||
) -> Dict[str, Any]:
|
||||
"""Build the LoRA Manager-shaped payload stored in our own sidecar."""
|
||||
source = _json_safe_metadata(metadata or {})
|
||||
if not isinstance(source, dict):
|
||||
source = {}
|
||||
|
||||
path_metadata = _as_dict(source.get("path_metadata"))
|
||||
details = _as_dict(source.get("civitai_details") or source.get("details"))
|
||||
selected_version = _as_dict(
|
||||
source.get("selected_version") or details.get("selected_version")
|
||||
)
|
||||
|
||||
basename = get_filename_from_path(dest_path)
|
||||
file_name = os.path.splitext(basename)[0]
|
||||
filename = _first_present(
|
||||
source.get("filename"),
|
||||
path_metadata.get("filename"),
|
||||
basename,
|
||||
)
|
||||
model_name = _first_present(
|
||||
source.get("model_name"),
|
||||
source.get("model"),
|
||||
source.get("name"),
|
||||
details.get("name"),
|
||||
path_metadata.get("model_name"),
|
||||
path_metadata.get("name"),
|
||||
os.path.splitext(str(filename))[0],
|
||||
file_name,
|
||||
)
|
||||
version_name = _first_present(
|
||||
source.get("version_name"),
|
||||
source.get("versionName"),
|
||||
source.get("version"),
|
||||
selected_version.get("name"),
|
||||
path_metadata.get("version_name"),
|
||||
)
|
||||
base_model = _first_present(
|
||||
source.get("base_model"),
|
||||
source.get("baseModel"),
|
||||
selected_version.get("base_model"),
|
||||
selected_version.get("baseModel"),
|
||||
path_metadata.get("base_model"),
|
||||
"Unknown",
|
||||
)
|
||||
source_name = str(
|
||||
_first_present(source.get("details_source"), source.get("source"), details.get("source"))
|
||||
or ""
|
||||
).lower()
|
||||
|
||||
model_id = _first_present(
|
||||
source.get("model_id"),
|
||||
source.get("modelId"),
|
||||
details.get("model_id"),
|
||||
path_metadata.get("model_id"),
|
||||
)
|
||||
version_id = _first_present(
|
||||
source.get("version_id"),
|
||||
source.get("versionId"),
|
||||
details.get("version_id"),
|
||||
selected_version.get("id"),
|
||||
path_metadata.get("version_id"),
|
||||
)
|
||||
|
||||
tags = _as_list(
|
||||
_first_present(
|
||||
source.get("tags"),
|
||||
details.get("tags"),
|
||||
path_metadata.get("tags"),
|
||||
)
|
||||
)
|
||||
trained_words = _as_list(
|
||||
_first_present(
|
||||
source.get("trained_words"),
|
||||
source.get("trainedWords"),
|
||||
selected_version.get("trained_words"),
|
||||
selected_version.get("trainedWords"),
|
||||
)
|
||||
)
|
||||
images = _as_list(
|
||||
_first_present(
|
||||
source.get("images"),
|
||||
selected_version.get("images"),
|
||||
details.get("images"),
|
||||
)
|
||||
)
|
||||
creator = _as_dict(
|
||||
_first_present(source.get("creator"), details.get("creator"), path_metadata.get("creator"))
|
||||
)
|
||||
file_info = _find_metadata_file_info(source, selected_version, str(filename))
|
||||
hashes = _as_dict(file_info.get("hashes") if file_info else source.get("hashes"))
|
||||
hash_values = (
|
||||
(
|
||||
file_info.get("sha256"),
|
||||
file_info.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
if file_info
|
||||
else (
|
||||
source.get("sha256"),
|
||||
source.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
sha256 = str(_first_present(*hash_values) or "").lower()
|
||||
direct_url = _first_present(
|
||||
source.get("download_url"),
|
||||
source.get("downloadUrl"),
|
||||
source.get("source_url"),
|
||||
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,
|
||||
)
|
||||
if source_name == "civarchive":
|
||||
civarchive_page_url = next(
|
||||
(
|
||||
str(value).strip()
|
||||
for value in (
|
||||
source.get("version_url"),
|
||||
source.get("model_url"),
|
||||
source.get("page_url"),
|
||||
source.get("source_url"),
|
||||
source.get("url"),
|
||||
)
|
||||
if value
|
||||
and host_matches_domain(urlparse(str(value)).hostname, "civarchive.com")
|
||||
),
|
||||
"",
|
||||
)
|
||||
if not civarchive_page_url and model_id:
|
||||
civarchive_page_url = f"https://civarchive.com/models/{model_id}"
|
||||
if version_id:
|
||||
civarchive_page_url += f"?modelVersionId={version_id}"
|
||||
if not civarchive_page_url and sha256:
|
||||
civarchive_page_url = f"https://civarchive.com/sha256/{sha256}"
|
||||
source_page_url = civarchive_page_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"),
|
||||
)
|
||||
explicit_model_description = _first_present(
|
||||
source.get("modelDescription"),
|
||||
source.get("model_description"),
|
||||
details.get("description"),
|
||||
_as_dict(source.get("model")).get("description"),
|
||||
)
|
||||
model_description = _first_present(
|
||||
explicit_model_description,
|
||||
source.get("description"),
|
||||
)
|
||||
version_description = _first_present(
|
||||
source.get("versionDescription"),
|
||||
source.get("version_description"),
|
||||
selected_version.get("description"),
|
||||
(
|
||||
source.get("description")
|
||||
if explicit_model_description
|
||||
and source.get("description") != explicit_model_description
|
||||
else None
|
||||
),
|
||||
)
|
||||
|
||||
is_civitai_source = source_name in {
|
||||
"civitai",
|
||||
"civarchive",
|
||||
"lora_manager_archive",
|
||||
} or (
|
||||
not source_name
|
||||
and bool(source.get("civitai") or source.get("civitai_details"))
|
||||
)
|
||||
civitai_payload = (
|
||||
dict(
|
||||
_as_dict(
|
||||
_first_present(
|
||||
source.get("civitai"),
|
||||
details.get("civitai"),
|
||||
)
|
||||
)
|
||||
)
|
||||
if is_civitai_source
|
||||
else {}
|
||||
)
|
||||
if is_civitai_source:
|
||||
if model_id and "modelId" not in civitai_payload:
|
||||
civitai_payload["modelId"] = _coerce_int_or_value(model_id)
|
||||
if version_id and "id" not in civitai_payload:
|
||||
civitai_payload["id"] = _coerce_int_or_value(version_id)
|
||||
if version_name and "name" not in civitai_payload:
|
||||
civitai_payload["name"] = str(version_name)
|
||||
if base_model and "baseModel" not in civitai_payload:
|
||||
civitai_payload["baseModel"] = str(base_model)
|
||||
if trained_words and "trainedWords" not in civitai_payload:
|
||||
civitai_payload["trainedWords"] = trained_words
|
||||
if images and "images" not in civitai_payload:
|
||||
civitai_payload["images"] = images
|
||||
if direct_url and "downloadUrl" not in civitai_payload:
|
||||
civitai_payload["downloadUrl"] = _strip_sensitive_url_params(str(direct_url))
|
||||
if version_description and "description" not in civitai_payload:
|
||||
civitai_payload["description"] = str(version_description)
|
||||
|
||||
files = _as_list(_first_present(selected_version.get("files"), source.get("files")))
|
||||
if is_civitai_source and files and "files" not in civitai_payload:
|
||||
civitai_payload["files"] = files
|
||||
|
||||
model_payload = dict(_as_dict(civitai_payload.get("model")))
|
||||
if model_name and "name" not in model_payload:
|
||||
model_payload["name"] = str(model_name)
|
||||
model_type = _first_present(source.get("type"), source.get("model_type"), details.get("type"))
|
||||
if model_type and "type" not in model_payload:
|
||||
model_payload["type"] = str(model_type)
|
||||
if model_description and "description" not in model_payload:
|
||||
model_payload["description"] = str(model_description)
|
||||
if tags and "tags" not in model_payload:
|
||||
model_payload["tags"] = tags
|
||||
if is_civitai_source and model_payload:
|
||||
civitai_payload["model"] = model_payload
|
||||
if is_civitai_source and creator and "creator" not in civitai_payload:
|
||||
civitai_payload["creator"] = creator
|
||||
|
||||
metadata_source = _metadata_source_value(source_name, source.get("metadata_source"))
|
||||
if os.path.exists(dest_path):
|
||||
size = os.path.getsize(dest_path)
|
||||
else:
|
||||
size = _coerce_size(_first_present(source.get("size"), file_info.get("size")))
|
||||
|
||||
payload: Dict[str, Any] = {
|
||||
"schema": MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
"schema_version": MODEL_RESOLVER_METADATA_SCHEMA_VERSION,
|
||||
"managed_by": MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
"file_name": file_name,
|
||||
"filename": basename,
|
||||
"model_name": str(model_name or file_name),
|
||||
"file_path": _normalise_metadata_file_path(dest_path),
|
||||
"size": size,
|
||||
"modified": time.time(),
|
||||
"sha256": sha256,
|
||||
"base_model": str(base_model or "Unknown"),
|
||||
"preview_url": str(preview_url or ""),
|
||||
"preview_nsfw_level": 0,
|
||||
"from_civitai": bool(
|
||||
is_civitai_source
|
||||
and (metadata_source or model_id or version_id or civitai_payload)
|
||||
),
|
||||
"civitai": civitai_payload,
|
||||
"tags": tags,
|
||||
"modelDescription": str(model_description or ""),
|
||||
"version_description": str(version_description or ""),
|
||||
"civitai_deleted": bool(source.get("is_deleted") or source.get("civitai_deleted")),
|
||||
"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",
|
||||
}
|
||||
|
||||
lora_manager_type = _resolve_lora_manager_model_type(category, model_type)
|
||||
if normalize_download_category(category) == "loras":
|
||||
payload["usage_tips"] = str(source.get("usage_tips") or "{}")
|
||||
elif lora_manager_type:
|
||||
payload["model_type"] = lora_manager_type
|
||||
payload["sub_type"] = lora_manager_type
|
||||
|
||||
return payload
|
||||
+17
-483
@@ -31,27 +31,21 @@ from .network_utils import (
|
||||
validate_public_http_url,
|
||||
)
|
||||
from .path_utils import (
|
||||
MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
MODEL_RESOLVER_METADATA_SCHEMA_VERSION,
|
||||
calculate_file_sha256,
|
||||
find_metadata_sidecar_path,
|
||||
get_comfy_root_path,
|
||||
get_filename_from_path,
|
||||
get_model_resolver_sidecar_path,
|
||||
get_path_identity,
|
||||
is_path_within,
|
||||
read_merged_model_metadata,
|
||||
write_json_atomic,
|
||||
)
|
||||
from .resolver import invalidate_local_hash_match_cache, normalize_sha256
|
||||
from .resolver import invalidate_local_hash_match_cache
|
||||
from .scanner import invalidate_model_files_cache
|
||||
from .type_utils import (
|
||||
as_dict,
|
||||
as_list,
|
||||
extract_response_file_size,
|
||||
first_non_empty,
|
||||
get_category_folder_keys,
|
||||
normalize_category_to_model_type,
|
||||
)
|
||||
from .type_utils import format_size_bytes as format_bytes
|
||||
|
||||
@@ -111,10 +105,22 @@ HF_XET_ARIA2_AUTH_HOSTS = {
|
||||
"cas-bridge-direct.xethub-eu.hf.co",
|
||||
}
|
||||
|
||||
from .download.metadata import (
|
||||
_coerce_int_or_value, # noqa: F401
|
||||
_coerce_size, # noqa: F401
|
||||
_extract_expected_sha256,
|
||||
_find_metadata_file_info, # noqa: F401
|
||||
_json_safe_metadata, # noqa: F401
|
||||
_metadata_source_value, # noqa: F401
|
||||
_normalise_metadata_file_path,
|
||||
_resolve_lora_manager_model_type, # noqa: F401
|
||||
build_model_resolver_metadata,
|
||||
read_completed_metadata_sha256,
|
||||
)
|
||||
from .download.validation import (
|
||||
DOWNLOAD_USER_AGENT,
|
||||
_get_header_value,
|
||||
_is_sensitive_metadata_key,
|
||||
_is_sensitive_metadata_key, # noqa: F401
|
||||
_sanitize_download_error,
|
||||
_strip_sensitive_url_params,
|
||||
build_download_headers,
|
||||
@@ -124,10 +130,12 @@ from .download.validation import (
|
||||
from .settings import (
|
||||
load_settings,
|
||||
normalize_download_backend,
|
||||
normalize_download_category,
|
||||
normalize_relative_subfolder,
|
||||
)
|
||||
|
||||
_as_dict = as_dict
|
||||
_as_list = as_list
|
||||
|
||||
|
||||
def _resolve_download_url_for_aria2(
|
||||
url: str,
|
||||
@@ -159,480 +167,6 @@ def _resolve_download_url_for_aria2(
|
||||
response.close()
|
||||
|
||||
|
||||
def _json_safe_metadata(value: Any, depth: int = 0) -> Any:
|
||||
if depth > 10:
|
||||
return str(value)
|
||||
|
||||
if isinstance(value, dict):
|
||||
cleaned = {}
|
||||
for key, item_value in value.items():
|
||||
if _is_sensitive_metadata_key(key):
|
||||
continue
|
||||
cleaned[str(key)] = _json_safe_metadata(item_value, depth + 1)
|
||||
return cleaned
|
||||
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
return [_json_safe_metadata(item, depth + 1) for item in value]
|
||||
|
||||
if isinstance(value, str):
|
||||
return _strip_sensitive_url_params(value)
|
||||
|
||||
if value is None or isinstance(value, (bool, int, float)):
|
||||
return value
|
||||
|
||||
return str(value)
|
||||
|
||||
|
||||
_as_dict = as_dict
|
||||
_as_list = as_list
|
||||
_first_present = first_non_empty
|
||||
|
||||
|
||||
def _coerce_int_or_value(value: Any) -> Any:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return value
|
||||
|
||||
|
||||
def _coerce_size(value: Any) -> int:
|
||||
try:
|
||||
return int(float(value or 0))
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def _normalise_metadata_file_path(path_value: str) -> str:
|
||||
return str(path_value or "").replace(os.sep, "/")
|
||||
|
||||
|
||||
def _resolve_lora_manager_model_type(category: str, source_type: Any = "") -> str:
|
||||
res = normalize_category_to_model_type(category)
|
||||
if res in ("checkpoint", "diffusion_model", "embedding"):
|
||||
return res
|
||||
|
||||
source_token = (
|
||||
str(source_type or "")
|
||||
.strip()
|
||||
.lower()
|
||||
.replace(" ", "_")
|
||||
.replace("-", "_")
|
||||
)
|
||||
if "checkpoint" in source_token:
|
||||
return "checkpoint"
|
||||
if "diffusion" in source_token or source_token == "unet":
|
||||
return "diffusion_model"
|
||||
if "textual" in source_token or "embedding" in source_token:
|
||||
return "embedding"
|
||||
return ""
|
||||
|
||||
|
||||
def _metadata_source_value(source_name: str, existing: Any = None) -> Optional[str]:
|
||||
if existing:
|
||||
return str(existing)
|
||||
|
||||
source_token = str(source_name or "").strip().lower()
|
||||
if source_token == "civitai":
|
||||
return "civitai_api"
|
||||
if source_token == "civarchive":
|
||||
return "civarchive"
|
||||
if source_token == "lora_manager_archive":
|
||||
return "archive_db"
|
||||
return None
|
||||
|
||||
|
||||
def _find_metadata_file_info(
|
||||
source: Dict[str, Any],
|
||||
selected_version: Dict[str, Any],
|
||||
filename: str,
|
||||
) -> Dict[str, Any]:
|
||||
# A selected file is the most specific provenance available. It must be
|
||||
# checked before the source-level hash because the latter can still belong
|
||||
# to the workflow's original model after the user chooses another variant.
|
||||
for key in ("selected_file", "selectedFile", "file_info", "file"):
|
||||
value = source.get(key)
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
|
||||
filename_lower = get_filename_from_path(str(filename or "")).lower()
|
||||
file_lists = [
|
||||
source.get("files"),
|
||||
selected_version.get("files"),
|
||||
]
|
||||
for file_list in file_lists:
|
||||
if not isinstance(file_list, list):
|
||||
continue
|
||||
first_file = None
|
||||
for file_info in file_list:
|
||||
if not isinstance(file_info, dict):
|
||||
continue
|
||||
if first_file is None:
|
||||
first_file = file_info
|
||||
candidate = str(
|
||||
file_info.get("name")
|
||||
or file_info.get("filename")
|
||||
or file_info.get("fileName")
|
||||
or ""
|
||||
).lower()
|
||||
if filename_lower and candidate == filename_lower:
|
||||
return file_info
|
||||
if first_file:
|
||||
return first_file
|
||||
return {}
|
||||
|
||||
|
||||
def _extract_expected_sha256(metadata: Optional[Dict[str, Any]]) -> str:
|
||||
source = metadata if isinstance(metadata, dict) else {}
|
||||
details = _as_dict(source.get("civitai_details") or source.get("details"))
|
||||
selected_version = _as_dict(
|
||||
source.get("selected_version") or details.get("selected_version")
|
||||
)
|
||||
path_metadata = _as_dict(source.get("path_metadata"))
|
||||
filename = _first_present(
|
||||
source.get("filename"),
|
||||
path_metadata.get("filename"),
|
||||
)
|
||||
file_info = _find_metadata_file_info(source, selected_version, str(filename))
|
||||
file_hashes = _as_dict(file_info.get("hashes"))
|
||||
file_sha256 = normalize_sha256(
|
||||
_first_present(
|
||||
file_info.get("sha256"),
|
||||
file_info.get("hash"),
|
||||
file_hashes.get("SHA256"),
|
||||
file_hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
if file_info:
|
||||
# An explicit file entry without a hash means this provider did not
|
||||
# declare a checksum for the selected URL. Do not fall back to a
|
||||
# stale source/workflow hash and reject a user-selected download.
|
||||
return file_sha256
|
||||
|
||||
# Keep the source-level fallback for providers that expose one hash but no
|
||||
# file object.
|
||||
hashes = _as_dict(source.get("hashes"))
|
||||
return normalize_sha256(
|
||||
_first_present(
|
||||
source.get("sha256"),
|
||||
source.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def read_completed_metadata_sha256(file_path: str) -> str:
|
||||
"""Read a trusted SHA256 from merged local metadata when available."""
|
||||
metadata_path = find_metadata_sidecar_path(file_path)
|
||||
payload = read_merged_model_metadata(file_path, {})
|
||||
if not isinstance(payload, dict) or not payload:
|
||||
return ""
|
||||
|
||||
hash_status = str(payload.get("hash_status") or "completed").strip().lower()
|
||||
if hash_status != "completed":
|
||||
log.debug(
|
||||
f"Skipping metadata SHA256 for {metadata_path}: hash_status={hash_status}"
|
||||
)
|
||||
return ""
|
||||
|
||||
return normalize_sha256(payload.get("sha256") or payload.get("hash"))
|
||||
|
||||
|
||||
def build_model_resolver_metadata(
|
||||
dest_path: str,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
category: str = "",
|
||||
source_url: str = "",
|
||||
) -> Dict[str, Any]:
|
||||
"""Build the LoRA Manager-shaped payload stored in our own sidecar."""
|
||||
source = _json_safe_metadata(metadata or {})
|
||||
if not isinstance(source, dict):
|
||||
source = {}
|
||||
|
||||
path_metadata = _as_dict(source.get("path_metadata"))
|
||||
details = _as_dict(source.get("civitai_details") or source.get("details"))
|
||||
selected_version = _as_dict(
|
||||
source.get("selected_version") or details.get("selected_version")
|
||||
)
|
||||
|
||||
basename = get_filename_from_path(dest_path)
|
||||
file_name = os.path.splitext(basename)[0]
|
||||
filename = _first_present(
|
||||
source.get("filename"),
|
||||
path_metadata.get("filename"),
|
||||
basename,
|
||||
)
|
||||
model_name = _first_present(
|
||||
source.get("model_name"),
|
||||
source.get("model"),
|
||||
source.get("name"),
|
||||
details.get("name"),
|
||||
path_metadata.get("model_name"),
|
||||
path_metadata.get("name"),
|
||||
os.path.splitext(str(filename))[0],
|
||||
file_name,
|
||||
)
|
||||
version_name = _first_present(
|
||||
source.get("version_name"),
|
||||
source.get("versionName"),
|
||||
source.get("version"),
|
||||
selected_version.get("name"),
|
||||
path_metadata.get("version_name"),
|
||||
)
|
||||
base_model = _first_present(
|
||||
source.get("base_model"),
|
||||
source.get("baseModel"),
|
||||
selected_version.get("base_model"),
|
||||
selected_version.get("baseModel"),
|
||||
path_metadata.get("base_model"),
|
||||
"Unknown",
|
||||
)
|
||||
source_name = str(
|
||||
_first_present(source.get("details_source"), source.get("source"), details.get("source"))
|
||||
or ""
|
||||
).lower()
|
||||
|
||||
model_id = _first_present(
|
||||
source.get("model_id"),
|
||||
source.get("modelId"),
|
||||
details.get("model_id"),
|
||||
path_metadata.get("model_id"),
|
||||
)
|
||||
version_id = _first_present(
|
||||
source.get("version_id"),
|
||||
source.get("versionId"),
|
||||
details.get("version_id"),
|
||||
selected_version.get("id"),
|
||||
path_metadata.get("version_id"),
|
||||
)
|
||||
|
||||
tags = _as_list(
|
||||
_first_present(
|
||||
source.get("tags"),
|
||||
details.get("tags"),
|
||||
path_metadata.get("tags"),
|
||||
)
|
||||
)
|
||||
trained_words = _as_list(
|
||||
_first_present(
|
||||
source.get("trained_words"),
|
||||
source.get("trainedWords"),
|
||||
selected_version.get("trained_words"),
|
||||
selected_version.get("trainedWords"),
|
||||
)
|
||||
)
|
||||
images = _as_list(
|
||||
_first_present(
|
||||
source.get("images"),
|
||||
selected_version.get("images"),
|
||||
details.get("images"),
|
||||
)
|
||||
)
|
||||
creator = _as_dict(
|
||||
_first_present(source.get("creator"), details.get("creator"), path_metadata.get("creator"))
|
||||
)
|
||||
file_info = _find_metadata_file_info(source, selected_version, str(filename))
|
||||
hashes = _as_dict(file_info.get("hashes") if file_info else source.get("hashes"))
|
||||
hash_values = (
|
||||
(
|
||||
file_info.get("sha256"),
|
||||
file_info.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
if file_info
|
||||
else (
|
||||
source.get("sha256"),
|
||||
source.get("hash"),
|
||||
hashes.get("SHA256"),
|
||||
hashes.get("sha256"),
|
||||
)
|
||||
)
|
||||
sha256 = str(
|
||||
_first_present(*hash_values)
|
||||
or ""
|
||||
).lower()
|
||||
direct_url = _first_present(
|
||||
source.get("download_url"),
|
||||
source.get("downloadUrl"),
|
||||
source.get("source_url"),
|
||||
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,
|
||||
)
|
||||
if source_name == "civarchive":
|
||||
civarchive_page_url = next(
|
||||
(
|
||||
str(value).strip()
|
||||
for value in (
|
||||
source.get("version_url"),
|
||||
source.get("model_url"),
|
||||
source.get("page_url"),
|
||||
source.get("source_url"),
|
||||
source.get("url"),
|
||||
)
|
||||
if value
|
||||
and host_matches_domain(urlparse(str(value)).hostname, "civarchive.com")
|
||||
),
|
||||
"",
|
||||
)
|
||||
if not civarchive_page_url and model_id:
|
||||
civarchive_page_url = f"https://civarchive.com/models/{model_id}"
|
||||
if version_id:
|
||||
civarchive_page_url += f"?modelVersionId={version_id}"
|
||||
if not civarchive_page_url and sha256:
|
||||
civarchive_page_url = f"https://civarchive.com/sha256/{sha256}"
|
||||
source_page_url = civarchive_page_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"),
|
||||
)
|
||||
explicit_model_description = _first_present(
|
||||
source.get("modelDescription"),
|
||||
source.get("model_description"),
|
||||
details.get("description"),
|
||||
_as_dict(source.get("model")).get("description"),
|
||||
)
|
||||
model_description = _first_present(
|
||||
explicit_model_description,
|
||||
source.get("description"),
|
||||
)
|
||||
version_description = _first_present(
|
||||
source.get("versionDescription"),
|
||||
source.get("version_description"),
|
||||
selected_version.get("description"),
|
||||
(
|
||||
source.get("description")
|
||||
if explicit_model_description
|
||||
and source.get("description") != explicit_model_description
|
||||
else None
|
||||
),
|
||||
)
|
||||
|
||||
is_civitai_source = source_name in {
|
||||
"civitai",
|
||||
"civarchive",
|
||||
"lora_manager_archive",
|
||||
} or (
|
||||
not source_name
|
||||
and bool(source.get("civitai") or source.get("civitai_details"))
|
||||
)
|
||||
civitai_payload = (
|
||||
dict(
|
||||
_as_dict(
|
||||
_first_present(
|
||||
source.get("civitai"),
|
||||
details.get("civitai"),
|
||||
)
|
||||
)
|
||||
)
|
||||
if is_civitai_source
|
||||
else {}
|
||||
)
|
||||
if is_civitai_source:
|
||||
if model_id and "modelId" not in civitai_payload:
|
||||
civitai_payload["modelId"] = _coerce_int_or_value(model_id)
|
||||
if version_id and "id" not in civitai_payload:
|
||||
civitai_payload["id"] = _coerce_int_or_value(version_id)
|
||||
if version_name and "name" not in civitai_payload:
|
||||
civitai_payload["name"] = str(version_name)
|
||||
if base_model and "baseModel" not in civitai_payload:
|
||||
civitai_payload["baseModel"] = str(base_model)
|
||||
if trained_words and "trainedWords" not in civitai_payload:
|
||||
civitai_payload["trainedWords"] = trained_words
|
||||
if images and "images" not in civitai_payload:
|
||||
civitai_payload["images"] = images
|
||||
if direct_url and "downloadUrl" not in civitai_payload:
|
||||
civitai_payload["downloadUrl"] = _strip_sensitive_url_params(
|
||||
str(direct_url)
|
||||
)
|
||||
if version_description and "description" not in civitai_payload:
|
||||
civitai_payload["description"] = str(version_description)
|
||||
|
||||
files = _as_list(_first_present(selected_version.get("files"), source.get("files")))
|
||||
if is_civitai_source and files and "files" not in civitai_payload:
|
||||
civitai_payload["files"] = files
|
||||
|
||||
model_payload = dict(_as_dict(civitai_payload.get("model")))
|
||||
if model_name and "name" not in model_payload:
|
||||
model_payload["name"] = str(model_name)
|
||||
model_type = _first_present(source.get("type"), source.get("model_type"), details.get("type"))
|
||||
if model_type and "type" not in model_payload:
|
||||
model_payload["type"] = str(model_type)
|
||||
if model_description and "description" not in model_payload:
|
||||
model_payload["description"] = str(model_description)
|
||||
if tags and "tags" not in model_payload:
|
||||
model_payload["tags"] = tags
|
||||
if is_civitai_source and model_payload:
|
||||
civitai_payload["model"] = model_payload
|
||||
if is_civitai_source and creator and "creator" not in civitai_payload:
|
||||
civitai_payload["creator"] = creator
|
||||
|
||||
metadata_source = _metadata_source_value(source_name, source.get("metadata_source"))
|
||||
if os.path.exists(dest_path):
|
||||
size = os.path.getsize(dest_path)
|
||||
else:
|
||||
size = _coerce_size(_first_present(source.get("size"), file_info.get("size")))
|
||||
|
||||
payload: Dict[str, Any] = {
|
||||
"schema": MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
"schema_version": MODEL_RESOLVER_METADATA_SCHEMA_VERSION,
|
||||
"managed_by": MODEL_RESOLVER_METADATA_SCHEMA,
|
||||
"file_name": file_name,
|
||||
"filename": basename,
|
||||
"model_name": str(model_name or file_name),
|
||||
"file_path": _normalise_metadata_file_path(dest_path),
|
||||
"size": size,
|
||||
"modified": time.time(),
|
||||
"sha256": sha256,
|
||||
"base_model": str(base_model or "Unknown"),
|
||||
"preview_url": str(preview_url or ""),
|
||||
"preview_nsfw_level": 0,
|
||||
"from_civitai": bool(
|
||||
is_civitai_source
|
||||
and (metadata_source or model_id or version_id or civitai_payload)
|
||||
),
|
||||
"civitai": civitai_payload,
|
||||
"tags": tags,
|
||||
"modelDescription": str(model_description or ""),
|
||||
"version_description": str(version_description or ""),
|
||||
"civitai_deleted": bool(source.get("is_deleted") or source.get("civitai_deleted")),
|
||||
"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",
|
||||
}
|
||||
|
||||
lora_manager_type = _resolve_lora_manager_model_type(category, model_type)
|
||||
if normalize_download_category(category) == "loras":
|
||||
payload["usage_tips"] = str(source.get("usage_tips") or "{}")
|
||||
elif lora_manager_type:
|
||||
payload["model_type"] = lora_manager_type
|
||||
payload["sub_type"] = lora_manager_type
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def get_existing_model_preview_path(dest_path: str) -> str:
|
||||
"""Return the first adjacent preview media file available for a model."""
|
||||
model_base_path, _model_ext = os.path.splitext(dest_path)
|
||||
|
||||
Reference in New Issue
Block a user