| """ |
| 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, |
| ) |
|
|
| |
| 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), |
| } |
|
|
