""" Forensic Metadata service — produces a comprehensive forensic metadata report from an image's embedded metadata. Combines: - EXIF (via cores.metadata.extract_all) - GPS coordinates (via cores.metadata.gps_to_coords) - ICC profile (via cores.metadata.extract_icc_profile) - Thumbnail extraction (via cores.metadata.extract_thumbnail) - Editing history (via cores.metadata.extract_editing_history) - Camera fingerprint (via cores.metadata.camera_fingerprint) - Compression analysis (via cores.metadata.analyze_compression) - Timestamp normalization (via cores.metadata.normalize_timestamp) - Lens info (via cores.metadata.extract_lens_info) Pure Pillow + NumPy — no external deps. """ from __future__ import annotations import time from cores.metadata import ( extract_all, extract_icc_profile, extract_thumbnail, extract_embedded_preview, extract_editing_history, camera_fingerprint, analyze_compression, normalize_timestamp, extract_lens_info, ) from models.jobs import JobRequest from models.reports import ForensicMetadataReport from pipeline import InputValidator, ImagePreprocessor, ImageHasher from utils.logging import execution_context, new_execution_id class ForensicMetadataService: """Produces a ForensicMetadataReport from image metadata.""" def __init__( self, validator: InputValidator, preprocessor: ImagePreprocessor, hasher: ImageHasher, ) -> None: self._validator = validator self._preprocessor = preprocessor self._hasher = hasher async def analyze(self, request: JobRequest) -> dict: """Extract comprehensive forensic metadata from an image.""" eid = new_execution_id() with execution_context(execution_id=eid, provider_id="forensic_metadata_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) original_bytes = pre.original_bytes or vr.image_bytes if not original_bytes: return {"success": False, "error": "No original bytes available for metadata extraction", "error_type": "ValidationError"} # Extract everything in one pass meta = extract_all(original_bytes) # ICC profile icc = extract_icc_profile(original_bytes) # Thumbnail thumb = extract_thumbnail(original_bytes) # Embedded preview has_preview = extract_embedded_preview(original_bytes) # Editing history edit_history = extract_editing_history(meta.get("exif", {})) # Lens info lens = extract_lens_info(meta.get("exif", {})) # Camera fingerprint fingerprint = camera_fingerprint(original_bytes, (pre.height, pre.width)) # Compression analysis compression = analyze_compression(original_bytes) # Timestamp normalization ts_normalized = None if meta.get("capture_time"): ts_normalized = normalize_timestamp(meta["capture_time"]) elapsed = (time.perf_counter() - t0) * 1000.0 report = ForensicMetadataReport( format=meta.get("format"), exif=meta.get("exif", {}), xmp={} if not meta.get("xmp") else {"raw_length": len(meta["xmp"])}, iptc=meta.get("iptc", {}), icc_profile=icc, gps=meta.get("gps_coords"), camera_make=meta.get("camera_make"), camera_model=meta.get("camera_model"), lens_model=lens, software=meta.get("software"), capture_time=meta.get("capture_time"), capture_time_iso=ts_normalized.get("iso") if ts_normalized else None, timezone_estimate=ts_normalized.get("timezone") if ts_normalized else None, editing_history=edit_history, thumbnail_extracted=bool(thumb and thumb.get("present")), embedded_preview=has_preview, camera_fingerprint=fingerprint, compression_analysis=compression, ) return { "success": True, "forensic_metadata": report.model_dump(), "elapsed_ms": round(elapsed, 3), }