Files
ComfyUI-Lora-Manager/py/services/metadata_sync_service.py
T
Will Miao acefe1a348 fix(metadata): fail loudly when the sidecar write cannot land
update_model_metadata ignored the save_metadata result, so a fetch on a
drive that went offline mid-run was counted as success: the cache got
the fresh CivitAI payload while the durable sidecar never landed, and
every later fetch skipped the model as already fetched.

The save result now raises MetadataPersistError. In the bulk fetch path
this fails the item (reported in the summary, cache untouched, model
stays eligible for the next run); relink surfaces an honest 500.

Also corrects the stale known-limitation note in the scoped-scan plan:
a startup-offline root has been listed and re-admitted since Wave 6.
2026-10-08 07:55:47 +08:00

677 lines
29 KiB
Python

"""Services for synchronising metadata with remote providers."""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime
from typing import Any, Awaitable, Callable, Dict, Iterable, Optional, Protocol
from ..services.settings_manager import SettingsManager
from ..utils.civitai_utils import resolve_license_payload
from ..utils.model_utils import determine_base_model
from ..utils.models import autov3_from_civitai_files
from ..utils.sidecar_paths import get_metadata_path
from .connectivity_guard import OFFLINE_FRIENDLY_MESSAGE, is_expected_offline_error
from .errors import MetadataPersistError, RateLimitError
from .model_metadata_provider import _LOCAL_PROVIDER_LABELS
from .model_sources import get_source_platform, has_external_source
logger = logging.getLogger(__name__)
# Providers restricted to specific model sub_types, keyed by their
# registration label. Providers not listed apply to every model type.
# OpenModelDB only indexes upscalers, so it is never consulted for other model
# types — that keeps its one-time bulk-catalogue download from being paid by
# users who manage no upscalers at all.
_PROVIDER_SUB_TYPE_RESTRICTIONS: Dict[str, frozenset] = {
"openmodeldb_api": frozenset({"upscaler"}),
}
#: External-source platforms that have their own hash-lookup metadata
#: provider. A model downloaded from one of these is refreshed against the
#: source's own catalogue first (upgrading the download-time card to the full
#: payload) before CivitAI is consulted at all.
_EXTERNAL_SOURCE_METADATA_PROVIDERS: Dict[str, str] = {
"openmodeldb": "openmodeldb_api",
}
def _restricted_providers_for_sub_type(sub_type: Optional[str]) -> list:
"""Return the restricted provider labels that apply to ``sub_type``."""
return [
name
for name, allowed in _PROVIDER_SUB_TYPE_RESTRICTIONS.items()
if sub_type in allowed
]
def _inapplicable_providers_for_sub_type(sub_type: Optional[str]) -> set:
"""Return the restricted provider labels that do NOT apply to ``sub_type``."""
return {
name
for name, allowed in _PROVIDER_SUB_TYPE_RESTRICTIONS.items()
if sub_type not in allowed
}
def _merge_ordered_unique(existing: Iterable[str], new: Iterable[str]) -> list[str]:
"""Concatenate two word lists, dropping duplicates without reordering.
Trigger word order is meaningful: the sequence stored in
``civitai.trainedWords`` is the order used when building prompts, and users
can reorder it in the UI. A plain ``set`` union used to shuffle that order on
every metadata refresh, so existing words are kept first (in their saved
order) and newly discovered ones are appended.
"""
merged: list[str] = []
seen: set[str] = set()
for word in list(existing) + list(new):
if word in seen:
continue
seen.add(word)
merged.append(word)
return merged
class MetadataProviderProtocol(Protocol):
"""Subset of metadata provider interface consumed by the sync service."""
async def get_model_by_hash(self, model_hash: str) -> tuple[Optional[Dict[str, Any]], Optional[str]]:
...
async def get_model_version(
self, model_id: Any = None, version_id: Any = None
) -> Optional[Dict[str, Any]]:
...
class MetadataSyncService:
"""High level orchestration for metadata synchronisation flows."""
def __init__(
self,
*,
metadata_manager,
preview_service,
settings: SettingsManager,
default_metadata_provider_factory: Callable[..., Awaitable[MetadataProviderProtocol]],
metadata_provider_selector: Callable[..., Awaitable[MetadataProviderProtocol]],
) -> None:
self._metadata_manager = metadata_manager
self._preview_service = preview_service
self._settings = settings
self._get_default_provider = default_metadata_provider_factory
self._get_provider = metadata_provider_selector
async def load_local_metadata(self, metadata_path: str) -> Dict[str, Any]:
"""Load metadata JSON from disk, returning an empty structure when missing."""
if not os.path.exists(metadata_path):
return {}
try:
with open(metadata_path, "r", encoding="utf-8") as handle:
return json.load(handle)
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Error loading metadata from %s: %s", metadata_path, exc)
return {}
async def mark_not_found_on_civitai(
self, metadata_path: str, local_metadata: Dict[str, Any]
) -> None:
"""Persist the not-found flag for a metadata payload."""
local_metadata["from_civitai"] = False
await self._metadata_manager.save_metadata(metadata_path, local_metadata)
@staticmethod
def is_civitai_api_metadata(meta: Dict[str, Any]) -> bool:
"""Determine if the metadata originated from the CivitAI public API."""
if not isinstance(meta, dict):
return False
files = meta.get("files")
images = meta.get("images")
source = meta.get("source")
return bool(files) and bool(images) and source not in ("archive_db", "civarchive")
async def update_model_metadata(
self,
metadata_path: str,
local_metadata: Dict[str, Any],
civitai_metadata: Dict[str, Any],
metadata_provider: Optional[MetadataProviderProtocol] = None,
) -> Dict[str, Any]:
"""Merge remote metadata into the local record and persist the result."""
existing_civitai = local_metadata.get("civitai") or {}
if (
not self.is_civitai_api_metadata(civitai_metadata)
and self.is_civitai_api_metadata(existing_civitai)
):
logger.info(
"Skip civitai update for %s (%s) - existing metadata is higher quality",
local_metadata.get("model_name", ""),
existing_civitai.get("name", ""),
)
else:
merged_civitai = existing_civitai.copy()
merged_civitai.update(civitai_metadata)
if civitai_metadata.get("source") == "archive_db":
model_name = civitai_metadata.get("model", {}).get("name", "")
version_name = civitai_metadata.get("name", "")
logger.info(
"Recovered metadata from archive_db for deleted model: %s (%s)",
model_name,
version_name,
)
if "trainedWords" in existing_civitai:
existing_trained = existing_civitai.get("trainedWords", []) or []
new_trained = civitai_metadata.get("trainedWords", []) or []
# Order preserving merge: the saved order drives prompt order.
merged_trained = _merge_ordered_unique(existing_trained, new_trained)
merged_civitai["trainedWords"] = merged_trained
local_metadata["civitai"] = merged_civitai
if "model" in civitai_metadata and civitai_metadata["model"]:
model_data = civitai_metadata["model"]
if model_data.get("name"):
local_metadata["model_name"] = model_data["name"]
if not local_metadata.get("modelDescription") and model_data.get("description"):
local_metadata["modelDescription"] = model_data["description"]
if not local_metadata.get("tags") and model_data.get("tags"):
local_metadata["tags"] = model_data["tags"]
if model_data.get("creator") and not local_metadata.get("civitai", {}).get(
"creator"
):
local_metadata.setdefault("civitai", {})["creator"] = model_data["creator"]
merged_civitai = local_metadata.get("civitai") or {}
civitai_model = merged_civitai.get("model")
if not isinstance(civitai_model, dict):
civitai_model = {}
license_payload = resolve_license_payload(model_data)
civitai_model.update(license_payload)
merged_civitai["model"] = civitai_model
local_metadata["civitai"] = merged_civitai
local_metadata["base_model"] = determine_base_model(
civitai_metadata.get("baseModel")
)
# Civitai-first AutoV3 propagation: the freshly fetched version
# metadata may report an AutoV3 for the file whose SHA256 matches the
# local model. Persist it now so recipe matching sees it immediately —
# no full rescan or restart required (the header is never re-read to
# upgrade the checked-unavailable '' state).
sha256_value = (local_metadata.get("sha256") or "").lower()
civitai_autov3 = autov3_from_civitai_files(
local_metadata.get("civitai"), sha256_value
)
if civitai_autov3:
local_metadata["autov3"] = civitai_autov3
await self._preview_service.ensure_preview_for_metadata(
metadata_path, local_metadata, civitai_metadata.get("images", [])
)
saved = await self._metadata_manager.save_metadata(metadata_path, local_metadata)
if not saved:
# A swallowed write failure would update the cache while the
# durable sidecar never lands (e.g. the drive went offline), and
# the model would be skipped by every later fetch as "already
# fetched". Fail loudly instead so the caller reports the item
# and it stays eligible for the next run.
raise MetadataPersistError(
f"Failed to write metadata sidecar: {metadata_path}"
)
return local_metadata
async def fetch_and_update_model(
self,
*,
sha256: str,
file_path: str,
model_data: Dict[str, Any],
update_cache_func: Callable[[str, str, Dict[str, Any]], Awaitable[bool]],
) -> tuple[bool, Optional[str]]:
"""Fetch metadata for a model and update both disk and cache state.
Callers should hydrate ``model_data`` via ``MetadataManager.hydrate_model_data``
before invoking this method so that the persisted payload retains all known
metadata fields.
"""
if not isinstance(model_data, dict):
error = f"Invalid model_data type: {type(model_data)}"
logger.error(error)
return False, error
metadata_path = get_metadata_path(file_path)
enable_archive = self._settings.get("enable_metadata_archive_db", False)
previous_source = model_data.get("metadata_source") or (model_data.get("civitai") or {}).get("source")
try:
provider_attempts: list[tuple[Optional[str], MetadataProviderProtocol]] = []
sqlite_attempted = False
if model_data.get("civitai_deleted") is True:
# Sub_type-restricted providers (e.g. OpenModelDB for
# upscalers) stay reachable for deleted models: their
# catalogues grow independently of CivitAI, so a model deleted
# from CivitAI may still gain metadata there later.
for restricted_name in _restricted_providers_for_sub_type(
model_data.get("sub_type")
):
try:
provider_attempts.append((restricted_name, await self._get_provider(restricted_name)))
except Exception as exc: # pragma: no cover - provider resolution fault
logger.debug("Unable to resolve %s provider: %s", restricted_name, exc)
if previous_source in (None, "civarchive"):
try:
provider_attempts.append(("civarchive_api", await self._get_provider("civarchive_api")))
except Exception as exc: # pragma: no cover - provider resolution fault
logger.debug("Unable to resolve civarchive provider: %s", exc)
if enable_archive and model_data.get("db_checked") is not True:
try:
provider_attempts.append(("sqlite", await self._get_provider("sqlite")))
except Exception as exc: # pragma: no cover - provider resolution fault
logger.debug("Unable to resolve sqlite provider: %s", exc)
if not provider_attempts:
if not enable_archive:
error_msg = "CivitAI model is deleted and metadata archive DB is not enabled"
elif model_data.get("db_checked") is True:
error_msg = "CivitAI model is deleted and not found in metadata archive DB"
else:
error_msg = "CivitAI model is deleted and no archive provider is available"
return False, error_msg
else:
is_hf_source = has_external_source(model_data)
if is_hf_source:
# External-source model (Hugging Face / ModelScope /
# TensorArt / OpenModelDB): a source with its own
# hash-lookup provider (OpenModelDB) is consulted first,
# then CivitAI API directly. CivArchive is almost
# guaranteed to have no record, and hitting it wastes
# rate-limit budget.
# Use a distinct provider name ("civitai_api" not None) so
# downstream code does NOT interpret a "Model not found"
# response as civitai_api_not_found — which would mark the
# model civitai_deleted=True when it was never on CivitAI.
source_provider = _EXTERNAL_SOURCE_METADATA_PROVIDERS.get(
get_source_platform(model_data)
)
provider_names = (
[source_provider, "civitai_api"]
if source_provider
else ["civitai_api"]
)
for provider_name in provider_names:
try:
provider_attempts.append(
(provider_name, await self._get_provider(provider_name))
)
except Exception as exc: # pragma: no cover - provider resolution fault
logger.debug(
"Unable to resolve %s provider: %s", provider_name, exc
)
if not provider_attempts:
default_provider = await self._get_default_provider()
# Drop sub_type-restricted providers that cannot apply to
# this model (e.g. OpenModelDB only indexes upscalers), so
# their cold-start cost is never paid pointlessly.
inapplicable = _inapplicable_providers_for_sub_type(
model_data.get("sub_type")
)
excluding = getattr(default_provider, "excluding", None)
if inapplicable and callable(excluding):
default_provider = excluding(inapplicable)
provider_attempts.append((None, default_provider))
civitai_metadata: Optional[Dict[str, Any]] = None
metadata_provider: Optional[MetadataProviderProtocol] = None
provider_used: Optional[str] = None
last_error: Optional[str] = None
civitai_api_not_found = False
any_rate_limited = False
skip_network_providers = False
for provider_name, provider in provider_attempts:
if skip_network_providers and provider_name not in _LOCAL_PROVIDER_LABELS:
# A network provider was already rate-limited; failing
# over to another network provider just spreads the flood
# (#1085). Local lookups (sqlite archive, the cached
# OpenModelDB index) stay available as a last resort.
continue
try:
civitai_metadata_candidate, error = await provider.get_model_by_hash(sha256)
except RateLimitError as exc:
logger.warning(
"Provider %s is rate-limited (retry_after=%.0fs); not failing over to other network providers",
provider_name or provider.__class__.__name__,
exc.retry_after or 0,
)
any_rate_limited = True
skip_network_providers = True
continue
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Provider %s failed for hash %s: %s", provider_name, sha256, exc)
civitai_metadata_candidate, error = None, str(exc)
if provider_name == "sqlite":
sqlite_attempted = True
is_default_provider = provider_name is None
if civitai_metadata_candidate:
civitai_metadata = civitai_metadata_candidate
metadata_provider = provider
provider_used = provider_name
break
if is_default_provider and error == "Model not found":
civitai_api_not_found = True
last_error = error or last_error
if civitai_metadata is None or metadata_provider is None:
# Track if we need to save metadata
needs_save = False
if sqlite_attempted:
model_data["db_checked"] = True
needs_save = True
if civitai_api_not_found:
model_data["from_civitai"] = False
model_data["civitai_deleted"] = True
model_data["db_checked"] = sqlite_attempted or (enable_archive and model_data.get("db_checked", False))
model_data["last_checked_at"] = datetime.now().timestamp()
needs_save = True
# When the model was already classified as "not on CivitAI" via
# .metadata.json (civitai_deleted=True) but the SQLite cache is
# stale (because the pre-fix code never persisted these flags),
# ensure the flags are written to the scanner cache + SQLite.
if not needs_save and model_data.get("civitai_deleted") is True:
model_data["last_checked_at"] = datetime.now().timestamp()
needs_save = True
# Save metadata if any state was updated
if needs_save:
data_to_save = model_data.copy()
data_to_save.pop("folder", None)
# Update last_checked_at for sqlite-only attempts if not already set
if "last_checked_at" not in data_to_save:
data_to_save["last_checked_at"] = datetime.now().timestamp()
await self._metadata_manager.save_metadata(file_path, data_to_save)
await update_cache_func(file_path, file_path, data_to_save)
default_error = (
"CivitAI model is deleted and metadata archive DB is not enabled"
if model_data.get("civitai_deleted") and not enable_archive
else "CivitAI model is deleted and not found in metadata archive DB"
if model_data.get("civitai_deleted") and (model_data.get("db_checked") is True or sqlite_attempted)
else "No provider returned metadata"
)
resolved_error = last_error or default_error
if any_rate_limited and "Rate limited" not in resolved_error:
resolved_error = "Rate limited"
if is_expected_offline_error(resolved_error):
resolved_error = OFFLINE_FRIENDLY_MESSAGE
error_msg = (
f"Error fetching metadata: {resolved_error} "
f"(file={os.path.basename(file_path)}, sha256={sha256})"
)
# Use case layer (BulkMetadataRefreshUseCase) logs failed models at WARNING level,
# so this level is demoted to DEBUG to avoid duplicate user-visible logging.
logger.debug(error_msg)
return False, error_msg
model_data["from_civitai"] = True
if provider_used is None:
model_data["civitai_deleted"] = False
elif civitai_api_not_found:
model_data["civitai_deleted"] = True
model_data["db_checked"] = enable_archive and (
civitai_metadata.get("source") == "archive_db" or sqlite_attempted
)
source = civitai_metadata.get("source") or "civitai_api"
if source == "api":
source = "civitai_api"
elif provider_used == "civarchive_api" and source != "civarchive":
source = "civarchive"
elif provider_used == "sqlite":
source = "archive_db"
model_data["metadata_source"] = source
model_data["last_checked_at"] = datetime.now().timestamp()
readable_source = {
"civitai_api": "CivitAI API",
"civarchive": "CivArchive API",
"openmodeldb": "OpenModelDB",
"archive_db": "Archive Database",
}.get(source, source)
logger.info(
"Fetched metadata for %s via %s",
model_data.get("model_name", ""),
readable_source,
)
local_metadata = model_data.copy()
local_metadata.pop("folder", None)
await self.update_model_metadata(
metadata_path,
local_metadata,
civitai_metadata,
metadata_provider,
)
update_payload = {
"model_name": local_metadata.get("model_name"),
"preview_url": local_metadata.get("preview_url"),
"civitai": local_metadata.get("civitai"),
}
model_data.update(update_payload)
await update_cache_func(file_path, file_path, local_metadata)
return True, None
except KeyError as exc:
error_msg = f"Error fetching metadata - Missing key: {exc} in model_data={model_data}"
logger.error(error_msg)
return False, error_msg
except RateLimitError as exc:
provider_label = exc.provider or "metadata provider"
wait_hint = (
f"; retry after approximately {int(exc.retry_after)}s"
if exc.retry_after and exc.retry_after > 0
else ""
)
error_msg = f"Rate limited by {provider_label}{wait_hint}"
logger.warning(error_msg)
return False, error_msg
except Exception as exc: # pragma: no cover - error path
error_msg = f"Error fetching metadata: {exc}"
if is_expected_offline_error(str(exc)):
logger.info(OFFLINE_FRIENDLY_MESSAGE)
return False, OFFLINE_FRIENDLY_MESSAGE
logger.error(error_msg, exc_info=True)
return False, error_msg
async def fetch_metadata_by_sha(
self, sha256: str, metadata_provider: Optional[MetadataProviderProtocol] = None
) -> tuple[Optional[Dict[str, Any]], Optional[str]]:
"""Fetch metadata for a SHA256 hash from the configured provider."""
provider = metadata_provider or await self._get_default_provider()
return await provider.get_model_by_hash(sha256)
async def relink_metadata(
self,
*,
file_path: str,
metadata: Dict[str, Any],
model_id: int,
model_version_id: Optional[int],
provider_name: Optional[str] = None,
) -> Dict[str, Any]:
"""Relink a local metadata record to a specific CivitAI model version.
When ``provider_name`` is given, the named provider is resolved via the
metadata provider selector instead of the default fallback chain. A
missing/disabled provider surfaces a user-friendly error instead of the
raw selector exception.
"""
if provider_name:
try:
provider = await self._get_provider(provider_name)
except ValueError as exc:
logger.warning(
"Unable to resolve metadata provider %s: %s", provider_name, exc
)
raise ValueError(
"CivitArchive is not available or not enabled. "
"Enable the CivitArchive API in settings to relink via CivArchive."
) from exc
else:
provider = await self._get_default_provider()
civitai_metadata = await provider.get_model_version(model_id, model_version_id)
if not civitai_metadata:
provider_label = (
"CivitArchive" if provider_name == "civarchive_api" else "CivitAI"
)
raise ValueError(
f"Model version not found on {provider_label} for ID: {model_id}"
+ (f" with version: {model_version_id}" if model_version_id else "")
)
metadata_path = get_metadata_path(file_path)
await self.update_model_metadata(
metadata_path,
metadata,
civitai_metadata,
provider,
)
return metadata
async def save_metadata_updates(
self,
*,
file_path: str,
updates: Dict[str, Any],
metadata_loader: Callable[[str], Awaitable[Dict[str, Any]]],
update_cache: Callable[[str, str, Dict[str, Any]], Awaitable[bool]],
) -> Dict[str, Any]:
"""Apply metadata updates and persist to disk and cache."""
metadata_path = get_metadata_path(file_path)
metadata = await metadata_loader(metadata_path)
for key, value in updates.items():
if key == "tags" and isinstance(value, list):
# Normalize tags: trim, lowercase, deduplicate
normalized = []
seen = set()
for tag in value:
if isinstance(tag, str):
t = tag.strip().lower()
if t and t not in seen:
normalized.append(t)
seen.add(t)
metadata[key] = normalized
elif isinstance(value, dict) and isinstance(metadata.get(key), dict):
metadata[key].update(value)
else:
metadata[key] = value
await self._metadata_manager.save_metadata(file_path, metadata)
await update_cache(file_path, file_path, metadata)
if "model_name" in updates:
logger.debug("Metadata update touched model_name; cache resort required")
return metadata
async def verify_duplicate_hashes(
self,
*,
file_paths: Iterable[str],
metadata_loader: Callable[[str], Awaitable[Dict[str, Any]]],
hash_calculator: Callable[[str], Awaitable[str]],
update_cache: Callable[[str, str, Dict[str, Any]], Awaitable[bool]],
) -> Dict[str, Any]:
"""Verify a collection of files share the same SHA256 hash."""
file_paths = list(file_paths)
if not file_paths:
raise ValueError("No file paths provided for verification")
results: Dict[str, Any] = {
"verified_as_duplicates": True,
"mismatched_files": [],
"new_hash_map": {},
}
expected_hash: Optional[str] = None
first_metadata_path = get_metadata_path(file_paths[0])
first_metadata = await metadata_loader(first_metadata_path)
if first_metadata and "sha256" in first_metadata:
expected_hash = first_metadata["sha256"].lower()
for path in file_paths:
if not os.path.exists(path):
continue
try:
actual_hash = await hash_calculator(path)
metadata_path = get_metadata_path(path)
metadata = await metadata_loader(metadata_path)
stored_hash = metadata.get("sha256", "").lower()
if not expected_hash:
expected_hash = stored_hash
if actual_hash != expected_hash:
results["verified_as_duplicates"] = False
results["mismatched_files"].append(path)
results["new_hash_map"][path] = actual_hash
if actual_hash != stored_hash:
metadata["sha256"] = actual_hash
await self._metadata_manager.save_metadata(path, metadata)
await update_cache(path, path, metadata)
except Exception as exc: # pragma: no cover - defensive path
logger.error("Error verifying hash for %s: %s", path, exc)
results["mismatched_files"].append(path)
results["new_hash_map"][path] = "error_calculating_hash"
results["verified_as_duplicates"] = False
return results