cores/metadata/extractor.py

"""EXIF / XMP / IPTC extractor — Pillow-based, single-pass."""

from __future__ import annotations

import io
from typing import Any, Optional

from PIL import Image, UnidentifiedImageError


# --------------------------------------------------------------------------- #
# Top-level: extract everything in one pass
# --------------------------------------------------------------------------- #
def extract_all(image_bytes: bytes) -> dict:
    """Extract EXIF + GPS + XMP + IPTC + format in a single Pillow open.

    Returns a dict with keys:
        format, exif (dict), gps (dict), gps_coords (dict|None),
        xmp (bytes|None), camera_make, camera_model, software, capture_time,
        error (str|None)
    """
    result = {
        "format": None,
        "exif": {},
        "gps": {},
        "gps_coords": None,
        "xmp": None,
        "iptc": {},
        "camera_make": None,
        "camera_model": None,
        "software": None,
        "capture_time": None,
        "error": None,
    }
    if not image_bytes:
        result["error"] = "No image bytes provided"
        return result

    try:
        img = Image.open(io.BytesIO(image_bytes))
    except UnidentifiedImageError:
        result["error"] = "Pillow could not identify image format"
        return result
    except Exception as e:
        result["error"] = f"Pillow open error: {e}"
        return result

    result["format"] = img.format or "UNKNOWN"

    # EXIF
    try:
        exif_info = img._getexif()
    except Exception:
        exif_info = None

    if exif_info:
        from PIL.ExifTags import TAGS, GPSTAGS
        for tag_id, value in exif_info.items():
            tag_name = TAGS.get(tag_id, f"Tag_{tag_id}")
            if tag_name == "GPSInfo":
                for gps_tag_id, gps_value in value.items():
                    gps_tag_name = GPSTAGS.get(gps_tag_id, f"GPS_{gps_tag_id}")
                    result["gps"][gps_tag_name] = gps_value
            else:
                result["exif"][tag_name] = (
                    value if isinstance(value, (int, float, str)) else str(value)
                )
            if tag_name == "Make":
                result["camera_make"] = str(value)
            elif tag_name == "Model":
                result["camera_model"] = str(value)
            elif tag_name == "Software":
                result["software"] = str(value)
            elif tag_name in ("DateTimeOriginal", "DateTime"):
                result["capture_time"] = str(value)

    # GPS coordinates
    if result["gps"]:
        result["gps_coords"] = gps_to_coords(result["gps"])

    # XMP (raw bytes)
    try:
        result["xmp"] = img.info.get("xmp") or img.info.get("XMP")
    except Exception:
        pass

    # IPTC
    try:
        iptc = img.info.get("iptc") or img.info.get("IPTC")
        if iptc:
            result["iptc"] = {"raw_length": len(iptc) if hasattr(iptc, "__len__") else 0}
    except Exception:
        pass

    return result


# --------------------------------------------------------------------------- #
# Granular accessors (for providers that only need one piece)
# --------------------------------------------------------------------------- #
def extract_exif(image_bytes: bytes) -> dict:
    """Return only the EXIF dict (no GPS, no XMP)."""
    return extract_all(image_bytes)["exif"]


def extract_gps(image_bytes: bytes) -> Optional[dict]:
    """Return only the GPS coordinates dict, or None."""
    return extract_all(image_bytes)["gps_coords"]


# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def gps_to_coords(gps: dict) -> Optional[dict]:
    """Convert EXIF GPS dict to {lat, lon} floats.

    Handles both raw degree tuples (48, 51, 24) and Pillow IFDRational
    tuples ((48, 1), (51, 1), (24, 1)).
    """
    try:
        if not gps:
            return None

        def _to_float(val):
            """Convert a value to float, handling IFDRational tuples."""
            if isinstance(val, (int, float)):
                return float(val)
            if isinstance(val, (tuple, list)) and len(val) == 2:
                num, den = val
                return float(num) / float(den) if den else 0.0
            return float(val)

        def _convert_dms(value):
            """Convert degrees/minutes/seconds to decimal degrees."""
            if value is None:
                return None
            d, m, s = value
            return _to_float(d) + _to_float(m) / 60.0 + _to_float(s) / 3600.0

        lat = _convert_dms(gps.get("GPSLatitude"))
        lon = _convert_dms(gps.get("GPSLongitude"))
        if lat is None or lon is None:
            return None

        if gps.get("GPSLatitudeRef", "N") == "S":
            lat = -lat
        if gps.get("GPSLongitudeRef", "E") == "W":
            lon = -lon
        return {"lat": round(lat, 6), "lon": round(lon, 6)}
    except Exception:
        return None


def parse_pillow_image(image_bytes: bytes) -> tuple[Optional[Image.Image], Optional[str]]:
    """Open a Pillow image from bytes.  Returns (image, error)."""
    try:
        return Image.open(io.BytesIO(image_bytes)), None
    except Exception as e:
        return None, str(e)