78fbc14f-966c-4b10-99c2-e2d762855856

download/face-intel/config/__init__.py

@@ -0,0 +1,23 @@
+"""Re-export the singleton settings for convenient imports."""
+
+from config.settings import (
+    settings,
+    BASE_DIR,
+    DATA_DIR,
+    MODELS_DIR,
+    GALLERY_DIR,
+    UPLOADS_DIR,
+    JOBS_DIR,
+    UI_DIR,
+)
+
+__all__ = [
+    "settings",
+    "BASE_DIR",
+    "DATA_DIR",
+    "MODELS_DIR",
+    "GALLERY_DIR",
+    "UPLOADS_DIR",
+    "JOBS_DIR",
+    "UI_DIR",
+]

download/face-intel/config/settings.py

@@ -0,0 +1,159 @@
+"""
+Centralized configuration for Face Intel.
+
+All settings are environment-driven via pydantic-settings so the same
+code runs in dev, test, and production without code changes.  See
+`.env.example` for every supported variable.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+DATA_DIR = BASE_DIR / "data"
+MODELS_DIR = DATA_DIR / "models"
+GALLERY_DIR = DATA_DIR / "gallery"
+UPLOADS_DIR = DATA_DIR / "uploads"
+JOBS_DIR = DATA_DIR / "jobs"
+UI_DIR = BASE_DIR / "ui" / "static"
+
+# Ensure runtime dirs exist (idempotent)
+for _d in (DATA_DIR, MODELS_DIR, GALLERY_DIR, UPLOADS_DIR, JOBS_DIR):
+    _d.mkdir(parents=True, exist_ok=True)
+
+
+class Settings(BaseSettings):
+    """Environment-driven settings.  All fields map to FI_* env vars."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_prefix="FI_",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    # ------------------------------------------------------------------ #
+    # Core
+    # ------------------------------------------------------------------ #
+    app_name: str = "Face Intel"
+    app_version: str = "1.0.0"
+    environment: str = "development"  # development | staging | production
+    host: str = "0.0.0.0"
+    port: int = 8000
+    debug: bool = False
+
+    # ------------------------------------------------------------------ #
+    # Provider enable flags
+    # ------------------------------------------------------------------ #
+    enable_haar: bool = True
+    enable_dnn: bool = True
+    enable_mtcnn: bool = True
+    enable_retinaface: bool = False  # requires insightface
+    enable_face_recognition: bool = True
+    enable_deepface: bool = False
+    enable_insightface: bool = False
+    enable_beautifulsoup_scraper: bool = True
+    enable_selenium_scraper: bool = True
+    enable_bing_scraper: bool = False
+    enable_duckduckgo_scraper: bool = True
+    enable_google_lens: bool = True
+    enable_serpapi: bool = False
+    enable_yandex: bool = False
+    enable_tineye: bool = False
+
+    # ------------------------------------------------------------------ #
+    # Detection tuning
+    # ------------------------------------------------------------------ #
+    dnn_confidence_threshold: float = 0.7
+    mtcnn_min_face_size: int = 20
+    haar_scale_factor: float = 1.1
+    haar_min_neighbors: int = 5
+
+    # ------------------------------------------------------------------ #
+    # Recognition tuning
+    # ------------------------------------------------------------------ #
+    face_recognition_tolerance: float = 0.6
+    face_recognition_model: str = "hog"  # hog | cnn
+    deepface_backend: str = "arcface"
+    insightface_model_pack: str = "buffalo_l"
+    recognition_match_threshold: float = 0.5  # cosine similarity cutoff
+
+    # ------------------------------------------------------------------ #
+    # Scraping
+    # ------------------------------------------------------------------ #
+    scrape_timeout: int = 30
+    scrape_max_images: int = 50
+    selenium_headless: bool = True
+    selenium_implicit_wait: int = 10
+    selenium_scroll_iterations: int = 5
+    user_agent: str = (
+        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"
+    )
+
+    # ------------------------------------------------------------------ #
+    # Reverse image search
+    # ------------------------------------------------------------------ #
+    reverse_search_max_results: int = 20
+    serpapi_key: str = ""
+    bing_api_key: str = ""
+    tineye_public_key: str = ""
+    tineye_private_key: str = ""
+
+    # ------------------------------------------------------------------ #
+    # Orchestrator
+    # ------------------------------------------------------------------ #
+    orchestrator_timeout_seconds: float = 90.0
+    orchestrator_max_concurrency: int = 8
+    retry_max_attempts: int = 3
+    retry_initial_backoff_seconds: float = 0.5
+    retry_max_backoff_seconds: float = 8.0
+    retry_retriable_exceptions: List[str] = Field(
+        default_factory=lambda: ["TimeoutError", "ConnectionError"]
+    )
+
+    # ------------------------------------------------------------------ #
+    # Cache
+    # ------------------------------------------------------------------ #
+    cache_enabled: bool = True
+    cache_ttl_seconds: int = 3600
+    cache_max_entries: int = 1000
+
+    # ------------------------------------------------------------------ #
+    # Health
+    # ------------------------------------------------------------------ #
+    health_check_interval_seconds: int = 60
+    circuit_breaker_failure_threshold: int = 5
+    circuit_breaker_recovery_seconds: int = 120
+
+    # ------------------------------------------------------------------ #
+    # Storage
+    # ------------------------------------------------------------------ #
+    db_path: str = str(DATA_DIR / "face_intel.db")
+    audit_log_path: str = str(DATA_DIR / "audit.log")
+    job_retention_days: int = 7
+
+    # ------------------------------------------------------------------ #
+    # API
+    # ------------------------------------------------------------------ #
+    rate_limit_per_minute: int = 30
+    cors_origins: List[str] = Field(
+        default_factory=lambda: ["*"]
+    )
+    require_consent_header: bool = True
+    consent_header_name: str = "X-Consent-Statement"
+
+    # ------------------------------------------------------------------ #
+    # Logging
+    # ------------------------------------------------------------------ #
+    log_level: str = "INFO"
+    log_json: bool = False
+
+
+settings = Settings()

download/face-intel/providers/__init__.py

@@ -0,0 +1,132 @@
+"""
+Provider package — auto-registers every enabled provider.
+
+Importing this module triggers construction + registration of every
+provider whose `enable_*` flag is true in settings.  Optional providers
+(importable only if their optional dep is installed) are wrapped in
+try/except so a missing dep degrades gracefully to NOT_CONFIGURED.
+"""
+
+from __future__ import annotations
+
+from loguru import logger
+
+from config import settings
+from providers.base import (
+    Provider,
+    ProviderCapability,
+    ProviderStatus,
+    register_provider,
+    list_providers,
+    get_provider,
+    clear_registry,
+)
+
+
+def _safe_register(module_path: str, class_name: str, enable_flag: bool) -> None:
+    """Import + register a provider; swallow ImportError if optional dep missing."""
+    if not enable_flag:
+        return
+    try:
+        import importlib
+        module = importlib.import_module(module_path)
+        cls = getattr(module, class_name)
+        instance = cls()
+        register_provider(instance)
+        logger.info(f"Registered provider: {instance.name} ({instance.capability.value})")
+    except ImportError as e:
+        logger.warning(
+            f"Provider {class_name} from {module_path} skipped (missing optional dep): {e}"
+        )
+    except Exception as e:
+        logger.error(f"Failed to register {class_name}: {e}")
+
+
+def initialize_providers() -> None:
+    """Construct and register every enabled provider.  Call once at startup."""
+    # Detection
+    _safe_register("providers.detection.haar", "HaarDetector", settings.enable_haar)
+    _safe_register("providers.detection.dnn", "DNNDetector", settings.enable_dnn)
+    _safe_register("providers.detection.mtcnn", "MTCNNDetector", settings.enable_mtcnn)
+    _safe_register("providers.detection.retinaface", "RetinaFaceDetector", settings.enable_retinaface)
+
+    # Recognition
+    _safe_register(
+        "providers.recognition.face_recognition_provider", "FaceRecognitionProvider",
+        settings.enable_face_recognition,
+    )
+    _safe_register(
+        "providers.recognition.deepface_provider", "DeepFaceProvider",
+        settings.enable_deepface,
+    )
+    _safe_register(
+        "providers.recognition.insightface_provider", "InsightFaceProvider",
+        settings.enable_insightface,
+    )
+
+    # Scraping
+    _safe_register(
+        "providers.scraper.beautifulsoup_scraper", "BeautifulSoupScraper",
+        settings.enable_beautifulsoup_scraper,
+    )
+    _safe_register(
+        "providers.scraper.selenium_scraper", "SeleniumScraper",
+        settings.enable_selenium_scraper,
+    )
+    _safe_register(
+        "providers.scraper.bing_scraper", "BingImageScraper",
+        settings.enable_bing_scraper,
+    )
+    _safe_register(
+        "providers.scraper.duckduckgo_scraper", "DuckDuckGoScraper",
+        settings.enable_duckduckgo_scraper,
+    )
+
+    # Reverse image search
+    _safe_register(
+        "providers.reverse.google_lens", "GoogleLensProvider",
+        settings.enable_google_lens,
+    )
+    _safe_register(
+        "providers.reverse.serpapi", "SerpAPIProvider",
+        settings.enable_serpapi,
+    )
+    _safe_register(
+        "providers.reverse.yandex", "YandexReverseProvider",
+        settings.enable_yandex,
+    )
+    _safe_register(
+        "providers.reverse.tineye", "TinEyeProvider",
+        settings.enable_tineye,
+    )
+
+
+def provider_statuses() -> list[dict]:
+    """Snapshot of every provider's availability + status."""
+    out: list[dict] = []
+    for p in list_providers():
+        try:
+            available = p.is_available()
+        except Exception:
+            available = False
+        status = ProviderStatus.HEALTHY if available else ProviderStatus.NOT_CONFIGURED
+        out.append({
+            "name": p.name,
+            "capability": p.capability.value,
+            "status": status.value,
+            "available": available,
+        })
+    return out
+
+
+__all__ = [
+    "Provider",
+    "ProviderCapability",
+    "ProviderStatus",
+    "initialize_providers",
+    "list_providers",
+    "get_provider",
+    "clear_registry",
+    "register_provider",
+    "provider_statuses",
+]

