Refactor shared helpers and split server into route modules

.github/workflows/build.yml

@@ -31,7 +31,7 @@ jobs:
           .venv/bin/pip install ruff pytest
 
       - name: Lint Python
-        run: .venv/bin/ruff check searchy/backend/*.py
+        run: .venv/bin/ruff check searchy/backend/*.py searchy/backend/routes/*.py
 
       - name: Test Python
         run: .venv/bin/pytest searchy/backend/tests/ -x -q

Makefile

@@ -89,7 +89,7 @@ setup: ## Set up Python venv with dependencies
 # ─── Lint ─────────────────────────────────────────────────
 lint: ## Run linters (ruff for Python)
 	@command -v ruff >/dev/null 2>&1 || (echo "Installing ruff..." && pip install ruff)
-	ruff check searchy/backend/*.py
+	ruff check searchy/backend/*.py searchy/backend/routes/*.py
 	@echo "✓ Lint passed"
 
 # ─── Test ─────────────────────────────────────────────────

Scripts/pre-push-checks.sh

@@ -20,14 +20,14 @@ echo "═══ Pre-push checks ═══"
 
 # 1. Syntax check
 echo -n "• py_compile... "
-for f in searchy/backend/*.py; do
+for f in searchy/backend/*.py searchy/backend/routes/*.py; do
     $PYTHON -m py_compile "$f" || FAILED=1
 done
 [ $FAILED -eq 0 ] && echo "ok" || echo "FAIL"
 
 # 2. Ruff lint
 echo -n "• ruff... "
-"$VENV/bin/ruff" check searchy/backend/*.py --quiet || FAILED=1
+"$VENV/bin/ruff" check searchy/backend/*.py searchy/backend/routes/*.py --quiet || FAILED=1
 [ $FAILED -eq 0 ] && echo "ok" || echo "FAIL"
 
 # 3. Tests

searchy/backend/constants.py

@@ -44,3 +44,14 @@
 WATCHER_DEBOUNCE_DELAY = 0.5  # Seconds
 WATCHER_NOTIFY_TIMEOUT = 5    # Seconds
 WATCHER_INDEX_TIMEOUT = 30    # Seconds
+
+# ─── Face Recognition ────────────────────────────────────
+FACE_CLUSTER_THRESHOLD = 0.65        # Cosine similarity for clustering (OpenCV+ArcFace)
+FACE_REASSIGN_THRESHOLD = 0.60       # Cosine similarity for re-assignment / best-match
+FACE_CONFIDENCE_MIN = 0.9            # Minimum detection confidence
+FACE_AREA_RATIO_MAX = 0.8            # Skip if face covers more than this fraction of image
+FACE_MIN_SIZE = 40                   # Minimum face bbox dimension (px)
+FACE_IMAGE_MAX_DIM = 1920            # Resize large images before detection
+FACE_EMBEDDING_SIZE = 112            # ArcFace input size (px)
+FACE_BRIGHTNESS_MIN = 30             # Skip dark faces below this mean brightness
+FACE_BRIGHTNESS_STD_MIN = 20         # Skip low-contrast faces below this std

searchy/backend/face_recognition_service.py

@@ -15,6 +15,13 @@
 from PIL import Image
 
 from atomic_write import atomic_pickle_dump
+from constants import (
+    FACE_CLUSTER_THRESHOLD, FACE_REASSIGN_THRESHOLD, FACE_CONFIDENCE_MIN,
+    FACE_AREA_RATIO_MAX, FACE_MIN_SIZE, FACE_IMAGE_MAX_DIM,
+    FACE_EMBEDDING_SIZE, FACE_BRIGHTNESS_MIN, FACE_BRIGHTNESS_STD_MIN,
+    LARGE_BATCH_SIZE,
+)
+from utils import UnionFind
 import hashlib
 
 logger = logging.getLogger(__name__)
@@ -396,7 +403,7 @@ def verify_face(self, face_id: str, cluster_id: str, is_correct: bool) -> Dict:
             return result
 
     def _find_best_cluster_for_face(self, face: FaceData, exclude_cluster_ids: List[str] = None,
-                                     similarity_threshold: float = 0.60) -> Optional[str]:
+                                     similarity_threshold: float = FACE_REASSIGN_THRESHOLD) -> Optional[str]:
         """
         Find the best matching cluster for a face, respecting negative constraints.
         Returns cluster_id or None if no suitable cluster found.
@@ -522,7 +529,7 @@ def _detect_faces_in_image(self, image_path: str) -> List[Dict]:
                 return []
 
             # Skip very large images (resize first)
-            max_dim = 1920
+            max_dim = FACE_IMAGE_MAX_DIM
             if img.width > max_dim or img.height > max_dim:
                 ratio = min(max_dim / img.width, max_dim / img.height)
                 new_size = (int(img.width * ratio), int(img.height * ratio))
@@ -560,16 +567,16 @@ def _detect_faces_in_image(self, image_path: str) -> List[Dict]:
 
                 # Skip low confidence
                 confidence = face_data.get("confidence", 0)
-                if confidence < 0.9:
+                if confidence < FACE_CONFIDENCE_MIN:
                     continue
 
                 # Skip very small faces
-                if bbox["w"] < 40 or bbox["h"] < 40:
+                if bbox["w"] < FACE_MIN_SIZE or bbox["h"] < FACE_MIN_SIZE:
                     continue
 
                 # Skip if face covers too much of image
                 face_area_ratio = (bbox["w"] * bbox["h"]) / (img_w * img_h)
-                if face_area_ratio > 0.8:
+                if face_area_ratio > FACE_AREA_RATIO_MAX:
                     continue
 
                 # Skip dark/low-contrast faces
@@ -578,7 +585,7 @@ def _detect_faces_in_image(self, image_path: str) -> List[Dict]:
                 if face_region.size > 0:
                     mean_brightness = np.mean(face_region)
                     std_brightness = np.std(face_region)
-                    if mean_brightness < 30 or std_brightness < 20:
+                    if mean_brightness < FACE_BRIGHTNESS_MIN or std_brightness < FACE_BRIGHTNESS_STD_MIN:
                         continue
 
                 face_id = self._generate_face_id(image_path, bbox)
@@ -660,7 +667,7 @@ def _generate_embeddings_batch(self, face_detections: List[Dict], batch_size: in
                         img = img.convert('RGB')
 
                     # Resize to ArcFace input size (112x112)
-                    img = img.resize((112, 112), Image.Resampling.LANCZOS)
+                    img = img.resize((FACE_EMBEDDING_SIZE, FACE_EMBEDDING_SIZE), Image.Resampling.LANCZOS)
                     face_images.append(np.array(img))
                     valid_detections.append(detection)
 
@@ -810,11 +817,8 @@ def scan_images(self, image_paths: List[str], incremental: bool = True, limit: i
             self.scan_status = f"Error: {str(e)}"
             return {"error": str(e)}
 
-    def cluster_faces(self, similarity_threshold: float = 0.65) -> List[FaceCluster]:
-        """
-        Cluster faces using Union-Find based on embedding similarity.
-        0.65 works well for OpenCV+ArcFace combo.
-        """
+    def cluster_faces(self, similarity_threshold: float = FACE_CLUSTER_THRESHOLD) -> List[FaceCluster]:
+        """Cluster faces using Union-Find based on embedding similarity."""
         if not self.faces:
             self.clusters = []
             return []
@@ -827,24 +831,12 @@ def cluster_faces(self, similarity_threshold: float = 0.65) -> List[FaceCluster]
         norms[norms == 0] = 1  # Avoid division by zero
         embeddings_normalized = embeddings / norms
 
-        # Union-Find
-        parent = list(range(n))
-
-        def find(x):
-            if parent[x] != x:
-                parent[x] = find(parent[x])
-            return parent[x]
-
-        def union(x, y):
-            px, py = find(x), find(y)
-            if px != py:
-                parent[px] = py
+        uf = UnionFind(n)
 
         # Compare faces (O(n^2) but necessary for clustering)
         # Process in batches for large datasets
-        batch_size = 500
-        for i in range(0, n, batch_size):
-            end_i = min(i + batch_size, n)
+        for i in range(0, n, LARGE_BATCH_SIZE):
+            end_i = min(i + LARGE_BATCH_SIZE, n)
             batch = embeddings_normalized[i:end_i]
 
             # Compare against all faces from i onwards
@@ -856,12 +848,12 @@ def union(x, y):
                 for rj, sim in enumerate(row):
                     global_j = i + rj
                     if global_j > global_i and sim >= similarity_threshold:
-                        union(global_i, global_j)
+                        uf.union(global_i, global_j)
 
         # Group faces by cluster
         clusters_dict = {}
         for idx in range(n):
-            root = find(idx)
+            root = uf.find(idx)
             if root not in clusters_dict:
                 clusters_dict[root] = []
             clusters_dict[root].append(self.faces[idx])
@@ -889,7 +881,7 @@ def union(x, y):
         logger.info(f"Clustered {n} faces into {len(self.clusters)} people")
         return self.clusters
 
-    def recluster_with_constraints(self, similarity_threshold: float = 0.60) -> Dict:
+    def recluster_with_constraints(self, similarity_threshold: float = FACE_REASSIGN_THRESHOLD) -> Dict:
         """
         Smart re-clustering that:
         1. Keeps verified faces as fixed anchors in their clusters

searchy/backend/generate_embeddings.py

@@ -21,8 +21,8 @@
 
 # Import centralized model manager
 from clip_model import model_manager
-from constants import SKIP_DIRS, IMAGE_EXTENSIONS
-from utils import matches_filter, is_user_image
+from constants import SKIP_DIRS
+from utils import matches_filter, is_user_image, is_image_file
 
 # OCR support using macOS Vision framework
 try:
@@ -275,7 +275,7 @@ def process_images(image_dirs, output_dir, fast_indexing=True, max_dimension=384
                     if file.startswith('.'):
                         continue
 
-                    if os.path.splitext(file.lower())[1] in IMAGE_EXTENSIONS:
+                    if is_image_file(file):
                         # Apply filter if specified
                         if filter_type and filter_value:
                             if not matches_filter(file, filter_type, filter_value):

searchy/backend/image_watcher.py

@@ -13,15 +13,14 @@
 import argparse
 import urllib.request
 import urllib.error
-from pathlib import Path
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
 
 from constants import (
-    IMAGE_EXTENSIONS, DEFAULT_BATCH_SIZE, DEFAULT_MAX_DIMENSION,
+    DEFAULT_BATCH_SIZE, DEFAULT_MAX_DIMENSION,
     WATCHER_DEBOUNCE_DELAY, WATCHER_NOTIFY_TIMEOUT, WATCHER_INDEX_TIMEOUT,
 )
-from utils import matches_filter
+from utils import matches_filter, is_image_file
 
 # Default server URL (can be overridden via --server-url)
 DEFAULT_SERVER_URL = "http://127.0.0.1:7860"
@@ -134,18 +133,13 @@ def on_moved(self, event):
             self.pending_files.add(dest_path)
             self.last_index_time = time.time()
 
-    def _is_image(self, file_path):
-        """Check if file is an image by extension"""
+    def _is_valid_image(self, file_path):
+        """Check if file is an image AND matches the filter"""
         filename = os.path.basename(file_path)
         # Skip macOS metadata files (AppleDouble resource forks)
         if filename.startswith('._'):
             return False
-        ext = Path(file_path).suffix.lower()
-        return ext in IMAGE_EXTENSIONS
-
-    def _is_valid_image(self, file_path):
-        """Check if file is an image AND matches the filter"""
-        if not self._is_image(file_path):
+        if not is_image_file(filename):
             return False
 
         filename = os.path.basename(file_path)