File size: 3,396 Bytes
23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
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")
# Step 1: 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('"')
# Step 2: search
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()
# Step 3: parse
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
|