File size: 3,492 Bytes
aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d | 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 99 100 101 | """
Search service — runs scraping + reverse-image-search jobs.
"""
from __future__ import annotations
import time
from confidence.engine import ConfidenceEngine
from confidence.conflicts import ConflictDetector
from metrics.collector import MetricsCollector
from models.jobs import JobRequest
from models.providers import ProviderCapability
from normalization.merger import ReportMerger
from orchestrator.runner import Orchestrator
from pipeline import (
InputValidator,
ImagePreprocessor,
ImageHasher,
FeatureExtractor,
)
from utils.logging import execution_context, new_execution_id
class SearchService:
"""Handles scraping + reverse-image-search jobs."""
def __init__(
self,
orchestrator: Orchestrator,
metrics: MetricsCollector,
validator: InputValidator,
preprocessor: ImagePreprocessor,
hasher: ImageHasher,
feature_extractor: FeatureExtractor,
confidence_engine: ConfidenceEngine,
conflict_detector: ConflictDetector,
) -> None:
self._orchestrator = orchestrator
self._metrics = metrics
self._validator = validator
self._preprocessor = preprocessor
self._hasher = hasher
self._feature_extractor = feature_extractor
self._merger = ReportMerger(confidence_engine, conflict_detector)
async def search(self, request: JobRequest) -> dict:
eid = new_execution_id()
with execution_context(execution_id=eid, provider_id="search_service"):
t0 = time.perf_counter()
vr = self._validator.validate(
image_url=request.image_url,
image_base64=request.image_base64,
)
if not vr.valid:
return {"success": False, "error": vr.error, "error_type": "ValidationError"}
if vr.source == "url":
pre = self._preprocessor.from_url(request.image_url)
else:
pre = self._preprocessor.from_bytes(vr.image_bytes, vr.source)
img_hash = self._hasher.hash(pre.image)
pipeline_output = self._feature_extractor.extract(
pre.image, img_hash, pre.width, pre.height, pre.source,
original_bytes=pre.original_bytes,
original_format=pre.original_format,
)
# Optionally target a specific URL for scraping
target_url = request.options.get("scrape_url")
if target_url:
pipeline_output.scrape_url = target_url
results = await self._orchestrator.run(
pipeline_output=pipeline_output,
capabilities=[
ProviderCapability.SCRAPING,
ProviderCapability.REVERSE_SEARCH,
],
provider_whitelist=request.providers or None,
execution_id=eid,
)
elapsed = (time.perf_counter() - t0) * 1000.0
report = self._merger.merge(
results=results,
image_hash=img_hash,
job_id=eid,
total_elapsed_ms=elapsed,
kind="search",
)
self._metrics.timings.record("job.search", elapsed)
self._metrics.counters.inc("jobs.search.completed")
return {
"success": True,
"report": report.model_dump(),
"elapsed_ms": round(elapsed, 3),
}
|