Converting exiftool to Python exifmodern

#!/usr/bin/env python3

"""Small wrapper around exiftool's ``-stay_open`` pipe/server mode.

exiftool can be run as a long-lived process that reads batched argument lists
on stdin and prints each batch's output followed by a ``{ready}`` sentinel
line. That avoids paying Perl startup per file, so it is the recommended way
to drive exiftool from another program.

This file is a single self-contained reference implementation of that
protocol with Python-native ergonomics:

* :class:`ExifTool` is a dataclass — pass an explicit ``executable`` path or
  let ``__post_init__`` discover one. Used as a context manager to spawn
  the underlying ``exiftool -stay_open True -@ -`` process.
* The original camelCase methods (``exifNumeric``, ``stripGPS``, ...) are
  preserved unchanged for existing callers.
* :meth:`read` and :meth:`strip` are the modern entry points: ``read``
  returns parsed JSON records (``list[dict]``), ``strip`` unifies the two
  strip helpers.
"""

from __future__ import annotations

import json
import os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, ClassVar, Iterable, Sequence

# Usage:
# with ExifTool() as e:
#    metadata = e.exifNumeric(*filenames)

@dataclass
class ExifTool:
    """Driver for an ``exiftool -stay_open`` subprocess.

    The instance carries both configuration (``executable``, ``timeout``) and,
    while inside a ``with`` block, the running :class:`subprocess.Popen`
    handle. Outside a ``with`` block ``process`` is ``None`` and the high-
    level methods will raise.
    """

    executable: str | None = "/usr/bin/exiftool"
    timeout: float = 5.0
    process: subprocess.Popen[bytes] | None = field(
        default=None, init=False, repr=False, compare=False
    )

    # Bytes, not str: stdout is read in binary mode so the sentinel scan
    # cannot be defeated by TextIOWrapper buffering between us and the pipe.
    sentinel: bytes = field(default=b"{ready}\n", repr=False)

    def __post_init__(self) -> None:
        """Init with executable, but if executable not found at default
        location, attempt to use exiftool from source package.

        Also note: our included exiftool version is patched for improved
        string output, so using system exiftool will show some numbers
        somewhat differently."""
        if not self.executable or not Path(self.executable).exists():
            executable: str | None = None
            current_dir = Path(__file__).resolve().parent
            others = [
                current_dir / "exiftool" / "exiftool",
                Path("/usr/local/bin/exiftool"),
                Path("/opt/homebrew/bin/exiftool"),
            ]
            for other in others:
                if other.exists():
                    executable = str(other)
                    break

            if not executable:
                raise FileNotFoundError("exiftool executable not found")

            self.executable = executable

    # ---- context manager --------------------------------------------------

    def __enter__(self) -> ExifTool:
        """Launch exiftool in pipe reader command accepting mode"""
        # Binary pipes throughout. The original wrapper opened the pipes in
        # text mode and then read stdout via ``os.read`` on the raw fd, which
        # is a layering bug: the TextIOWrapper can buffer bytes the raw read
        # never sees. Staying in bytes is simpler and correct.
        self.process = subprocess.Popen(
            [self.executable, "-stay_open", "True", "-@", "-"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
        )
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore[no-untyped-def]
        """Send exiftool close pipe (thus, exit) command"""
        proc = self.process
        self.process = None
        if proc is None or proc.stdin is None:
            return
        try:
            proc.stdin.write(b"-stay_open\nFalse\n")
            proc.stdin.flush()
            proc.stdin.close()
        except (BrokenPipeError, ValueError):
            pass
        # Reap the child so we don't leak a zombie. The original __exit__
        # only flushed stdin and returned, which sometimes left exiftool
        # alive past the end of the with-block.
        try:
            proc.wait(timeout=self.timeout)
        except subprocess.TimeoutExpired:
            proc.terminate()
            try:
                proc.wait(timeout=2.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait()

    # ---- introspection ----------------------------------------------------

    @property
    def version(self) -> str:
        """Return ``exiftool -ver`` output. One-shot subprocess; does not
        require an active session."""
        result = subprocess.run(
            [str(self.executable), "-ver"],
            check=False,
            capture_output=True,
            text=True,
        )
        return result.stdout.strip() or "unknown"

    # ---- pipe protocol ----------------------------------------------------

    def execute(self, *args: str) -> bytes:
        if not self.process or not self.process.stdin or not self.process.stdout:
            raise RuntimeError(
                "ExifTool process not started. Use as a context manager."
            )
        payload = "\n".join(str(a) for a in args) + "\n-execute\n"
        self.process.stdin.write(payload.encode("utf-8"))
        self.process.stdin.flush()

        buf = bytearray()
        fd: int = self.process.stdout.fileno()
        while True:
            idx = buf.find(self.sentinel)
            if idx >= 0:
                return bytes(buf[:idx])
            chunk = os.read(fd, 4096)
            if not chunk:
                raise RuntimeError(
                    "exiftool pipe closed before sentinel was seen"
                )
            buf.extend(chunk)

    # ---- flag reference ---------------------------------------------------
    # -r means recurse into any directories given as input
    # -g means nest each tag in an owner's group
    # (-G means prefix each tag as GROUP:Tag instead of nesting)
    # e.g: instead of GPSPosition, the key is Composite:GPSPosition because
    #      the value is synthensized by exiftool itself
    # -a means allow repeated tags in multiple groups
    # -j means json output, please
    # -n means numeric output, don't convert to values/strings
    # -c "%+.9f" means always convert GPS coords to float degrees of:
    #    - -121.898169444
    #    instead of the default exiftool string desc format:
    #    - 121 deg 53' 53.41" W
    # -fast1 means only read first metadata and don't scan through to end of file

    DEFAULT_READ_FLAGS: ClassVar[tuple[str, ...]] = (
        "-g",
        "-a",
        "-c",
        "%+.9f",
        "-j",
        "-fast1",
    )

    # ---- original API (preserved) -----------------------------------------

    def exif(self, extraArgs: Iterable[str] | None = None, *paths: str) -> bytes:
        safeExtraArgs = list(extraArgs) if extraArgs else []
        args = [*self.DEFAULT_READ_FLAGS, *safeExtraArgs, *paths]
        return self.execute(*args)

    def exifHuman(self, *paths: str) -> bytes:
        """Return EXIF output in Human Readable mode from one or more paths"""
        return self.exif(["-r"], *paths)

    def exifNumeric(self, *paths: str) -> bytes:
        """Return EXIF output in Numeric mode from one or more paths"""
        return self.exif(["-r", "-n"], *paths)

    def exifHumanNoRecurse(self, *paths: str) -> bytes:
        """Return EXIF output in Human Readable mode from one or more paths"""
        return self.exif([], *paths)

    def exifNumericNoRecurse(self, *paths: str) -> bytes:
        """Return EXIF output in Numeric mode from one or more paths"""
        return self.exif(["-n"], *paths)

    def stripGPS(self, *filenames: str) -> bool:
        self.execute("-gps:all=", "-j", *filenames)
        return True

    def stripTags(self, tags: list[str], *filenames: str) -> bool:
        genTags = []
        for tag in tags:
            # Specifying an argument of -[TAG]= deletes the tag
            genTags.append(f"-{tag}=")

        self.execute(*genTags, *filenames)
        return True

    # ---- modern API -------------------------------------------------------

    def read(
        self,
        *paths: str | Path,
        recurse: bool = True,
        numeric: bool = False,
        extra: Sequence[str] = (),
    ) -> list[dict[str, Any]]:
        """Read metadata from ``paths`` and return parsed JSON records.

        Each record is the dict exiftool emits in ``-j -g`` form: a
        ``SourceFile`` key plus per-group dicts (``EXIF``, ``Composite``,
        ``QuickTime``, ...). Returns ``[]`` for empty input or empty result.
        """
        if not paths:
            return []
        flags: list[str] = []
        if recurse:
            flags.append("-r")
        if numeric:
            flags.append("-n")
        flags.extend(extra)
        raw = self.exif(flags, *(str(p) for p in paths))
        text = raw.decode("utf-8")
        return json.loads(text) if text.strip() else []

    def strip(
        self,
        *paths: str | Path,
        tags: Sequence[str] | str = "gps:all",
    ) -> None:
        """Strip ``tags`` from ``paths``. Default removes all GPS tags."""
        if not paths:
            return
        tag_list = [tags] if isinstance(tags, str) else list(tags)
        # Specifying an argument of -[TAG]= deletes the tag
        args = [f"-{t}=" for t in tag_list]
        args.extend(str(p) for p in paths)
        self.execute(*args)

if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        print(f"{sys.argv[0]}: file1 file2 ... fileN")
        sys.exit(1)

    inPaths = sys.argv[1:]

    with ExifTool() as e:
        # ALSO check for UTC timestamp if exists and use instead of non-tz timestamp
        # -> Composite.GPSDateTime
        print(json.dumps(e.read(*inPaths), indent=2))