siftlode/backend/app/downloads/probe.py

"""Ground-truth media probe for a produced library file — the SINGLE place both ingest paths ask
"what is actually on disk?".

`media_kind` (and the stored `vcodec`/`acodec`/`width`/`height`) must reflect the real file, not an
extractor's metadata: yt-dlp's `info` dict omits codec fields for some sources (e.g. Facebook), and
deriving the kind from those nulls misclassified a genuine H.264 video as a download-only `file`
(no in-app play). So the download worker (`pipeline`) and the local upload ingest (`upload`) both
probe the finished file here; `mediakind.derive` then turns the probed codecs into the kind.
"""

from pathlib import Path

from app.downloads import edit as editmod

# ffprobe container formats that mean "this is a still image / image sequence", NOT playable media.
# A standalone jpg/png/gif/webp/bmp reports a real video stream (codec mjpeg/png/…) with no
# `attached_pic` flag, so without this guard it would classify as media_kind='video' and get a Play
# button + a broken in-app <video> player (and inline serving on the public share). Keyed on the
# container format so an animated gif/webp also lands as a generic 'file' (no browser plays those in
# a <video> either). SVG and anything else ffprobe can't decode already fails the probe → 'file'.
_IMAGE_FORMATS = {
    "image2",
    "image2pipe",
    "png_pipe",
    "mjpeg",
    "mjpeg_pipe",
    "jpeg_pipe",
    "gif",
    "webp_pipe",
    "bmp_pipe",
    "tiff_pipe",
    "ico",
    "svg_pipe",
}


def _facts_from_probe(
    data: dict,
) -> tuple[str | None, str | None, int | None, int | None, int | None]:
    """(vcodec, acodec, width, height, duration_s) parsed from a NON-None ffprobe JSON dict. A still
    image / image-sequence container yields no vcodec (→ generic 'file'); a video stream flagged
    `attached_pic` (embedded cover art in an mp3/m4a) is skipped — it's a thumbnail, not a real video
    track, so an audio file with album art still classifies as audio."""
    fmt = data.get("format") or {}
    fmt_names = {n.strip() for n in (fmt.get("format_name") or "").split(",")}
    if fmt_names & _IMAGE_FORMATS:
        return None, None, None, None, None

    vcodec = acodec = None
    width = height = None
    for stream in data.get("streams", []):
        kind = stream.get("codec_type")
        if kind == "video" and vcodec is None:
            if (stream.get("disposition") or {}).get("attached_pic"):
                continue  # cover art, not a video track
            vcodec = stream.get("codec_name")
            width = stream.get("width")
            height = stream.get("height")
        elif kind == "audio" and acodec is None:
            acodec = stream.get("codec_name")

    duration = None
    try:
        raw = fmt.get("duration")
        if raw is not None:
            duration = int(float(raw))
    except (TypeError, ValueError):
        duration = None
    return vcodec, acodec, width, height, duration


def probe_media(path: Path) -> tuple[str | None, str | None, int | None, int | None, int | None]:
    """(vcodec, acodec, width, height, duration_s) from one ffprobe call; all None if ffprobe fails
    or the file isn't playable media (an unprobeable file — or a still image — is treated as a generic
    non-media file, so it becomes a download-only 'file' rather than a broken 'video')."""
    data = editmod.run_ffprobe_json(path, show_streams=True, show_format=True)
    if data is None:
        return None, None, None, None, None
    return _facts_from_probe(data)


def resolve_media_facts(
    path: Path, info: dict
) -> tuple[str | None, str | None, int | None, int | None, int | None]:
    """(vcodec, acodec, width, height, duration_s) for the stored download asset: PROBE the finished
    file on disk (the ground truth). yt-dlp's `info` omits codec/dimension fields for some sources
    (Facebook among them), so deriving media_kind from it there labelled a real H.264 video a
    download-only 'file'. It omits the duration too (a Facebook reel, an Instagram reel), and a NULL
    duration hides the Library row's trim action although the file plays for minutes.

    Precedence is by whether the probe RAN, not per-field: if ffprobe read the file, its codec verdict
    is authoritative — INCLUDING a deliberate None (cover-art-only audio, a still image). A per-field
    `probe or info` fallback would let `info`'s stale vcodec resurrect a video track the file doesn't
    have (yt-dlp reports a progressive `vcodec` for an audio-mode download whose video it then strips),
    mislabelling an audio download as a broken in-app 'video'. Only when ffprobe could NOT read the
    file at all do we fall back to `info` (the pre-probe behaviour, a safety net); the cosmetic
    dimensions and the duration still gap-fill from `info` in either case."""
    data = editmod.run_ffprobe_json(path, show_streams=True, show_format=True)
    info_v = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
    info_a = info.get("acodec") if info.get("acodec") not in (None, "none") else None
    info_w = int(info["width"]) if info.get("width") else None
    info_h = int(info["height"]) if info.get("height") else None
    info_d = int(info["duration"]) if info.get("duration") else None
    if data is None:
        # ffprobe could not read the file — fall back to the extractor's info entirely.
        return info_v, info_a, info_w, info_h, info_d
    pv, pa, pw, ph, pd = _facts_from_probe(data)
    # Probe ran → its codecs win outright (a None means "no such track", not "unknown"); info only
    # gap-fills the width/height/duration the probe couldn't read.
    return (
        pv,
        pa,
        pw if pw is not None else info_w,
        ph if ph is not None else info_h,
        pd if pd is not None else info_d,
    )