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