File size: 4,836 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 | """
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),
}
|