download/face-intel/providers/base.py

@@ -0,0 +1,195 @@
+"""
+Provider Framework — the contract every provider must satisfy.
+
+Key design decisions (matching Phone Intel methodology):
+
+1. **Provider Protocol, not ABC** — structural typing, no inheritance
+   ceremony.  A provider is any object with the right methods.
+
+2. **Completely isolated** — a provider only imports its own deps,
+   never imports from siblings.  A failing optional dep (e.g.
+   `insightface` missing) must not affect other providers.
+
+3. **ProviderResult** carries the raw response (preserved verbatim),
+   the normalized fields, the latency, and any error.  Nothing is
+   discarded.
+
+4. **ProviderHealth** is tracked by the orchestrator's circuit
+   breaker — a provider that fails N times in a row is shorted
+   until recovery.
+
+5. **Capability flags** declare what a provider can do so the
+   orchestrator can route queries correctly (e.g. detection vs.
+   recognition vs. scraping vs. reverse-search).
+"""
+
+from __future__ import annotations
+
+import enum
+import time
+from dataclasses import dataclass, field
+from typing import Any, Optional, Protocol, runtime_checkable
+
+
+# --------------------------------------------------------------------------- #
+# Capability & status enums
+# --------------------------------------------------------------------------- #
+class ProviderCapability(str, enum.Enum):
+    DETECTION = "detection"
+    RECOGNITION = "recognition"
+    SCRAPING = "scraping"
+    REVERSE_SEARCH = "reverse_search"
+
+
+class ProviderStatus(str, enum.Enum):
+    HEALTHY = "healthy"
+    DEGRADED = "degraded"        # working but slow / intermittent
+    UNHEALTHY = "unhealthy"      # circuit breaker open
+    DISABLED = "disabled"        # disabled by config
+    NOT_CONFIGURED = "not_configured"  # missing API key / optional dep
+
+
+# --------------------------------------------------------------------------- #
+# Result objects
+# --------------------------------------------------------------------------- #
+@dataclass
+class ProviderResult:
+    """
+    The contract returned by every provider.
+
+    `raw` is preserved verbatim for evidence/audit.
+    `normalized` is provider-specific cleaned data (boxes, encodings,
+    image URLs, etc.) — see normalization/schema.py for the post-merge
+    unified schema.
+    """
+    provider: str
+    capability: ProviderCapability
+    success: bool
+    elapsed_ms: float
+    raw: Any = None                    # untouched provider response
+    normalized: dict = field(default_factory=dict)
+    error: Optional[str] = None
+    error_type: Optional[str] = None
+    metadata: dict = field(default_factory=dict)  # rate-limit info, warnings, etc.
+
+    def to_dict(self) -> dict:
+        return {
+            "provider": self.provider,
+            "capability": self.capability.value,
+            "success": self.success,
+            "elapsed_ms": self.elapsed_ms,
+            "raw": self.raw,
+            "normalized": self.normalized,
+            "error": self.error,
+            "error_type": self.error_type,
+            "metadata": self.metadata,
+        }
+
+
+# --------------------------------------------------------------------------- #
+# Provider Protocol
+# --------------------------------------------------------------------------- #
+@runtime_checkable
+class Provider(Protocol):
+    """
+    Structural contract for every provider.
+
+    A provider class must define:
+      - name: str                     — unique slug
+      - capability: ProviderCapability
+      - is_available() -> bool        — quick config/dep check
+      - execute(input_data) -> ProviderResult
+
+    Providers may be sync or async.  The orchestrator wraps sync calls
+    in `asyncio.to_thread`.
+    """
+
+    name: str
+    capability: ProviderCapability
+
+    def is_available(self) -> bool: ...
+
+    def execute(self, input_data: Any) -> ProviderResult: ...
+
+
+# --------------------------------------------------------------------------- #
+# Base helper class (optional convenience — providers may subclass
+# this OR implement the Protocol directly)
+# --------------------------------------------------------------------------- #
+class BaseProvider:
+    """Convenience base with timing + error handling boilerplate."""
+
+    name: str = "base"
+    capability: ProviderCapability = ProviderCapability.DETECTION
+
+    def is_available(self) -> bool:
+        return True
+
+    def execute(self, input_data: Any) -> ProviderResult:  # pragma: no cover
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------ #
+    # Helpers used by subclasses
+    # ------------------------------------------------------------------ #
+    def _run(self, input_data: Any) -> tuple[Any, dict]:
+        """Subclass implements: returns (raw_response, normalized_dict)."""
+        raise NotImplementedError
+
+    def _safe_execute(self, input_data: Any) -> ProviderResult:
+        """Wrap _run with timing + exception capture."""
+        t0 = time.perf_counter()
+        try:
+            if not self.is_available():
+                return ProviderResult(
+                    provider=self.name,
+                    capability=self.capability,
+                    success=False,
+                    elapsed_ms=0.0,
+                    error="Provider not available (disabled or missing dependencies)",
+                    error_type="NotConfigured",
+                )
+            raw, normalized = self._run(input_data)
+            elapsed = (time.perf_counter() - t0) * 1000.0
+            return ProviderResult(
+                provider=self.name,
+                capability=self.capability,
+                success=True,
+                elapsed_ms=round(elapsed, 3),
+                raw=raw,
+                normalized=normalized,
+            )
+        except Exception as e:
+            elapsed = (time.perf_counter() - t0) * 1000.0
+            return ProviderResult(
+                provider=self.name,
+                capability=self.capability,
+                success=False,
+                elapsed_ms=round(elapsed, 3),
+                error=str(e),
+                error_type=type(e).__name__,
+            )
+
+
+# --------------------------------------------------------------------------- #
+# Provider registry
+# --------------------------------------------------------------------------- #
+_REGISTRY: dict[str, Provider] = {}
+
+
+def register_provider(provider: Provider) -> Provider:
+    """Register a provider instance.  Idempotent."""
+    _REGISTRY[provider.name] = provider
+    return provider
+
+
+def get_provider(name: str) -> Optional[Provider]:
+    return _REGISTRY.get(name)
+
+
+def list_providers() -> list[Provider]:
+    return list(_REGISTRY.values())
+
+
+def clear_registry() -> None:
+    """Used by tests."""
+    _REGISTRY.clear()

download/face-intel/providers/detection/haar.py

@@ -0,0 +1,57 @@
+"""
+OpenCV Haar Cascade face detector.
+
+Verified facts (Phase 2):
+  - Cascade file ships with OpenCV at `cv2.data.haarcascades`.
+  - Returns rectangles only (no confidence, no landmarks).
+  - Fast (~5–15 ms CPU), poor on profile faces.
+"""
+
+from __future__ import annotations
+
+import cv2
+import numpy as np
+
+from config import settings
+from providers.base import BaseProvider, ProviderCapability, ProviderResult
+from utils.image import BBox
+
+
+class HaarDetector(BaseProvider):
+    name = "haar"
+    capability = ProviderCapability.DETECTION
+
+    def __init__(self) -> None:
+        cascade_path = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
+        self._cascade = cv2.CascadeClassifier(cascade_path)
+        if self._cascade.empty():
+            raise RuntimeError("Failed to load Haar cascade classifier.")
+
+    def is_available(self) -> bool:
+        return not self._cascade.empty()
+
+    def _run(self, img: np.ndarray) -> tuple[dict, dict]:
+        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+        gray = cv2.equalizeHist(gray)
+        rects = self._cascade.detectMultiScale(
+            gray,
+            scaleFactor=settings.haar_scale_factor,
+            minNeighbors=settings.haar_min_neighbors,
+            minSize=(30, 30),
+            flags=cv2.CASCADE_SCALE_IMAGE,
+        )
+        boxes = [
+            BBox(int(x), int(y), int(w), int(h)).to_dict()
+            for (x, y, w, h) in rects
+        ]
+        raw = {
+            "rectangles": [[int(x), int(y), int(w), int(h)] for (x, y, w, h) in rects],
+            "num_faces": len(rects),
+        }
+        normalized = {
+            "boxes": boxes,
+            "num_faces": len(boxes),
+            "landmarks": None,  # Haar gives no landmarks
+            "confidences": [1.0] * len(boxes),  # Haar gives no score
+        }
+        return raw, normalized

