magnus919_agent-skills/ffmpeg/scripts/ffmpeg-preflight

#!/usr/bin/env python3
"""Check local FFmpeg/ffprobe availability and named build capabilities.

Non-mutating, dependency-free, and offline: it runs read-only inventory
probes, never touches media files, and makes no network requests.

Checks reported:
  ffmpeg / ffprobe  availability, resolved path, probe status, return code,
                    and the first diagnostic line of the version probe.
  filters / encoders / hwaccels
                    whether each inventory probe produced usable output, plus
                    a conservative count of parseable entries. With no named
                    query, a count of zero is a warning rather than proof of
                    absence.

Named capability queries:
  --filter NAME, --encoder NAME, --hwaccel NAME (each repeatable)
  Report whether each exact name appears in the corresponding inventory.
  Matching is case-sensitive and exact. A query against an unavailable, empty,
  or unparseable inventory is unknown (JSON null) and is a probe failure.
  These FFmpeg-only queries do not require ffprobe.

Output:
  Default: concise human-readable lines. --json: one JSON document with the
  shape {"ffmpeg", "ffprobe", "filters", "encoders", "hwaccels", "queries"}.
  Raw inventory text is never printed; only counts and named results are.

Exit codes:
  0  required tools and probes succeeded, and every requested capability is present
  1  required tool or probe failure (including timeout or unparseable inventory)
  2  usable inventories were parsed but at least one requested capability is absent

Human output is evidence for people; the parsed counts and query results in
--json output are the stable interface. Inventory text itself varies across
FFmpeg versions and builds and is intentionally not treated as an API.
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
from typing import Any

FIRST_LINE_LIMIT = 200
DEFAULT_TIMEOUT_SECONDS = 10.0
EMPTY_INVENTORY_MESSAGE = "no parseable entries; inventory empty or unexpected format"
NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
ENCODER_FLAG_RE = re.compile(r"[VAS][A-Za-z.]{5}")


def first_line(text: str) -> str:
    for line in text.splitlines():
        if line.strip():
            return line.strip()[:FIRST_LINE_LIMIT]
    return ""


def positive_seconds(value: str) -> float:
    try:
        seconds = float(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("timeout must be a number") from exc
    if seconds <= 0:
        raise argparse.ArgumentTypeError("timeout must be greater than zero")
    return seconds


def probe(
    binary: str,
    *args: str,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
) -> tuple[dict[str, Any], str]:
    """Run one bounded read-only probe. Return its report and parseable stdout."""
    path = shutil.which(binary)
    if not path:
        return {
            "available": False,
            "status": "missing",
            "error": f"{binary} was not found on PATH",
        }, ""
    try:
        result = subprocess.run(
            [path, *args],
            capture_output=True,
            text=True,
            check=False,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return {
            "available": False,
            "status": "timeout",
            "path": path,
            "returncode": None,
            "timeout_seconds": timeout,
            "error": f"{binary} probe timed out after {timeout:g} seconds",
        }, ""
    report = {
        "available": result.returncode == 0,
        "status": "ok" if result.returncode == 0 else "failed",
        "path": path,
        "returncode": result.returncode,
        "first_line": first_line(result.stdout + result.stderr),
    }
    return report, result.stdout


def parse_filters(stdout: str) -> set[str]:
    """Parse filter names from `ffmpeg -filters` output.

    Entry lines look like ` .. scale V->V Scale the input video size.`
    (flags column, name, media spec containing '->'). Legend and separator
    lines do not have a media spec and are skipped conservatively.
    """
    names: set[str] = set()
    for line in stdout.splitlines():
        fields = line.split()
        if len(fields) < 3 or "->" not in fields[2]:
            continue
        if NAME_RE.fullmatch(fields[1]):
            names.add(fields[1])
    return names


def parse_encoders(stdout: str) -> set[str]:
    """Parse encoder names from `ffmpeg -encoders` output.

    Entry lines look like ` V....D libx264 libx264 H.264 ...` (type flag,
    name, description). Legend lines have '=' as the second field and are
    skipped because '=' is not a valid name character.
    """
    names: set[str] = set()
    for line in stdout.splitlines():
        fields = line.split()
        if len(fields) < 3 or not ENCODER_FLAG_RE.fullmatch(fields[0]):
            continue
        if NAME_RE.fullmatch(fields[1]):
            names.add(fields[1])
    return names


def parse_hwaccels(stdout: str) -> set[str]:
    """Parse hardware acceleration methods from `ffmpeg -hwaccels` output.

    Each method is listed as a bare word on its own line; the header line
    and blank lines are skipped.
    """
    names: set[str] = set()
    for line in stdout.splitlines():
        token = line.strip()
        if NAME_RE.fullmatch(token):
            names.add(token)
    return names


PARSERS = {
    "filters": parse_filters,
    "encoders": parse_encoders,
    "hwaccels": parse_hwaccels,
}


def summarize_inventory(
    report: dict[str, Any],
    stdout: str,
    kind: str,
    *,
    require_entries: bool = False,
) -> tuple[dict[str, Any], set[str]]:
    summary: dict[str, Any] = {
        "available": report.get("available", False),
        "status": report.get("status", "failed"),
        "returncode": report.get("returncode"),
    }
    entries = PARSERS[kind](stdout) if summary["available"] and stdout else set()
    summary["entry_count"] = len(entries)
    if summary["available"] and not entries:
        if require_entries:
            summary["available"] = False
            summary["status"] = "unparseable"
            summary["error"] = EMPTY_INVENTORY_MESSAGE
        else:
            summary["status"] = "empty"
            summary["warning"] = EMPTY_INVENTORY_MESSAGE
    for key in ("error", "first_line", "timeout_seconds"):
        if key in report:
            summary[key] = report[key]
    return summary, entries


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Check local FFmpeg tools and named build capabilities before automating.",
    )
    parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
    parser.add_argument(
        "--timeout",
        type=positive_seconds,
        default=DEFAULT_TIMEOUT_SECONDS,
        metavar="SECONDS",
        help=f"bound each external probe (default: {DEFAULT_TIMEOUT_SECONDS:g} seconds)",
    )
    parser.add_argument(
        "--filter", action="append", default=[], metavar="NAME",
        help="check whether filter NAME is available (repeatable)",
    )
    parser.add_argument(
        "--encoder", action="append", default=[], metavar="NAME",
        help="check whether encoder NAME is available (repeatable)",
    )
    parser.add_argument(
        "--hwaccel", action="append", default=[], metavar="NAME",
        help="check whether hardware acceleration method NAME is available (repeatable)",
    )
    args = parser.parse_args(argv)

    requested = {
        "filters": list(dict.fromkeys(args.filter)),
        "encoders": list(dict.fromkeys(args.encoder)),
        "hwaccels": list(dict.fromkeys(args.hwaccel)),
    }
    has_named_queries = any(requested.values())

    ffmpeg_report, _ = probe("ffmpeg", "-version", timeout=args.timeout)
    ffprobe_report, _ = probe("ffprobe", "-version", timeout=args.timeout)
    ffmpeg_report["required"] = True
    ffprobe_report["required"] = not has_named_queries

    inventories: dict[str, dict[str, Any]] = {}
    entries: dict[str, set[str]] = {}
    for kind in ("filters", "encoders", "hwaccels"):
        if ffmpeg_report.get("available"):
            inventory_report, stdout = probe(
                "ffmpeg",
                "-hide_banner",
                f"-{kind}",
                timeout=args.timeout,
            )
        else:
            inventory_report, stdout = dict(ffmpeg_report), ""
        inventories[kind], entries[kind] = summarize_inventory(
            inventory_report,
            stdout,
            kind,
            require_entries=bool(requested[kind]),
        )

    query_labels = {
        "filters": "filter",
        "encoders": "encoder",
        "hwaccels": "hwaccel",
    }
    queries: dict[str, dict[str, bool | None]] = {}
    for kind, names in requested.items():
        if not names:
            continue
        inventory_is_usable = inventories[kind]["available"]
        queries[query_labels[kind]] = {
            name: name in entries[kind] if inventory_is_usable else None
            for name in names
        }

    report = {
        "ffmpeg": ffmpeg_report,
        "ffprobe": ffprobe_report,
        "filters": inventories["filters"],
        "encoders": inventories["encoders"],
        "hwaccels": inventories["hwaccels"],
        "queries": queries,
    }

    if args.json:
        print(json.dumps(report, indent=2, sort_keys=True))
    else:
        for name in ("ffmpeg", "ffprobe"):
            result = report[name]
            state = "available" if result["available"] else "unavailable"
            detail = result.get("first_line") or result.get("error", "")
            print(f"{name}: {state} - {detail}")
        for kind in ("filters", "encoders", "hwaccels"):
            summary = report[kind]
            if not summary["available"]:
                detail = summary.get("error") or f"probe exited {summary.get('returncode')}"
                print(f"{kind}: failed - {detail}")
            elif summary["entry_count"] == 0:
                print(f"{kind}: 0 entries parsed ({summary.get('warning', 'empty inventory')})")
            else:
                print(f"{kind}: {summary['entry_count']} entries parsed")
        for kind, names in queries.items():
            for name, present in names.items():
                print(f"{kind} '{name}': {'present' if present else 'absent'}")

    probe_failures = (
        not ffmpeg_report["available"]
        or (ffprobe_report["required"] and not ffprobe_report["available"])
        or any(not inventories[kind]["available"] for kind in inventories)
    )
    if probe_failures:
        return 1
    if any(not present for names in queries.values() for present in names.values()):
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())