From 510aa570cef98c2edfc35eea7cbf42dd0458ac51 Mon Sep 17 00:00:00 2001 From: Dariusz L Date: Mon, 3 Aug 2026 12:54:55 +0200 Subject: [PATCH] Unify remote dead-link detection --- core/services/civitai_search_service.py | 20 +++---------------- core/sources/civarchive.py | 21 ++------------------ core/sources/common.py | 17 ++++++++++++++++ tests/test_source_internals.py | 26 +++++++++++++++++++++++++ 4 files changed, 48 insertions(+), 36 deletions(-) diff --git a/core/services/civitai_search_service.py b/core/services/civitai_search_service.py index b11f84b..bf6c8c6 100644 --- a/core/services/civitai_search_service.py +++ b/core/services/civitai_search_service.py @@ -1,6 +1,7 @@ """CivitAI model search service.""" from ..routes.context import RouteContext +from ..sources.common import is_remote_link_marked_dead from .model_utils import CivitAISearchDependencies, ModelServiceDependencies @@ -357,24 +358,9 @@ class CivitAISearchService(ModelServiceDependencies): or result.get("from_metadata") ) - 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 collect_result_download_urls(result): urls = [] - if remote_link_is_marked_dead(result): + if is_remote_link_marked_dead(result): return urls expected_filename = result.get("filename") or filename dead_urls = set() @@ -384,7 +370,7 @@ class CivitAISearchService(ModelServiceDependencies): for mirror in mirrors: if not isinstance(mirror, dict): continue - if remote_link_is_marked_dead(mirror): + if is_remote_link_marked_dead(mirror): dead_url = str(mirror.get("url") or "").strip() if dead_url.startswith(("http://", "https://")): dead_urls.add(dead_url) diff --git a/core/sources/civarchive.py b/core/sources/civarchive.py index c82a92a..8cb544e 100644 --- a/core/sources/civarchive.py +++ b/core/sources/civarchive.py @@ -48,12 +48,13 @@ from ..type_utils import ( select_primary_model_file, to_int, ) -from .common import build_unified_search_result +from .common import build_unified_search_result, is_remote_link_marked_dead log = create_module_logger(__name__) _coerce_int = to_int +_archive_link_is_dead = is_remote_link_marked_dead CIVARCHIVE_BASE_URL = "https://civarchive.com" CIVARCHIVE_API_URL = f"{CIVARCHIVE_BASE_URL}/api" @@ -631,24 +632,6 @@ def _download_url_looks_like_model_file( return False return looks_like_model_file(normalized, expected_filename) - - -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 _fetch_remote_file_size_bytes(url: Any, timeout: int = 15) -> Optional[int]: normalized = _normalize_download_url(url) if not normalized: diff --git a/core/sources/common.py b/core/sources/common.py index c2fa335..84197e9 100644 --- a/core/sources/common.py +++ b/core/sources/common.py @@ -7,6 +7,23 @@ from typing import Any, Dict, List, Optional from ..type_utils import build_search_result, normalize_sha256 +def is_remote_link_marked_dead(item: Any) -> bool: + """Return whether remote metadata marks a link as unavailable.""" + 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 normalize_hashes_dict(hashes: Optional[Dict[str, Any]]) -> Dict[str, str]: """ Normalize a hashes dictionary (e.g. from CivitAI API or CivArchive) so that diff --git a/tests/test_source_internals.py b/tests/test_source_internals.py index 528f708..05bb7c2 100644 --- a/tests/test_source_internals.py +++ b/tests/test_source_internals.py @@ -55,6 +55,7 @@ from core.sources.civitai import ( parse_civitai_url, search_civitai_for_file, ) +from core.sources.common import is_remote_link_marked_dead from core.type_utils import prepare_remote_size_probe_url # --------------------------------------------------------------------------- @@ -416,6 +417,31 @@ class ArchiveLinkIsDeadTests(unittest.TestCase): self.assertFalse(_archive_link_is_dead([])) +class RemoteLinkIsDeadTests(unittest.TestCase): + def test_common_predicate_matches_all_supported_dead_link_markers(self): + dead_items = ( + {"deletedAt": "2024-01-01"}, + {"deleted_at": "2024-01-01"}, + {"isDead": True}, + {"is_dead": True}, + {"likelyDead": True}, + {"likely_dead": True}, + {"dead": True}, + {"status": "dead"}, + {"status": "Deleted"}, + {"status": "unavailable"}, + {"status": "missing"}, + ) + for item in dead_items: + self.assertTrue(is_remote_link_marked_dead(item)) + + def test_common_predicate_keeps_live_and_non_mapping_items_alive(self): + self.assertFalse(is_remote_link_marked_dead({"url": "https://example.com/model.safetensors"})) + self.assertFalse(is_remote_link_marked_dead("dead")) + self.assertFalse(is_remote_link_marked_dead(None)) + self.assertFalse(is_remote_link_marked_dead([])) + + # =========================================================================== # civarchive - _collect_download_urls_unified (new consolidated helper) # ===========================================================================