download/face-intel/utils/audit.py

@@ -0,0 +1,41 @@
+"""
+Append-only audit log for sensitive operations (identity resolution,
+reverse image search, gallery mutations).
+
+Format: one JSON object per line (JSONL).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from config import settings
+from loguru import logger
+
+
+def audit_log(
+    action: str,
+    actor: str = "anonymous",
+    target: str | None = None,
+    outcome: str = "success",
+    details: dict[str, Any] | None = None,
+) -> None:
+    """Append a structured audit entry."""
+    entry = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "action": action,
+        "actor": actor,
+        "target": target,
+        "outcome": outcome,
+        "details": details or {},
+    }
+    path = Path(settings.audit_log_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+    except OSError as e:
+        logger.warning(f"Audit log write failed: {e}")

download/face-intel/utils/http.py

@@ -0,0 +1,47 @@
+"""
+Shared HTTP session and helpers for HTTP-based providers.
+"""
+
+from __future__ import annotations
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+from config import settings
+
+
+def make_session(
+    pool_connections: int = 10,
+    pool_maxsize: int = 10,
+    retries: int = 2,
+    backoff: float = 0.3,
+) -> requests.Session:
+    """Build a requests.Session with sensible connection pooling and retry."""
+    session = requests.Session()
+    session.headers.update({"User-Agent": settings.user_agent})
+    retry = Retry(
+        total=retries,
+        backoff_factor=backoff,
+        status_forcelist=[502, 503, 504],
+        allowed_methods=["GET", "POST", "HEAD"],
+    )
+    adapter = HTTPAdapter(
+        pool_connections=pool_connections,
+        pool_maxsize=pool_maxsize,
+        max_retries=retry,
+    )
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    return session
+
+
+# Module-level shared session (lazy-initialized)
+_session: requests.Session | None = None
+
+
+def shared_session() -> requests.Session:
+    global _session
+    if _session is None:
+        _session = make_session()
+    return _session

download/face-intel/utils/image.py

@@ -0,0 +1,168 @@
+"""
+Image utility helpers used across every provider.
+
+All providers receive and return BGR numpy arrays (OpenCV convention).
+These helpers handle the boundary conversions: bytes <-> numpy,
+base64 <-> numpy, URL -> numpy, image hashing, cropping, resizing,
+and side-by-side annotation.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Tuple
+
+import cv2
+import numpy as np
+from PIL import Image
+
+
+# --------------------------------------------------------------------------- #
+# Decode / encode
+# --------------------------------------------------------------------------- #
+def bytes_to_numpy(image_bytes: bytes) -> np.ndarray:
+    """Decode raw image bytes into an OpenCV BGR numpy array."""
+    nparr = np.frombuffer(image_bytes, np.uint8)
+    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+    if img is None:
+        raise ValueError("Could not decode image bytes. Unsupported format or corrupted data.")
+    return img
+
+
+def base64_to_numpy(b64_string: str) -> np.ndarray:
+    """Decode a base64-encoded image string into a BGR numpy array."""
+    if "," in b64_string:
+        b64_string = b64_string.split(",", 1)[1]
+    raw = base64.b64decode(b64_string)
+    return bytes_to_numpy(raw)
+
+
+def numpy_to_base64(img: np.ndarray, fmt: str = ".jpg", quality: int = 85) -> str:
+    """Encode a BGR numpy array as a base64 string."""
+    params = []
+    if fmt.lower() in (".jpg", ".jpeg"):
+        params = [cv2.IMWRITE_JPEG_QUALITY, quality]
+    ok, buffer = cv2.imencode(fmt, img, params)
+    if not ok:
+        raise ValueError("Could not encode image.")
+    return base64.b64encode(buffer).decode("utf-8")
+
+
+def save_image(img: np.ndarray, path: Path, fmt: str = ".jpg") -> Path:
+    """Save an image to disk. Returns the path."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    cv2.imwrite(str(path), img)
+    return path
+
+
+# --------------------------------------------------------------------------- #
+# Hashing (used as cache key)
+# --------------------------------------------------------------------------- #
+def image_hash(img: np.ndarray) -> str:
+    """SHA-256 hash of the JPEG-encoded image — stable cache key."""
+    ok, buffer = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 90])
+    if not ok:
+        raise ValueError("Could not encode image for hashing.")
+    return hashlib.sha256(buffer.tobytes()).hexdigest()
+
+
+def bytes_hash(data: bytes) -> str:
+    """SHA-256 hash of raw bytes."""
+    return hashlib.sha256(data).hexdigest()
+
+
+# --------------------------------------------------------------------------- #
+# Geometry
+# --------------------------------------------------------------------------- #
+@dataclass
+class BBox:
+    """Axis-aligned bounding box."""
+    x: int
+    y: int
+    w: int
+    h: int
+
+    def to_dict(self) -> dict:
+        return {"x": self.x, "y": self.y, "w": self.w, "h": self.h}
+
+    @property
+    def area(self) -> int:
+        return self.w * self.h
+
+    def to_face_recognition_tuple(self) -> Tuple[int, int, int, int]:
+        """Convert to (top, right, bottom, left) tuple used by face_recognition."""
+        return (self.y, self.x + self.w, self.y + self.h, self.x)
+
+
+def crop_face(img: np.ndarray, bbox: BBox, margin: float = 0.2) -> np.ndarray:
+    """Crop a face region with optional fractional margin."""
+    x0 = max(0, bbox.x - int(bbox.w * margin))
+    y0 = max(0, bbox.y - int(bbox.h * margin))
+    x1 = min(img.shape[1], bbox.x + bbox.w + int(bbox.w * margin))
+    y1 = min(img.shape[0], bbox.y + bbox.h + int(bbox.h * margin))
+    return img[y0:y1, x0:x1]
+
+
+def resize_with_aspect(img: np.ndarray, max_dim: int = 1024) -> np.ndarray:
+    """Resize so the longest side is at most max_dim, preserving aspect."""
+    h, w = img.shape[:2]
+    if max(h, w) <= max_dim:
+        return img
+    scale = max_dim / max(h, w)
+    return cv2.resize(img, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)
+
+
+# --------------------------------------------------------------------------- #
+# Network I/O
+# --------------------------------------------------------------------------- #
+def url_to_numpy(url: str, timeout: int = 15) -> np.ndarray:
+    """Download an image from a URL and return it as a BGR numpy array."""
+    import requests
+    headers = {"User-Agent": "Mozilla/5.0"}
+    resp = requests.get(url, headers=headers, timeout=timeout, stream=True)
+    resp.raise_for_status()
+    return bytes_to_numpy(resp.content)
+
+
+def url_to_bytes(url: str, timeout: int = 15) -> bytes:
+    """Download a URL and return raw bytes."""
+    import requests
+    headers = {"User-Agent": "Mozilla/5.0"}
+    resp = requests.get(url, headers=headers, timeout=timeout, stream=True)
+    resp.raise_for_status()
+    return resp.content
+
+
+def validate_image_file(path: Path) -> bool:
+    """Validate that a file is a readable image (PIL verify)."""
+    try:
+        with Image.open(path) as im:
+            im.verify()
+        return True
+    except Exception:
+        return False
+
+
+# --------------------------------------------------------------------------- #
+# Drawing helpers (for UI annotations)
+# --------------------------------------------------------------------------- #
+def draw_boxes(
+    img: np.ndarray,
+    boxes: list,
+    color: Tuple[int, int, int] = (0, 255, 0),
+    thickness: int = 2,
+) -> np.ndarray:
+    """Draw a list of BBox (or dict) onto a copy of the image."""
+    out = img.copy()
+    for b in boxes:
+        if isinstance(b, dict):
+            b = BBox(b["x"], b["y"], b["w"], b["h"])
+        elif not isinstance(b, BBox):
+            b = BBox(*b)
+        cv2.rectangle(out, (b.x, b.y), (b.x + b.w, b.y + b.h), color, thickness)
+    return out

download/face-intel/utils/timing.py

@@ -0,0 +1,20 @@
+"""
+Timing utility — context manager and decorator for measuring latency.
+"""
+
+from __future__ import annotations
+
+import time
+from contextlib import contextmanager
+from typing import Iterator
+
+
+@contextmanager
+def timed(label: str = "") -> Iterator[dict]:
+    """Context manager that yields a dict and fills it with elapsed ms."""
+    info: dict = {"label": label, "elapsed_ms": 0.0}
+    t0 = time.perf_counter()
+    try:
+        yield info
+    finally:
+        info["elapsed_ms"] = round((time.perf_counter() - t0) * 1000.0, 3)

download/face-intelligence-api/config.py

@@ -1,87 +0,0 @@
-"""
-Configuration module for the Face Intelligence API.
-Centralizes all settings and provides ethical-use enforcement flags.
-"""
-
-from pathlib import Path
-from pydantic_settings import BaseSettings
-from pydantic import Field
-
-
-BASE_DIR = Path(__file__).resolve().parent
-DATA_DIR = BASE_DIR / "data"
-KNOWN_FACES_DIR = DATA_DIR / "known_faces"
-TEMP_DIR = DATA_DIR / "temp"
-UPLOADS_DIR = DATA_DIR / "uploads"
-MODELS_DIR = DATA_DIR / "models"
-
-for _d in (KNOWN_FACES_DIR, TEMP_DIR, UPLOADS_DIR, MODELS_DIR):
-    _d.mkdir(parents=True, exist_ok=True)
-
-
-class Settings(BaseSettings):
-    """Runtime settings, overridable via environment variables or .env file."""
-
-    # --- API ---
-    app_name: str = "Face Intelligence API"
-    app_version: str = "1.0.0"
-    host: str = "0.0.0.0"
-    port: int = 8000
-    debug: bool = False
-
-    # --- Face Detection ---
-    # Options: "haar", "dnn", "mtcnn"
-    default_detector: str = "dnn"
-    # DNN model files (downloaded automatically on first run)
-    dnn_prototxt_url: str = (
-        "https://raw.githubusercontent.com/opencv/opencv_3rdparty"
-        "/dnn_samples_face_detector_20170830/deploy.prototxt"
-    )
-    dnn_model_url: str = (
-        "https://raw.githubusercontent.com/opencv/opencv_3rdparty"
-        "/dnn_samples_face_detector_20170830/res10_300x300_ssd_iter_140000.caffemodel"
-    )
-    dnn_confidence_threshold: float = 0.7
-
-    # Haar cascade file (ships with OpenCV)
-    haar_cascade_path: str = str(
-        Path(cv2.__file__).parent / "data" / "haarcascade_frontalface_default.xml"
-    ) if (cv2 := __import__("cv2", globals(), locals(), [], 0)) else "haarcascade_frontalface_default.xml"
-
-    # --- Face Recognition ---
-    face_recognition_tolerance: float = 0.6  # lower = stricter
-    face_recognition_model: str = "hog"  # "hog" (fast) or "cnn" (accurate, GPU)
-
-    # --- Web Scraping ---
-    scrape_timeout: int = 30
-    scrape_max_images: int = 50
-    selenium_headless: bool = True
-    selenium_implicit_wait: int = 10
-    user_agent: str = (
-        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
-        "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"
-    )
-
-    # --- Reverse Image Search ---
-    # Method: "google_lens" (default, free, via Selenium) or "serpapi" (paid)
-    reverse_search_method: str = "google_lens"
-    serpapi_key: str = ""
-    reverse_search_max_results: int = 20
-
-    # --- Ethical Guardrails (DO NOT disable in production) ---
-    # Require an explicit authorization token for identity-resolution endpoints
-    require_authorization: bool = True
-    authorization_token: str = "CHANGE_ME_IN_PRODUCTION"
-    # Require consent acknowledgment header for online lookup endpoints
-    require_consent_header: bool = True
-    # Rate limiting (requests per minute per IP)
-    rate_limit_per_minute: int = 20
-    # Log all identity-resolution operations for audit
-    audit_log_path: str = str(DATA_DIR / "audit.log")
-
-    class Config:
-        env_file = ".env"
-        env_prefix = "FI_"
-
-
-settings = Settings()

download/face-intelligence-api/modules/__init__.py

@@ -1 +0,0 @@
-"""Modules package."""

download/face-intelligence-api/modules/face_detection.py

@@ -1,261 +0,0 @@
-"""
-Face Detection Module
-=====================
-Provides three interchangeable face-detection backends:
-
-  1. Haar Cascade  - fastest, least accurate (ships with OpenCV)
-  2. DNN (Caffe)   - good accuracy/speed tradeoff (default)
-  3. MTCNN         - high accuracy, slower, detects landmarks
-
-All detectors return a normalized list of FaceBox dataclasses so the
-rest of the pipeline is detector-agnostic.
-"""
-
-from __future__ import annotations
-
-import urllib.request
-from dataclasses import dataclass, asdict
-from pathlib import Path
-from typing import List, Optional
-
-import cv2
-import numpy as np
-
-from config import settings, MODELS_DIR
-from utils.image_utils import crop_face
-
-
-# --------------------------------------------------------------------------- #
-# Data model
-# --------------------------------------------------------------------------- #
-@dataclass
-class FaceBox:
-    """A single detected face."""
-    x: int
-    y: int
-    w: int
-    h: int
-    confidence: float
-    detector: str
-    landmarks: Optional[dict] = None  # eyes, nose, mouth corners (MTCNN only)
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-    @property
-    def area(self) -> int:
-        return self.w * self.h
-
-
-# --------------------------------------------------------------------------- #
-# Base detector
-# --------------------------------------------------------------------------- #
-class BaseDetector:
-    name: str = "base"
-
-    def detect(self, img: np.ndarray) -> List[FaceBox]:
-        raise NotImplementedError
-
-
-# --------------------------------------------------------------------------- #
-# 1. Haar Cascade Detector
-# --------------------------------------------------------------------------- #
-class HaarCascadeDetector(BaseDetector):
-    """Classic Viola-Jones cascade. Very fast, struggles with angles/lighting."""
-
-    name = "haar"
-
-    def __init__(self):
-        cascade_path = settings.haar_cascade_path
-        if not Path(cascade_path).exists():
-            # Fallback: try OpenCV's bundled path
-            cascade_path = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
-        self.cascade = cv2.CascadeClassifier(cascade_path)
-        if self.cascade.empty():
-            raise RuntimeError("Failed to load Haar cascade classifier.")
-
-    def detect(self, img: np.ndarray) -> List[FaceBox]:
-        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
-        gray = cv2.equalizeHist(gray)
-        rects = self.cascade.detectMultiScale(
-            gray,
-            scaleFactor=1.1,
-            minNeighbors=5,
-            minSize=(30, 30),
-            flags=cv2.CASCADE_SCALE_IMAGE,
-        )
-        boxes: List[FaceBox] = []
-        for (x, y, w, h) in rects:
-            boxes.append(FaceBox(int(x), int(y), int(w), int(h), 1.0, self.name))
-        return boxes
-
-
-# --------------------------------------------------------------------------- #
-# 2. DNN (Caffe SSD) Detector
-# --------------------------------------------------------------------------- #
-class DNNDetector(BaseDetector):
-    """OpenCV's DNN module with the SSD Caffe face model. Recommended default."""
-
-    name = "dnn"
-
-    PROTOTXT = MODELS_DIR / "deploy.prototxt"
-    CAFFEMODEL = MODELS_DIR / "res10_300x300_ssd_iter_140000.caffemodel"
-
-    def __init__(self):
-        self._download_models()
-        self.net = cv2.dnn.readNetFromCaffe(str(self.PROTOTXT), str(self.CAFFEMODEL))
-        # Use GPU if available
-        try:
-            self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)
-            self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)
-        except Exception:
-            pass  # CPU fallback
-
-    def _download_models(self):
-        if not self.PROTOTXT.exists():
-            print("[DNN] Downloading prototxt...")
-            urllib.request.urlretrieve(settings.dnn_prototxt_url, self.PROTOTXT)
-        if not self.CAFFEMODEL.exists():
-            print("[DNN] Downloading caffemodel (this may take a moment)...")
-            urllib.request.urlretrieve(settings.dnn_model_url, self.CAFFEMODEL)
-
-    def detect(self, img: np.ndarray) -> List[FaceBox]:
-        h, w = img.shape[:2]
-        blob = cv2.dnn.blobFromImage(
-            cv2.resize(img, (300, 300)),
-            1.0,
-            (300, 300),
-            (104.0, 177.0, 123.0),
-        )
-        self.net.setInput(blob)
-        detections = self.net.forward()
-
-        boxes: List[FaceBox] = []
-        for i in range(detections.shape[2]):
-            confidence = detections[0, 0, i, 2]
-            if confidence < settings.dnn_confidence_threshold:
-                continue
-            x1 = int(detections[0, 0, i, 3] * w)
-            y1 = int(detections[0, 0, i, 4] * h)
-            x2 = int(detections[0, 0, i, 5] * w)
-            y2 = int(detections[0, 0, i, 6] * h)
-            boxes.append(
-                FaceBox(
-                    x=x1, y=y1, w=x2 - x1, h=y2 - y1,
-                    confidence=float(confidence), detector=self.name,
-                )
-            )
-        return boxes
-
-
-# --------------------------------------------------------------------------- #
-# 3. MTCNN Detector
-# --------------------------------------------------------------------------- #
-class MTCNNDetector(BaseDetector):
-    """MTCNN: high accuracy, also returns 5 facial landmarks."""
-
-    name = "mtcnn"
-
-    def __init__(self):
-        try:
-            from mtcnn.mtcnn import MTCNN
-        except ImportError as e:
-            raise ImportError(
-                "MTCNN requires the 'mtcnn' and 'tensorflow' packages. "
-                "Install with: pip install mtcnn tensorflow"
-            ) from e
-        self.detector = MTCNN()
-
-    def detect(self, img: np.ndarray) -> List[FaceBox]:
-        rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
-        results = self.detector.detect_faces(rgb)
-        boxes: List[FaceBox] = []
-        for r in results:
-            x, y, w, h = r["box"]
-            boxes.append(
-                FaceBox(
-                    x=int(x), y=int(y), w=int(w), h=int(h),
-                    confidence=float(r["confidence"]), detector=self.name,
-                    landmarks={
-                        "left_eye": r.get("keypoints", {}).get("left_eye"),
-                        "right_eye": r.get("keypoints", {}).get("right_eye"),
-                        "nose": r.get("keypoints", {}).get("nose"),
-                        "mouth_left": r.get("keypoints", {}).get("mouth_left"),
-                        "mouth_right": r.get("keypoints", {}).get("mouth_right"),
-                    },
-                )
-            )
-        return boxes
-
-
-# --------------------------------------------------------------------------- #
-# Factory + high-level API
-# --------------------------------------------------------------------------- #
-_DETECTORS: dict = {}
-
-def get_detector(name: Optional[str] = None) -> BaseDetector:
-    """Get a cached detector instance by name."""
-    name = name or settings.default_detector
-    if name not in _DETECTORS:
-        if name == "haar":
-            _DETECTORS[name] = HaarCascadeDetector()
-        elif name == "dnn":
-            _DETECTORS[name] = DNNDetector()
-        elif name == "mtcnn":
-            _DETECTORS[name] = MTCNNDetector()
-        else:
-            raise ValueError(f"Unknown detector: {name}. Use 'haar', 'dnn', or 'mtcnn'.")
-    return _DETECTORS[name]
-
-
-def detect_faces(
-    img: np.ndarray,
-    detector_name: Optional[str] = None,
-    draw: bool = False,
-) -> dict:
-    """Detect faces and optionally return an annotated image.
-
-    Returns:
-        {
-            "detector": str,
-            "num_faces": int,
-            "faces": List[dict],
-            "annotated_image_base64": str | None,
-        }
-    """
-    detector = get_detector(detector_name)
-    boxes = detector.detect(img)
-
-    annotated_b64 = None
-    if draw:
-        annotated = img.copy()
-        for box in boxes:
-            cv2.rectangle(
-                annotated, (box.x, box.y), (box.x + box.w, box.y + box.h),
-                (0, 255, 0), 2,
-            )
-            label = f"{box.detector} {box.confidence:.2f}"
-            cv2.putText(
-                annotated, label, (box.x, max(box.y - 8, 12)),
-                cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1,
-            )
-            if box.landmarks:
-                for pt in box.landmarks.values():
-                    if pt:
-                        cv2.circle(annotated, pt, 3, (0, 0, 255), -1)
-        from utils.image_utils import numpy_to_base64
-        annotated_b64 = numpy_to_base64(annotated)
-
-    return {
-        "detector": detector.name,
-        "num_faces": len(boxes),
-        "faces": [b.to_dict() for b in boxes],
-        "annotated_image_base64": annotated_b64,
-    }
-
-
-def extract_face_crops(img: np.ndarray, detector_name: Optional[str] = None) -> List[np.ndarray]:
-    """Return a list of cropped face images (with 20% margin)."""
-    detector = get_detector(detector_name)
-    boxes = detector.detect(img)
-    return [crop_face(img, (b.x, b.y, b.w, b.h)) for b in boxes]

