""" Provider Registry — auto-discovers and registers providers. The registry is a CLASS (not a module-level singleton) so it can be instantiated and injected through the DI container. Adding a new provider requires ONLY: 1. Create the provider file under providers//.py 2. Add one entry to PROVIDER_MANIFEST below. The orchestrator never changes — it queries the registry by capability. IMPORTANT: Every entry in PROVIDER_MANIFEST must point to a real file. Do not add phantom entries — if a provider is not implemented, do not list it. The /providers endpoint surfaces exactly what is real. """ from __future__ import annotations import importlib from dataclasses import dataclass from typing import Optional from loguru import logger from config.settings import Settings from models.providers import ProviderCapability, ProviderStatus, ProviderInfo from providers.base import Provider, BaseProvider # --------------------------------------------------------------------------- # # Manifest — one line per provider. Add new providers here ONLY. # Every entry MUST point to an existing file. # --------------------------------------------------------------------------- # @dataclass class ManifestEntry: name: str module_path: str class_name: str capability: ProviderCapability enable_flag: str # attribute name on Settings description: str = "" optional_dependency: bool = False # True = gracefully skip if import fails PROVIDER_MANIFEST: list[ManifestEntry] = [ # ------------------------------------------------------------------ # # Detection # ------------------------------------------------------------------ # ManifestEntry("haar", "providers.detection.haar", "HaarDetector", ProviderCapability.DETECTION, "enable_haar", "OpenCV Haar Cascade — fast, frontal faces only"), ManifestEntry("dnn", "providers.detection.dnn", "DNNDetector", ProviderCapability.DETECTION, "enable_dnn", "OpenCV DNN Caffe SSD — balanced accuracy/speed", optional_dependency=True), # ------------------------------------------------------------------ # # Recognition # ------------------------------------------------------------------ # ManifestEntry("insightface", "providers.recognition.insightface", "InsightFaceProvider", ProviderCapability.RECOGNITION, "enable_insightface", "ArcFace 512-d embeddings via ONNX — SOTA accuracy", optional_dependency=True), # ------------------------------------------------------------------ # # Reverse image search # ------------------------------------------------------------------ # ManifestEntry("serpapi", "providers.reverse.serpapi", "SerpAPIProvider", ProviderCapability.REVERSE_SEARCH, "enable_serpapi", "SerpAPI Google Reverse Image (paid, structured)"), ManifestEntry("social_lookup", "providers.reverse.social_lookup", "SocialLookupProvider", ProviderCapability.REVERSE_SEARCH, "enable_social_lookup", "Constructs reverse-search URLs for Google/Bing/Yandex/TinEye/Reddit"), # ------------------------------------------------------------------ # # Reverse face search (PimEyes-style indexed search) # ------------------------------------------------------------------ # ManifestEntry("face_index", "providers.reverse.face_index_provider", "FaceIndexProvider", ProviderCapability.REVERSE_FACE_SEARCH, "enable_face_index", "Indexed reverse face search — sqlite-vec + ArcFace embeddings"), # ------------------------------------------------------------------ # # Image Analysis # ------------------------------------------------------------------ # ManifestEntry("image_quality", "providers.image_analysis.image_quality", "ImageQualityProvider", ProviderCapability.IMAGE_ANALYSIS, "enable_image_quality", "Image quality: brightness, contrast, sharpness, noise"), ManifestEntry("image_properties", "providers.image_analysis.image_properties", "ImagePropertiesProvider", ProviderCapability.IMAGE_ANALYSIS, "enable_image_properties", "Image properties: dimensions, color profile, dominant colors"), # ------------------------------------------------------------------ # # Metadata # ------------------------------------------------------------------ # ManifestEntry("exif", "providers.metadata.exif", "EXIFProvider", ProviderCapability.METADATA, "enable_exif", "EXIF metadata extraction (Pillow)"), # ------------------------------------------------------------------ # # Forensics # ------------------------------------------------------------------ # ManifestEntry("image_integrity", "providers.forensics.image_integrity", "ImageIntegrityProvider", ProviderCapability.FORENSICS, "enable_image_integrity", "Image integrity: corruption check, hash, size sanity"), ManifestEntry("duplicate_detector", "providers.forensics.duplicate_detector", "DuplicateDetectorProvider", ProviderCapability.FORENSICS, "enable_duplicate_detector", "Duplicate detection via perceptual hashing (pHash + dHash)"), ManifestEntry("ela", "providers.forensics.ela", "ELAProvider", ProviderCapability.FORENSICS, "enable_ela", "Error Level Analysis — JPEG re-encode diff for manipulation detection"), ManifestEntry("image_similarity", "providers.forensics.image_similarity", "ImageSimilarityProvider", ProviderCapability.FORENSICS, "enable_image_similarity", "Image similarity via perceptual hash comparison"), # ------------------------------------------------------------------ # # OCR (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("ocr", "providers.ocr.rapidocr_provider", "RapidOCRProvider", ProviderCapability.OCR, "enable_ocr", "OCR via RapidOCR (ONNX) — multilingual text extraction", optional_dependency=True), # ------------------------------------------------------------------ # # Object Detection (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("yolov8", "providers.object_detection.yolov8", "YOLOv8Provider", ProviderCapability.OBJECT_DETECTION, "enable_yolov8", "YOLOv8n object detection (80 COCO classes) via ONNX", optional_dependency=True), ManifestEntry("qr_code", "providers.object_intelligence.qr_code", "QRCodeProvider", ProviderCapability.OBJECT_DETECTION, "enable_qr_code", "QR code detection + decoding via OpenCV"), ManifestEntry("barcode", "providers.object_intelligence.barcode", "BarcodeProvider", ProviderCapability.OBJECT_DETECTION, "enable_barcode", "Barcode detection + decoding via OpenCV (>= 4.5.2)"), # ------------------------------------------------------------------ # # Scene Classification (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("places365", "providers.scene.places365", "Places365Provider", ProviderCapability.SCENE_RECOGNITION, "enable_places365", "Scene classification (365 categories) via MobileNet ONNX", optional_dependency=True), # ------------------------------------------------------------------ # # NSFW Detection (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("nudenet", "providers.nsfw.nudenet", "NudeNetProvider", ProviderCapability.NSFW_DETECTION, "enable_nudenet", "NSFW detection via NudeNet ONNX — body-part detection", optional_dependency=True), # ------------------------------------------------------------------ # # AI Generated Image Detection (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("ai_image_detector", "providers.ai_detection.laid", "AIImageDetectorProvider", ProviderCapability.AI_IMAGE_DETECTION, "enable_ai_image_detector", "AI-generated image detection via lightweight CNN ONNX", optional_dependency=True), # ------------------------------------------------------------------ # # Embeddings (optional — requires onnxruntime) # ------------------------------------------------------------------ # ManifestEntry("mobilenet_embed", "providers.embedding.mobilenet", "MobileNetEmbeddingProvider", ProviderCapability.EMBEDDING, "enable_mobilenet_embed", "Image embeddings (1280-d) via MobileNetV2-Lite ONNX", optional_dependency=True), ] # --------------------------------------------------------------------------- # # Registry # --------------------------------------------------------------------------- # class ProviderRegistry: """Injectable registry. Owns provider instances.""" def __init__(self, settings: Settings) -> None: self._settings = settings self._providers: dict[str, Provider] = {} self._manifest_errors: dict[str, str] = {} def discover(self) -> None: """Auto-discover and instantiate every enabled provider.""" for entry in PROVIDER_MANIFEST: enabled = getattr(self._settings, entry.enable_flag, False) if not enabled: logger.debug(f"Provider {entry.name} disabled by config ({entry.enable_flag}=False)") continue try: module = importlib.import_module(entry.module_path) cls = getattr(module, entry.class_name) instance = cls(settings=self._settings) self._providers[entry.name] = instance logger.info(f"Registered provider: {entry.name} ({entry.capability.value})") except ImportError as e: self._manifest_errors[entry.name] = f"missing dependency: {e}" logger.warning(f"Provider {entry.name} skipped (missing optional dep): {e}") except Exception as e: self._manifest_errors[entry.name] = f"init error: {e}" logger.error(f"Failed to register {entry.name}: {e}") def get(self, name: str) -> Optional[Provider]: return self._providers.get(name) def list(self) -> list[Provider]: return list(self._providers.values()) def list_by_capability(self, capability: ProviderCapability) -> list[Provider]: return [p for p in self._providers.values() if p.capability == capability] def list_names(self) -> list[str]: return sorted(self._providers.keys()) def info(self) -> list[ProviderInfo]: out: list[ProviderInfo] = [] for entry in PROVIDER_MANIFEST: provider = self._providers.get(entry.name) if provider is None: enabled = getattr(self._settings, entry.enable_flag, False) if not enabled: status = ProviderStatus.DISABLED elif entry.name in self._manifest_errors: status = ProviderStatus.NOT_CONFIGURED else: status = ProviderStatus.DISABLED out.append(ProviderInfo( name=entry.name, capability=entry.capability, status=status, available=False, description=entry.description, )) else: try: available = provider.is_available() except Exception: available = False status = ProviderStatus.HEALTHY if available else ProviderStatus.NOT_CONFIGURED out.append(ProviderInfo( name=entry.name, capability=entry.capability, status=status, available=available, description=entry.description, )) return out def manifest_errors(self) -> dict[str, str]: return dict(self._manifest_errors)