Unify remote dead-link detection

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)

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:

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

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)
 # ===========================================================================