download/face-intelligence-api/modules/face_recognition.py

@@ -1,255 +0,0 @@
-"""
-Face Recognition Module
-=======================
-Wraps the `face_recognition` library to provide:
-
-  - Face encoding (128-d vector per face)
-  - Known-faces database management (register / list / delete)
-  - Identity matching against the database
-  - Batch comparison utilities
-
-The known-faces database is persisted as a JSON file mapping
-person_name -> list of encodings. Each registered person may have
-multiple reference photos for robustness.
-"""
-
-from __future__ import annotations
-
-import json
-from dataclasses import dataclass, asdict
-from pathlib import Path
-from typing import List, Optional, Dict, Any
-
-import numpy as np
-
-from config import settings, KNOWN_FACES_DIR, DATA_DIR
-from modules.face_detection import get_detector, FaceBox
-from utils.image_utils import crop_face
-
-DB_PATH = DATA_DIR / "face_database.json"
-
-
-# --------------------------------------------------------------------------- #
-# Data model
-# --------------------------------------------------------------------------- #
-@dataclass
-class RecognitionResult:
-    """Result of recognizing a single face."""
-    location: dict          # bounding box
-    encoding: List[float]   # 128-d embedding (for client-side use if needed)
-    best_match: Optional[str]   # person name or None
-    confidence: float       # 1 - (distance / tolerance) clipped to [0,1]
-    all_distances: Dict[str, float]  # person_name -> min distance
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-
-# --------------------------------------------------------------------------- #
-# Database
-# --------------------------------------------------------------------------- #
-class FaceDatabase:
-    """JSON-backed known-faces database."""
-
-    def __init__(self, path: Path = DB_PATH):
-        self.path = path
-        self.data: Dict[str, List[List[float]]] = {}
-        self._load()
-
-    def _load(self):
-        if self.path.exists():
-            with open(self.path, "r") as f:
-                self.data = json.load(f)
-        else:
-            self.data = {}
-
-    def _save(self):
-        with open(self.path, "w") as f:
-            json.dump(self.data, f, indent=2)
-
-    def list_persons(self) -> List[Dict[str, Any]]:
-        return [
-            {"name": name, "num_encodings": len(encs)}
-            for name, encs in self.data.items()
-        ]
-
-    def add_encoding(self, name: str, encoding: np.ndarray):
-        name = name.strip().lower().replace(" ", "_")
-        if name not in self.data:
-            self.data[name] = []
-        self.data[name].append(encoding.tolist())
-        self._save()
-
-    def remove_person(self, name: str) -> bool:
-        name = name.strip().lower().replace(" ", "_")
-        if name in self.data:
-            del self.data[name]
-            self._save()
-            return True
-        return False
-
-    def clear(self):
-        self.data = {}
-        self._save()
-
-    def get_all_encodings(self) -> Dict[str, List[np.ndarray]]:
-        return {
-            name: [np.array(e) for e in encs]
-            for name, encs in self.data.items()
-        }
-
-
-# Singleton database
-db = FaceDatabase()
-
-
-# --------------------------------------------------------------------------- #
-# Core recognition
-# --------------------------------------------------------------------------- #
-def encode_face(img: np.ndarray, face_box: Optional[FaceBox] = None) -> np.ndarray:
-    """Compute the 128-d face encoding for a face in the image.
-
-    If face_box is None, the first detected face is used.
-    Uses the `face_recognition` library (CNN or HOG model).
-    """
-    import face_recognition
-
-    rgb = cv2_bgr_to_rgb(img)
-
-    if face_box is not None:
-        # Convert our FaceBox (x,y,w,h) to face_recognition format (top,right,bottom,left)
-        loc = (
-            face_box.y,
-            face_box.x + face_box.w,
-            face_box.y + face_box.h,
-            face_box.x,
-        )
-        encs = face_recognition.face_encodings(rgb, [loc], model=settings.face_recognition_model)
-    else:
-        encs = face_recognition.face_encodings(rgb, model=settings.face_recognition_model)
-
-    if not encs:
-        raise ValueError("No encodable face found in the image.")
-    return encs[0]
-
-
-def register_face(img: np.ndarray, name: str, detector_name: Optional[str] = None) -> dict:
-    """Register a face from an image under a given person name.
-
-    Returns:
-        {"name": str, "encodings_added": int, "total_encodings": int}
-    """
-    detector = get_detector(detector_name)
-    boxes = detector.detect(img)
-    if not boxes:
-        raise ValueError("No face detected in the provided image.")
-    if len(boxes) > 1:
-        # Use the largest face (most prominent subject)
-        boxes.sort(key=lambda b: b.area, reverse=True)
-
-    box = boxes[0]
-    encoding = encode_face(img, box)
-    name = name.strip().lower().replace(" ", "_")
-    db.add_encoding(name, encoding)
-
-    return {
-        "name": name,
-        "encodings_added": 1,
-        "total_encodings": len(db.data[name]),
-    }
-
-
-def recognize_faces(
-    img: np.ndarray,
-    detector_name: Optional[str] = None,
-    tolerance: Optional[float] = None,
-) -> dict:
-    """Recognize all faces in an image against the known-faces database.
-
-    Args:
-        img: BGR image array.
-        detector_name: which detector to use for locating faces.
-        tolerance: distance threshold (lower = stricter). Defaults to settings.
-
-    Returns:
-        {
-            "num_faces": int,
-            "tolerance": float,
-            "results": List[RecognitionResult dict],
-            "unknown_count": int,
-        }
-    """
-    import face_recognition
-
-    tol = tolerance or settings.face_recognition_tolerance
-    detector = get_detector(detector_name)
-    boxes = detector.detect(img)
-    rgb = cv2_bgr_to_rgb(img)
-
-    # Convert boxes to face_recognition location tuples
-    locations = [
-        (b.y, b.x + b.w, b.y + b.h, b.x) for b in boxes
-    ]
-    encodings = face_recognition.face_encodings(
-        rgb, locations, model=settings.face_recognition_model
-    )
-
-    known = db.get_all_encodings()
-    results: List[RecognitionResult] = []
-
-    for loc, enc in zip(locations, encodings):
-        all_distances: Dict[str, float] = {}
-        best_name: Optional[str] = None
-        best_dist: float = float("inf")
-
-        for person, person_encs in known.items():
-            dists = face_recognition.face_distance(person_encs, enc)
-            min_dist = float(min(dists)) if len(dists) else float("inf")
-            all_distances[person] = round(min_dist, 4)
-            if min_dist < best_dist:
-                best_dist = min_dist
-                best_name = person
-
-        if best_dist > tol:
-            best_name = None  # unknown
-
-        confidence = max(0.0, min(1.0, 1.0 - (best_dist / (tol * 2)))) if best_name else 0.0
-
-        results.append(RecognitionResult(
-            location={"top": loc[0], "right": loc[1], "bottom": loc[2], "left": loc[3]},
-            encoding=enc.tolist(),
-            best_match=best_name,
-            confidence=round(confidence, 4),
-            all_distances=all_distances,
-        ))
-
-    unknown_count = sum(1 for r in results if r.best_match is None)
-
-    return {
-        "num_faces": len(results),
-        "tolerance": tol,
-        "results": [r.to_dict() for r in results],
-        "unknown_count": unknown_count,
-    }
-
-
-# --------------------------------------------------------------------------- #
-# Helpers
-# --------------------------------------------------------------------------- #
-def cv2_bgr_to_rgb(img: np.ndarray) -> np.ndarray:
-    return img[:, :, ::-1]
-
-
-def compare_two_faces(img1: np.ndarray, img2: np.ndarray, tolerance: Optional[float] = None) -> dict:
-    """Compare the primary face in two images and return whether they match."""
-    tol = tolerance or settings.face_recognition_tolerance
-    enc1 = encode_face(img1)
-    enc2 = encode_face(img2)
-    import face_recognition
-    dist = float(face_recognition.face_distance([enc1], enc2)[0])
-    return {
-        "distance": round(dist, 4),
-        "tolerance": tol,
-        "is_same_person": dist <= tol,
-        "confidence": round(max(0.0, 1.0 - dist / (tol * 2)), 4),
-    }

download/face-intelligence-api/modules/reverse_search.py

@@ -1,283 +0,0 @@
-"""
-Reverse Image Search Module
-===========================
-Finds visually similar images online. Two backends:
-
-  1. Google Lens (default, free)   - automated via Selenium
-  2. SerpAPI Google Images         - paid, returns structured JSON
-
-Returns a list of ReverseSearchResult dataclasses with the matching
-image URL, source page, title, and (when available) a thumbnail.
-
-NOTE ON ETHICS & COMPLIANCE:
-  Reverse image search of faces to identify private individuals raises
-  serious privacy and legal concerns. This module is intended for:
-    - Lost-and-found / family reunification
-    - Authorized security and fraud-prevention workflows
-    - Journalism with documented public-interest justification
-    - Personal self-monitoring (searching for your OWN face)
-  It is NOT intended for de-anonymizing strangers, stalking, harassment,
-  or surveillance of people without their consent. Users are responsible
-  for complying with GDPR, CCPA, BIPA, ILCS, and all applicable laws.
-"""
-
-from __future__ import annotations
-
-import io
-import time
-from dataclasses import dataclass, asdict
-from pathlib import Path
-from typing import List, Optional
-
-import requests
-
-from config import settings, TEMP_DIR
-from utils.image_utils import save_temp_image
-import numpy as np
-
-
-# --------------------------------------------------------------------------- #
-# Data model
-# --------------------------------------------------------------------------- #
-@dataclass
-class ReverseSearchResult:
-    image_url: str
-    source_page: str
-    title: str
-    snippet: str = ""
-    thumbnail: str = ""
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-
-@dataclass
-class ReverseSearchResponse:
-    query_image: str
-    method: str
-    num_results: int
-    results: List[ReverseSearchResult]
-    error: Optional[str] = None
-
-    def to_dict(self) -> dict:
-        return {
-            "query_image": self.query_image,
-            "method": self.method,
-            "num_results": self.num_results,
-            "results": [r.to_dict() for r in self.results],
-            "error": self.error,
-        }
-
-
-# --------------------------------------------------------------------------- #
-# 1. Google Lens via Selenium (free)
-# --------------------------------------------------------------------------- #
-def reverse_search_google_lens(img: np.ndarray, max_results: Optional[int] = None) -> ReverseSearchResponse:
-    """Upload an image to Google Lens and parse visual-match results.
-
-    Uses Selenium to drive a headless Chrome browser. This is the free
-    method; it is somewhat brittle because Google's DOM changes over time.
-    For production use, prefer the SerpAPI backend.
-    """
-    max_results = max_results or settings.reverse_search_max_results
-    temp_path = save_temp_image(img, prefix="reverse_query")
-    query_image_str = str(temp_path)
-
-    try:
-        from selenium import webdriver
-        from selenium.webdriver.chrome.options import Options
-        from selenium.webdriver.chrome.service import Service
-        from selenium.webdriver.common.by import By
-        from selenium.webdriver.common.keys import Keys
-        from selenium.webdriver.support.ui import WebDriverWait
-        from selenium.webdriver.support import expected_conditions as EC
-        from webdriver_manager.chrome import ChromeDriverManager
-    except ImportError as e:
-        return ReverseSearchResponse(
-            query_image=query_image_str, method="google_lens",
-            num_results=0, results=[], error=f"Selenium not installed: {e}",
-        )
-
-    options = Options()
-    if settings.selenium_headless:
-        options.add_argument("--headless=new")
-    options.add_argument("--no-sandbox")
-    options.add_argument("--disable-dev-shm-usage")
-    options.add_argument("--disable-gpu")
-    options.add_argument(f"--user-agent={settings.user_agent}")
-    options.add_argument("--window-size=1920,1080")
-
-    try:
-        service = Service(ChromeDriverManager().install())
-        driver = webdriver.Chrome(service=service, options=options)
-    except Exception as e:
-        return ReverseSearchResponse(
-            query_image=query_image_str, method="google_lens",
-            num_results=0, results=[], error=f"Could not start Selenium: {e}",
-        )
-
-    results: List[ReverseSearchResult] = []
-    try:
-        driver.get("https://lens.google.com/")
-        wait = WebDriverWait(driver, 20)
-
-        # Click the "upload image" tab / file input
-        # Google Lens has a file input; we send the file path directly
-        file_inputs = driver.find_elements(By.CSS_SELECTOR, "input[type='file']")
-        if not file_inputs:
-            return ReverseSearchResponse(
-                query_image=query_image_str, method="google_lens",
-                num_results=0, results=[],
-                error="Could not locate file upload input on Google Lens page.",
-            )
-        file_inputs[0].send_keys(query_image_str)
-
-        # Wait for results to load
-        time.sleep(6)
-
-        # Parse visual matches - Google Lens renders result cards
-        # Selector may change; we try several known patterns
-        card_selectors = [
-            ".Vd9T9c",            # visual match card
-            "[jsname='R8qPlf']",
-            "div[jsname]",
-            "a[href]",
-        ]
-        cards = []
-        for sel in card_selectors:
-            cards = driver.find_elements(By.CSS_SELECTOR, sel)
-            if cards:
-                break
-
-        for card in cards[:max_results]:
-            try:
-                href = card.get_attribute("href") or ""
-                title = card.get_attribute("title") or card.text or ""
-                # Try to extract thumbnail
-                thumb = ""
-                try:
-                    img_el = card.find_element(By.TAG_NAME, "img")
-                    thumb = img_el.get_attribute("src") or ""
-                except Exception:
-                    pass
-                if href and href.startswith("http"):
-                    results.append(ReverseSearchResult(
-                        image_url=thumb or href,
-                        source_page=href,
-                        title=title.strip()[:200],
-                        thumbnail=thumb,
-                    ))
-            except Exception:
-                continue
-
-        return ReverseSearchResponse(
-            query_image=query_image_str, method="google_lens",
-            num_results=len(results), results=results,
-        )
-    except Exception as e:
-        return ReverseSearchResponse(
-            query_image=query_image_str, method="google_lens",
-            num_results=0, results=[], error=str(e),
-        )
-    finally:
-        driver.quit()
-
-
-# --------------------------------------------------------------------------- #
-# 2. SerpAPI Google Images (paid, structured)
-# --------------------------------------------------------------------------- #
-def reverse_search_serpapi(img: np.ndarray, max_results: Optional[int] = None) -> ReverseSearchResponse:
-    """Use SerpAPI's Google Reverse Image endpoint. Requires a paid API key."""
-    if not settings.serpapi_key:
-        return ReverseSearchResponse(
-            query_image="", method="serpapi",
-            num_results=0, results=[],
-            error="SerpAPI key not configured. Set FI_SERPAPI_KEY in environment.",
-        )
-
-    max_results = max_results or settings.reverse_search_max_results
-
-    # Step 1: upload image to SerpAPI's /upload endpoint
-    import cv2
-    ok, buffer = cv2.imencode(".jpg", img)
-    if not ok:
-        return ReverseSearchResponse(
-            query_image="", method="serpapi", num_results=0, results=[],
-            error="Could not encode query image.",
-        )
-
-    try:
-        upload_resp = requests.post(
-            "https://assets.serpapi.com/upload",
-            files={"file": ("query.jpg", io.BytesIO(buffer.tobytes()), "image/jpeg")},
-            data={"serp_api_key": settings.serpapi_key},
-            timeout=60,
-        )
-        upload_resp.raise_for_status()
-        uploaded_url = upload_resp.text.strip().strip('"')
-    except requests.RequestException as e:
-        return ReverseSearchResponse(
-            query_image="", method="serpapi", num_results=0, results=[],
-            error=f"Upload failed: {e}",
-        )
-
-    # Step 2: query the reverse image search engine
-    try:
-        params = {
-            "engine": "google_reverse_image",
-            "image_url": uploaded_url,
-            "api_key": settings.serpapi_key,
-        }
-        search_resp = requests.get("https://serpapi.com/search", params=params, timeout=60)
-        search_resp.raise_for_status()
-        data = search_resp.json()
-    except requests.RequestException as e:
-        return ReverseSearchResponse(
-            query_image=uploaded_url, method="serpapi", num_results=0, results=[],
-            error=f"Search failed: {e}",
-        )
-
-    results: List[ReverseSearchResult] = []
-    for match in data.get("image_results", [])[:max_results]:
-        results.append(ReverseSearchResult(
-            image_url=match.get("image", ""),
-            source_page=match.get("link", ""),
-            title=match.get("title", ""),
-            snippet=match.get("snippet", ""),
-            thumbnail=match.get("thumbnail", ""),
-        ))
-    # Also include full-page image matches if present
-    for match in data.get("inline_images", [])[:max_results]:
-        results.append(ReverseSearchResult(
-            image_url=match.get("image", ""),
-            source_page=match.get("link", ""),
-            title=match.get("title", ""),
-            snippet=match.get("snippet", ""),
-            thumbnail=match.get("thumbnail", ""),
-        ))
-
-    return ReverseSearchResponse(
-        query_image=uploaded_url, method="serpapi",
-        num_results=len(results), results=results[:max_results],
-    )
-
-
-# --------------------------------------------------------------------------- #
-# Dispatcher
-# --------------------------------------------------------------------------- #
-def reverse_image_search(
-    img: np.ndarray,
-    method: Optional[str] = None,
-    max_results: Optional[int] = None,
-) -> ReverseSearchResponse:
-    """Run a reverse image search using the configured or specified method."""
-    method = method or settings.reverse_search_method
-    if method == "google_lens":
-        return reverse_search_google_lens(img, max_results)
-    elif method == "serpapi":
-        return reverse_search_serpapi(img, max_results)
-    else:
-        return ReverseSearchResponse(
-            query_image="", method=method, num_results=0, results=[],
-            error=f"Unknown method: {method}",
-        )

download/face-intelligence-api/modules/web_scraper.py

@@ -1,235 +0,0 @@
-"""
-Web Scraping Module
-===================
-Extracts images from web pages using two strategies:
-
-  1. BeautifulSoup + requests  - fast, works on static HTML
-  2. Selenium (headless Chrome) - handles JavaScript-rendered content
-
-Both strategies return a list of ImageResult dataclasses containing
-the image URL, alt text, source page, and (optionally) downloaded
-bytes for downstream processing.
-"""
-
-from __future__ import annotations
-
-import re
-from dataclasses import dataclass, asdict
-from pathlib import Path
-from typing import List, Optional
-from urllib.parse import urljoin, urlparse
-
-import requests
-from bs4 import BeautifulSoup
-
-from config import settings
-from utils.image_utils import url_to_numpy, bytes_to_numpy
-import numpy as np
-
-
-# --------------------------------------------------------------------------- #
-# Data model
-# --------------------------------------------------------------------------- #
-@dataclass
-class ImageResult:
-    url: str
-    alt: str
-    source_page: str
-    width: Optional[int] = None
-    height: Optional[int] = None
-    downloaded: bool = False
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-
-@dataclass
-class ScrapeResult:
-    url: str
-    method: str
-    num_images: int
-    images: List[ImageResult]
-    title: str = ""
-    error: Optional[str] = None
-
-    def to_dict(self) -> dict:
-        return {**asdict(self), "images": [i.to_dict() for i in self.images]}
-
-
-# --------------------------------------------------------------------------- #
-# Static scraper (BeautifulSoup)
-# --------------------------------------------------------------------------- #
-def scrape_static(url: str, max_images: Optional[int] = None) -> ScrapeResult:
-    """Scrape images from a static HTML page using requests + BeautifulSoup."""
-    max_images = max_images or settings.scrape_max_images
-    headers = {"User-Agent": settings.user_agent}
-
-    try:
-        resp = requests.get(url, headers=headers, timeout=settings.scrape_timeout)
-        resp.raise_for_status()
-    except requests.RequestException as e:
-        return ScrapeResult(url=url, method="static", num_images=0, images=[], error=str(e))
-
-    soup = BeautifulSoup(resp.text, "lxml")
-    title = soup.title.string.strip() if soup.title and soup.title.string else ""
-
-    images: List[ImageResult] = []
-    seen: set = set()
-
-    for img_tag in soup.find_all("img"):
-        if len(images) >= max_images:
-            break
-        src = img_tag.get("src") or img_tag.get("data-src") or img_tag.get("data-lazy-src")
-        if not src:
-            continue
-        absolute_url = urljoin(url, src)
-        if absolute_url in seen:
-            continue
-        # Filter out tiny icons / tracking pixels
-        w = img_tag.get("width")
-        h = img_tag.get("height")
-        if w and h:
-            try:
-                if int(w) < 50 or int(h) < 50:
-                    continue
-            except ValueError:
-                pass
-        seen.add(absolute_url)
-        images.append(ImageResult(
-            url=absolute_url,
-            alt=(img_tag.get("alt") or "").strip(),
-            source_page=url,
-            width=int(w) if w and w.isdigit() else None,
-            height=int(h) if h and h.isdigit() else None,
-        ))
-
-    return ScrapeResult(url=url, method="static", num_images=len(images), images=images, title=title)
-
-
-# --------------------------------------------------------------------------- #
-# Dynamic scraper (Selenium)
-# --------------------------------------------------------------------------- #
-def scrape_dynamic(url: str, max_images: Optional[int] = None, scroll: bool = True) -> ScrapeResult:
-    """Scrape images from a JS-rendered page using headless Selenium."""
-    max_images = max_images or settings.scrape_max_images
-
-    try:
-        from selenium import webdriver
-        from selenium.webdriver.chrome.options import Options
-        from selenium.webdriver.chrome.service import Service
-        from selenium.webdriver.common.by import By
-        from webdriver_manager.chrome import ChromeDriverManager
-    except ImportError as e:
-        return ScrapeResult(url=url, method="dynamic", num_images=0, images=[],
-                            error=f"Selenium not installed: {e}")
-
-    options = Options()
-    if settings.selenium_headless:
-        options.add_argument("--headless=new")
-    options.add_argument("--no-sandbox")
-    options.add_argument("--disable-dev-shm-usage")
-    options.add_argument("--disable-gpu")
-    options.add_argument(f"--user-agent={settings.user_agent}")
-    options.add_argument("--window-size=1920,1080")
-
-    try:
-        service = Service(ChromeDriverManager().install())
-        driver = webdriver.Chrome(service=service, options=options)
-    except Exception as e:
-        return ScrapeResult(url=url, method="dynamic", num_images=0, images=[],
-                            error=f"Could not start Selenium: {e}")
-
-    driver.implicitly_wait(settings.selenium_implicit_wait)
-    images: List[ImageResult] = []
-    seen: set = set()
-
-    try:
-        driver.get(url)
-        title = driver.title or ""
-
-        if scroll:
-            # Scroll to bottom to trigger lazy-loading
-            last_height = driver.execute_script("return document.body.scrollHeight")
-            for _ in range(5):  # max 5 scroll iterations
-                driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
-                import time
-                time.sleep(1.5)
-                new_height = driver.execute_script("return document.body.scrollHeight")
-                if new_height == last_height:
-                    break
-                last_height = new_height
-
-        img_elements = driver.find_elements(By.TAG_NAME, "img")
-        for el in img_elements:
-            if len(images) >= max_images:
-                break
-            src = el.get_attribute("src") or el.get_attribute("data-src")
-            if not src or src.startswith("data:"):
-                continue
-            absolute_url = urljoin(url, src)
-            if absolute_url in seen:
-                continue
-            seen.add(absolute_url)
-            w = el.get_attribute("naturalWidth")
-            h = el.get_attribute("naturalHeight")
-            try:
-                w_int = int(w) if w else None
-                h_int = int(h) if h else None
-                if w_int and h_int and (w_int < 50 or h_int < 50):
-                    continue
-            except (TypeError, ValueError):
-                w_int = h_int = None
-            images.append(ImageResult(
-                url=absolute_url,
-                alt=(el.get_attribute("alt") or "").strip(),
-                source_page=url,
-                width=w_int,
-                height=h_int,
-            ))
-
-        return ScrapeResult(url=url, method="dynamic", num_images=len(images),
-                            images=images, title=title)
-    except Exception as e:
-        return ScrapeResult(url=url, method="dynamic", num_images=0, images=[], error=str(e))
-    finally:
-        driver.quit()
-
-
-# --------------------------------------------------------------------------- #
-# Convenience: download images from a ScrapeResult
-# --------------------------------------------------------------------------- #
-def download_images(scrape_result: ScrapeResult, limit: Optional[int] = None) -> List[Optional[np.ndarray]]:
-    """Download up to `limit` images from the scrape result. Returns BGR arrays."""
-    limit = limit or len(scrape_result.images)
-    downloaded: List[Optional[np.ndarray]] = []
-    for img_res in scrape_result.images[:limit]:
-        try:
-            arr = url_to_numpy(img_res.url)
-            downloaded.append(arr)
-            img_res.downloaded = True
-        except Exception:
-            downloaded.append(None)
-    return downloaded
-
-
-# --------------------------------------------------------------------------- #
-# Site-specific helpers (extensible)
-# --------------------------------------------------------------------------- #
-def is_social_media_url(url: str) -> dict:
-    """Identify which social platform a URL belongs to."""
-    parsed = urlparse(url)
-    host = parsed.netloc.lower()
-    platforms = {
-        "instagram": ("instagram.com",),
-        "twitter": ("twitter.com", "x.com"),
-        "facebook": ("facebook.com", "fb.com"),
-        "linkedin": ("linkedin.com",),
-        "tiktok": ("tiktok.com",),
-        "youtube": ("youtube.com", "youtu.be"),
-        "pinterest": ("pinterest.com",),
-        "reddit": ("reddit.com",),
-    }
-    for platform, domains in platforms.items():
-        if any(d in host for d in domains):
-            return {"is_social": True, "platform": platform}
-    return {"is_social": False, "platform": None}

