File size: 5,202 Bytes
7e25f7a | 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """
OSINT service — orchestrates reverse-image-search across multiple
providers and merges results into a unified evidence set.
This is NOT a provider itself — it's a service that:
1. Runs the existing reverse-search providers (serpapi, social_lookup)
via the orchestrator.
2. Collects raw results from each provider.
3. Merges + deduplicates via cores.osint.merger.
4. Classifies each source (social/news/forum/marketplace/etc.) via
cores.osint.sources.
5. Returns an OSINTResult with confidence-scored, classified matches.
Pure stdlib for the merging logic; relies on existing providers for
network calls.
"""
from __future__ import annotations
import time
from typing import Optional
from cores.osint import merge_reverse_results, classify_source, SourceType
from metrics.collector import MetricsCollector
from models.jobs import JobRequest
from models.reports import OSINTResult, OSINTMatch
from models.providers import ProviderCapability
from orchestrator.runner import Orchestrator
from pipeline import InputValidator, ImagePreprocessor, ImageHasher, FeatureExtractor
from utils.logging import execution_context, new_execution_id
class OSINTService:
"""Orchestrates reverse-image-search across providers + merges results."""
def __init__(
self,
orchestrator: Orchestrator,
metrics: MetricsCollector,
validator: InputValidator,
preprocessor: ImagePreprocessor,
hasher: ImageHasher,
feature_extractor: FeatureExtractor,
) -> None:
self._orchestrator = orchestrator
self._metrics = metrics
self._validator = validator
self._preprocessor = preprocessor
self._hasher = hasher
self._feature_extractor = feature_extractor
async def reverse_search(self, request: JobRequest) -> dict:
"""Run reverse-image-search across all enabled providers + merge."""
eid = new_execution_id()
with execution_context(execution_id=eid, provider_id="osint_service"):
t0 = time.perf_counter()
# Validate input — OSINT requires an image (URL or base64)
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"}
# Preprocess
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,
)
# If the user provided a URL, attach it so social_lookup can use it
if request.image_url:
pipeline_output.scrape_url = request.image_url
# Run all reverse-search providers
results = await self._orchestrator.run(
pipeline_output=pipeline_output,
capabilities=[ProviderCapability.REVERSE_SEARCH],
provider_whitelist=request.providers or None,
execution_id=eid,
)
# Collect raw matches per provider
provider_results: dict[str, list[dict]] = {}
providers_invoked: list[str] = []
providers_succeeded: list[str] = []
providers_failed: list[str] = []
for name, result in results.items():
providers_invoked.append(name)
if result.success:
providers_succeeded.append(name)
raw_matches = result.normalized.get("results", [])
if raw_matches:
provider_results[name] = raw_matches
else:
providers_failed.append(name)
# Merge + classify
merged = merge_reverse_results(provider_results)
# Build source-type breakdown
breakdown: dict[str, int] = {}
for m in merged:
st = m.get("source_type", "unknown")
breakdown[st] = breakdown.get(st, 0) + 1
elapsed = (time.perf_counter() - t0) * 1000.0
osint_result = OSINTResult(
providers_invoked=providers_invoked,
providers_succeeded=providers_succeeded,
providers_failed=providers_failed,
total_matches=len(merged),
matches=[OSINTMatch(**m) for m in merged],
source_type_breakdown=breakdown,
elapsed_ms=round(elapsed, 3),
)
self._metrics.timings.record("job.osint", elapsed)
self._metrics.counters.inc("jobs.osint.completed")
return {
"success": True,
"osint": osint_result.model_dump(),
"elapsed_ms": round(elapsed, 3),
}
|