1aa93ed1-1171-4784-95b4-bff53f7711a3

download/face-intelligence-api/config.py

@@ -0,0 +1,87 @@
+"""
+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

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

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

@@ -0,0 +1,261 @@
+"""
+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

@@ -0,0 +1,255 @@
+"""
+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

@@ -0,0 +1,283 @@
+"""
+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

@@ -0,0 +1,235 @@
+"""
+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

@@ -0,0 +1,37 @@
+# ============================================================
+# 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

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

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

@@ -0,0 +1,94 @@
+"""
+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