download/face-intelligence-api/requirements.txt

@@ -1,37 +0,0 @@
-# ============================================================
-# Face Intelligence API - Dependencies
-# ============================================================
-
-# --- Web Framework ---
-fastapi==0.110.0
-uvicorn[standard]==0.27.1
-python-multipart==0.0.9
-pydantic==2.6.1
-pydantic-settings==2.2.1
-
-# --- Face Detection & Recognition ---
-opencv-python==4.9.0.80
-opencv-contrib-python==4.9.0.80
-face-recognition==1.3.0
-dlib==19.24.2
-numpy==1.26.4
-Pillow==10.2.0
-
-# --- MTCNN (optional alternative detector) ---
-mtcnn==0.1.1
-tensorflow==2.16.1
-
-# --- Web Scraping ---
-beautifulsoup4==4.12.3
-lxml==5.1.0
-requests==2.31.0
-selenium==4.18.1
-webdriver-manager==4.0.1
-urllib3==2.2.1
-
-# --- Utilities ---
-aiofiles==23.2.1
-httpx==0.27.0
-tqdm==4.66.2
-python-dotenv==1.0.1
-loguru==0.7.2

download/face-intelligence-api/utils/__init__.py

@@ -1 +0,0 @@
-"""Utils package."""

download/face-intelligence-api/utils/image_utils.py

@@ -1,94 +0,0 @@
-"""
-Image utility helpers: loading, decoding, encoding, face cropping.
-"""
-
-from __future__ import annotations
-
-import base64
-import io
-from pathlib import Path
-from typing import Tuple
-
-import cv2
-import numpy as np
-from PIL import Image
-
-from config import TEMP_DIR
-
-
-def bytes_to_numpy(image_bytes: bytes) -> np.ndarray:
-    """Decode raw image bytes into an OpenCV BGR numpy array."""
-    nparr = np.frombuffer(image_bytes, np.uint8)
-    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
-    if img is None:
-        raise ValueError("Could not decode image bytes. Unsupported format or corrupted data.")
-    return img
-
-
-def base64_to_numpy(b64_string: str) -> np.ndarray:
-    """Decode a base64-encoded image string into an OpenCV BGR numpy array."""
-    if "," in b64_string:
-        b64_string = b64_string.split(",", 1)[1]
-    raw = base64.b64decode(b64_string)
-    return bytes_to_numpy(raw)
-
-
-def numpy_to_base64(img: np.ndarray, fmt: str = ".jpg") -> str:
-    """Encode an OpenCV BGR numpy array as a base64 string."""
-    ok, buffer = cv2.imencode(fmt, img)
-    if not ok:
-        raise ValueError("Could not encode image.")
-    return base64.b64encode(buffer).decode("utf-8")
-
-
-def save_temp_image(img: np.ndarray, prefix: str = "img", fmt: str = ".jpg") -> Path:
-    """Save an image to the temp directory and return its path."""
-    path = TEMP_DIR / f"{prefix}_{int(cv2.getTickCount())}{fmt}"
-    cv2.imwrite(str(path), img)
-    return path
-
-
-def crop_face(img: np.ndarray, box: Tuple[int, int, int, int], margin: float = 0.2) -> np.ndarray:
-    """Crop a face region from an image with an optional margin.
-
-    Args:
-        img: BGR image array.
-        box: (x, y, w, h) bounding box.
-        margin: fractional margin to add around the face.
-    """
-    x, y, w, h = box
-    dx = int(w * margin)
-    dy = int(h * margin)
-    x0 = max(0, x - dx)
-    y0 = max(0, y - dy)
-    x1 = min(img.shape[1], x + w + dx)
-    y1 = min(img.shape[0], y + h + dy)
-    return img[y0:y1, x0:x1]
-
-
-def resize_for_display(img: np.ndarray, max_width: int = 800) -> np.ndarray:
-    """Resize image maintaining aspect ratio for display purposes."""
-    h, w = img.shape[:2]
-    if w <= max_width:
-        return img
-    scale = max_width / w
-    return cv2.resize(img, (max_width, int(h * scale)), interpolation=cv2.INTER_AREA)
-
-
-def url_to_numpy(url: str, timeout: int = 15) -> np.ndarray:
-    """Download an image from a URL and return it as a BGR numpy array."""
-    import requests
-    headers = {"User-Agent": "Mozilla/5.0"}
-    resp = requests.get(url, headers=headers, timeout=timeout, stream=True)
-    resp.raise_for_status()
-    return bytes_to_numpy(resp.content)
-
-
-def validate_image_file(path: Path) -> bool:
-    """Validate that a file is a readable image."""
-    try:
-        with Image.open(path) as im:
-            im.verify()
-        return True
-    except Exception:
-        return False