| """ |
| SerpAPI Google Reverse Image Search provider. |
| |
| Uses cores.search.http for the shared session — no duplicated |
| requests-session code. All hashing delegated to cores.vision. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import io |
| from typing import Any |
|
|
| import cv2 |
| import numpy as np |
|
|
| from config.settings import Settings, settings as _default_settings |
| from cores.search import shared_session |
| from pipeline.feature_extraction import PipelineOutput |
| from providers.base import BaseProvider, ProviderCapability |
|
|
|
|
| class SerpAPIProvider(BaseProvider): |
| name = "serpapi" |
| capability = ProviderCapability.REVERSE_SEARCH |
|
|
| UPLOAD_URL = "https://assets.serpapi.com/upload" |
| SEARCH_URL = "https://serpapi.com/search" |
|
|
| def __init__(self, settings: Settings | None = None) -> None: |
| super().__init__(settings=settings or _default_settings) |
| self._api_key = self._settings.serpapi_key |
| self._session = shared_session() |
|
|
| def is_available(self) -> bool: |
| return bool(self._api_key) |
|
|
| def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]: |
| if not self._api_key: |
| raise RuntimeError("SerpAPI key not configured") |
|
|
| img: np.ndarray = pipeline_output.image |
| ok, buffer = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 90]) |
| if not ok: |
| raise RuntimeError("Could not encode image for SerpAPI upload") |
|
|
| |
| upload_resp = self._session.post( |
| self.UPLOAD_URL, |
| files={"file": ("query.jpg", io.BytesIO(buffer.tobytes()), "image/jpeg")}, |
| data={"serp_api_key": self._api_key}, |
| timeout=60, |
| ) |
| upload_resp.raise_for_status() |
| uploaded_url = upload_resp.text.strip().strip('"') |
|
|
| |
| params = { |
| "engine": "google_reverse_image", |
| "image_url": uploaded_url, |
| "api_key": self._api_key, |
| } |
| search_resp = self._session.get(self.SEARCH_URL, params=params, timeout=60) |
| search_resp.raise_for_status() |
| data: dict[str, Any] = search_resp.json() |
|
|
| |
| max_results = self._settings.reverse_search_max_results |
| results: list[dict] = [] |
| for match in data.get("image_results", [])[:max_results]: |
| results.append({ |
| "image_url": match.get("image", ""), |
| "source_page": match.get("link", ""), |
| "title": match.get("title", ""), |
| "snippet": match.get("snippet", ""), |
| "thumbnail": match.get("thumbnail", ""), |
| }) |
| for match in data.get("inline_images", [])[:max_results]: |
| results.append({ |
| "image_url": match.get("image", ""), |
| "source_page": match.get("link", ""), |
| "title": match.get("title", ""), |
| "snippet": match.get("snippet", ""), |
| "thumbnail": match.get("thumbnail", ""), |
| }) |
|
|
| raw = { |
| "uploaded_image_url": uploaded_url, |
| "total_results": len(results), |
| "search_parameters": params, |
| "search_metadata": data.get("search_metadata", {}), |
| } |
| normalized = { |
| "results": results, |
| "total": len(results), |
| "uploaded_image_url": uploaded_url, |
| } |
| return raw, normalized |
|
|
