Share hybrid search logic between CLI and API

scene_search/cli.py

@@ -29,6 +29,7 @@ from .jellyfin_index import (
 from .maintenance import list_media_by_source, remove_media_by_source
 from .meili import Meili
 from .routing import ExplicitContentRouter, apply_context_routing
+from .search_service import SearchService
 
 
 def format_ts(ms: int) -> str:
@@ -650,88 +651,47 @@ def cmd_status(_: argparse.Namespace) -> None:
     print(f"Indexing:    {stats.get('isIndexing', 'unknown')}")
 
 def cmd_search(args: argparse.Namespace) -> None:
-    init_db(settings.db_path)
-    embedder = build_embedder()
-    vector = embedder.embed_text(args.query)
-    meili = Meili(settings.meili_url, settings.meili_master_key, settings.meili_index)
-    result = meili.search_vector(
-        vector,
-        args.query,
-        args.limit,
-        settings.hybrid_semantic_ratio,
+    service = SearchService(
+        db_path=settings.db_path,
+        meili_url=settings.meili_url,
+        meili_master_key=settings.meili_master_key,
+        meili_index=settings.meili_index,
+        model_id=settings.model_id,
+        device=settings.device,
+        semantic_ratio=settings.hybrid_semantic_ratio,
     )
-
-    hits = result.get("hits", [])
-    if not hits:
+    result = service.search(
+        args.query,
+        scene_limit=args.limit,
+        item_limit=args.limit,
+        jellyfin_only=False,
+    )
+    scenes = result["scenes"]
+    if not scenes:
         print("No hits")
         return
 
-    frame_ids = [hit.get("id") for hit in hits if hit.get("id")]
-    frame_vectors: dict[str, object] = {}
-    if frame_ids:
-        placeholders = ",".join("?" for _ in frame_ids)
-        with connect(settings.db_path) as con:
-            rows = con.execute(
-                f"""
-                SELECT f.id, f.embedding_f16, m.embedding_dim
-                FROM frames f
-                JOIN media m ON m.id = f.media_id
-                WHERE f.id IN ({placeholders})
-                """,
-                frame_ids,
-            ).fetchall()
-        frame_vectors = {
-            row["id"]: unpack_vector(row["embedding_f16"], row["embedding_dim"])
-            for row in rows
-        }
-
-    scored_vectors = [frame_vectors.get(hit.get("id")) for hit in hits]
-    if all(item is not None for item in scored_vectors):
-        import numpy as np
-
-        siglip_scores = embedder.match_probabilities(
-            vector,
-            np.stack(scored_vectors),  # type: ignore[arg-type]
-        )
-    else:
-        siglip_scores = [None] * len(hits)
-
     print(f"\nQuery: {args.query!r}\n")
-    for i, (hit, siglip_score) in enumerate(zip(hits, siglip_scores, strict=True), 1):
-        meili_score = hit.get("_rankingScore")
-        meili_text = (
-            f" meili={meili_score:.4f}"
-            if isinstance(meili_score, (float, int))
-            else ""
-        )
-        siglip_text = (
-            f" siglip={float(siglip_score):.4f}"
-            if siglip_score is not None
-            else ""
-        )
+    for i, scene in enumerate(scenes, 1):
+        meili_score = scene.get("ranking_score")
+        siglip_score = scene.get("siglip_score")
+        meili_text = f" meili={meili_score:.4f}" if meili_score is not None else ""
+        siglip_text = f" siglip={siglip_score:.4f}" if siglip_score is not None else ""
         print(
-            f"{i:2d}. {hit.get('title')} @ "
-            f"{format_ts(int(hit.get('timestamp_ms', 0)))}"
+            f"{i:2d}. {scene.get('title')} @ "
+            f"{format_ts(int(scene.get('timestamp_ms') or 0))}"
             f"{meili_text}{siglip_text}"
         )
-        caption = hit.get("caption")
-        if caption:
-            print(f"    {caption}")
-        route = hit.get("caption_route")
-        route_score = hit.get("route_score")
-        if route:
-            score_text = (
-                f" score={float(route_score):.3f}"
-                if isinstance(route_score, (float, int))
-                else ""
-            )
-            print(f"    route={route}{score_text}")
-        jellyfin_item_id = hit.get("jellyfin_item_id")
-        if jellyfin_item_id:
-            tags = hit.get("jellyfin_tags") or []
+        if scene.get("caption"):
+            print(f"    {scene['caption']}")
+        if scene.get("caption_route"):
+            route_score = scene.get("route_score")
+            score_text = f" score={float(route_score):.3f}" if route_score is not None else ""
+            print(f"    route={scene['caption_route']}{score_text}")
+        if scene.get("jellyfin_item_id"):
+            tags = scene.get("tags") or []
             tags_text = f" tags={', '.join(tags)}" if tags else ""
-            print(f"    jellyfin={jellyfin_item_id}{tags_text}")
-        print(f"    {hit.get('thumb_path')}")
+            print(f"    jellyfin={scene['jellyfin_item_id']}{tags_text}")
 
 
 def make_parser() -> argparse.ArgumentParser: