Build a conformance gate that skips the encode when the upload is already fine

TL;DR

We're building a conformance gate in Python: probe an upload once with ffprobe, run it through ordered predicates, and get back one of three verdicts (passthrough, remux, transcode). The gate includes the keyframe-cadence check that most passthrough implementations forget, which is why they break at segmentation time.

Most upload pipelines transcode every file by reflex. But a lot of what lands in a UGC pipeline is already H.264 8-bit 4:2:0 straight off a phone, and re-encoding it costs compute and adds a second generation of loss for very little gain.

The fix isn't "stop encoding." It's classifying each file so you do the cheapest correct operation. Let's build the classifier.

Requirements: ffmpeg 8.x (7.x is fine too) and python 3.11+.

ffmpeg -version | head -1
python3 --version

1. One probe call, everything we need

Don't make three ffprobe calls. Make one and parse it.

# probe.sh
ffprobe -v error \
  -show_entries stream=index,codec_type,codec_name,profile,level,pix_fmt,width,height,avg_frame_rate,bit_rate \
  -show_entries format=duration,bit_rate,format_name \
  -of json input.mp4

Output looks like this:

{
  "streams": [
    {
      "index": 0,
      "codec_name": "h264",
      "codec_type": "video",
      "profile": "High",
      "level": 40,
      "pix_fmt": "yuv420p",
      "width": 1920,
      "height": 1080,
      "avg_frame_rate": "30000/1001"
    },
    {
      "index": 1,
      "codec_name": "aac",
      "codec_type": "audio"
    }
  ],
  "format": {
    "duration": "184.320000",
    "bit_rate": "8412160",
    "format_name": "mov,mp4,m4a,3gp,3g2,mj2"
  }
}

Wrap it:

# gate/probe.py
import json, subprocess

def probe(path: str) -> dict:
    cmd = [
        "ffprobe", "-v", "error",
        "-show_entries",
        "stream=index,codec_type,codec_name,profile,level,pix_fmt,"
        "width,height,avg_frame_rate,bit_rate",
        "-show_entries", "format=duration,bit_rate,format_name",
        "-of", "json", path,
    ]
    out = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return json.loads(out.stdout)

def first(streams, kind):
    return next((s for s in streams if s.get("codec_type") == kind), None)

💡 Tip: check=True will raise on a file ffprobe can't open at all. That's its own verdict (reject), so catch CalledProcessError at the call site rather than letting it kill your worker.

2. The predicates, in the order they bite

Order matters here, not for correctness but for debuggability. You want the failure reason to be the most fundamental thing wrong with the file.

# gate/rules.py
from dataclasses import dataclass, field

DELIVERABLE_CODECS = {"h264"}
DELIVERABLE_PIX_FMTS = {"yuv420p", "yuvj420p"}
DELIVERABLE_AUDIO = {"aac"}
MAX_LEVEL = 42          # H.264 level 4.2
MAX_BITRATE = 12_000_000  # above this, re-encode even if conformant

@dataclass
class Verdict:
    action: str                     # passthrough | remux | transcode | reject
    reasons: list = field(default_factory=list)
    rules_version: str = "2026.09.1"

def classify(info: dict, container_ok: bool, keyframes_regular: bool) -> Verdict:
    streams = info.get("streams", [])
    v = first(streams, "video")
    a = first(streams, "audio")
    reasons = []

    if v is None:
        return Verdict("reject", ["no video stream"])

    # 1. codec
    if v["codec_name"] not in DELIVERABLE_CODECS:
        reasons.append(f"codec {v['codec_name']} not deliverable")

    # 2. pixel format: 10-bit H.264 is real and browsers refuse it
    if v.get("pix_fmt") not in DELIVERABLE_PIX_FMTS:
        reasons.append(f"pix_fmt {v.get('pix_fmt')} not 8-bit 4:2:0")

    # 3. level
    if v.get("level") and v["level"] > MAX_LEVEL:
        reasons.append(f"level {v['level']} above {MAX_LEVEL}")

    # 4. bitrate ceiling: conformant but wasteful
    br = int(info["format"].get("bit_rate") or 0)
    if br > MAX_BITRATE:
        reasons.append(f"bitrate {br} above ceiling")

    # 5. keyframe cadence: the one everybody forgets
    if not keyframes_regular:
        reasons.append("irregular keyframe interval, cannot segment cleanly")

    if reasons:
        return Verdict("transcode", reasons)

    # video is clean. is audio?
    if a and a["codec_name"] not in DELIVERABLE_AUDIO:
        return Verdict("remux", [f"audio {a['codec_name']} needs re-encode"])

    if not container_ok:
        return Verdict("remux", ["container or faststart fix needed"])

    return Verdict("passthrough", [])

3. The keyframe check (do not skip this one)

This is where naive passthrough implementations die. HLS and DASH cut segments on keyframes. A phone recording with a 10-second GOP, or an irregular scene-change-driven one, cannot be cut into 4-second segments without re-encoding.

Read the first couple hundred frames and look at the gaps:

# gate/keyframes.py
import subprocess, statistics

def keyframe_intervals(path: str, n_frames: int = 300) -> list[float]:
    cmd = [
        "ffprobe", "-v", "error", "-select_streams", "v:0",
        "-show_entries", "frame=pts_time,key_frame",
        "-of", "csv=p=0",
        "-read_intervals", f"%+#{n_frames}",
        path,
    ]
    out = subprocess.run(cmd, capture_output=True, text=True, check=True)
    times = []
    for line in out.stdout.strip().splitlines():
        parts = line.split(",")
        if len(parts) < 2:
            continue
        pts, is_key = parts[0], parts[1]
        if is_key == "1" and pts not in ("N/A", ""):
            times.append(float(pts))
    return [round(b - a, 3) for a, b in zip(times, times[1:])]

def keyframes_regular(path: str, target_segment: float = 4.0,
                      tolerance: float = 0.15) -> tuple[bool, str]:
    gaps = keyframe_intervals(path)
    if len(gaps) < 3:
        return False, "too few keyframes sampled"

    median = statistics.median(gaps)
    spread = max(gaps) - min(gaps)

    if spread > median * tolerance:
        return False, f"irregular GOP (median {median}s, spread {spread}s)"
    if target_segment % median > 0.05 and median % target_segment > 0.05:
        return False, f"GOP {median}s does not divide {target_segment}s segments"
    return True, f"regular GOP of {median}s"

Run it and you get something honest:

$ python3 -c "from gate.keyframes import keyframes_regular; print(keyframes_regular('phone.mp4'))"
(False, 'irregular GOP (median 2.0s, spread 4.867s)')

$ python3 -c "from gate.keyframes import keyframes_regular; print(keyframes_regular('screencap.mp4'))"
(True, 'regular GOP of 2.0s')

⚠️ Note: -read_intervals "%+#300" only samples the beginning of the file. A source can start regular and go irregular later. For high-value assets, sample a second window mid-file. For a first version, the head is enough, and it's the difference between a check that runs in 40ms and one that decodes the whole file.

4. Wire the three actions

Each verdict maps to a command. This is the whole payoff.

# gate/actions.py
import subprocess

def run_passthrough(src: str, dst: str):
    """Source becomes the top ladder rung as-is."""
    subprocess.run(["cp", src, dst], check=True)

def run_remux(src: str, dst: str, reencode_audio: bool):
    cmd = ["ffmpeg", "-y", "-i", src, "-c:v", "copy"]
    cmd += ["-c:a", "aac", "-b:a", "128k"] if reencode_audio else ["-c:a", "copy"]
    cmd += ["-movflags", "+faststart", dst]
    subprocess.run(cmd, check=True)

def run_transcode(src: str, dst: str):
    subprocess.run([
        "ffmpeg", "-y", "-i", src,
        "-c:v", "libx264", "-preset", "medium", "-crf", "21",
        "-pix_fmt", "yuv420p",
        "-g", "96", "-keyint_min", "96", "-sc_threshold", "0",
        "-c:a", "aac", "-b:a", "128k",
        "-movflags", "+faststart", dst,
    ], check=True)

Note the transcode path forces -pix_fmt yuv420p and a fixed GOP with -sc_threshold 0. That's not decoration: it guarantees the output passes your own gate, which matters when you re-run the pipeline later.

5. Put it together

# gate/__main__.py
import sys, json
from .probe import probe
from .keyframes import keyframes_regular
from .rules import classify
from .actions import run_passthrough, run_remux, run_transcode

def process(src: str, dst: str) -> dict:
    info = probe(src)
    regular, kf_note = keyframes_regular(src)
    container_ok = src.endswith(".mp4")   # replace with a real faststart check
    verdict = classify(info, container_ok, regular)

    if verdict.action == "passthrough":
        run_passthrough(src, dst)
    elif verdict.action == "remux":
        needs_audio = any("audio" in r for r in verdict.reasons)
        run_remux(src, dst, needs_audio)
    elif verdict.action == "transcode":
        run_transcode(src, dst)
    else:
        raise ValueError(f"rejected: {verdict.reasons}")

    return {
        "source": src,
        "action": verdict.action,
        "reasons": verdict.reasons,
        "keyframes": kf_note,
        "rules_version": verdict.rules_version,
    }

if __name__ == "__main__":
    print(json.dumps(process(sys.argv[1], sys.argv[2]), indent=2))

$ python3 -m gate uploads/phone-clip.mp4 out/phone-clip.mp4
{
  "source": "uploads/phone-clip.mp4",
  "action": "remux",
  "reasons": ["audio opus needs re-encode"],
  "keyframes": "regular GOP of 2.0s",
  "rules_version": "2026.09.1"
}

6. Store the verdict on the asset

Do not just log this. Put action and rules_version on the asset record.

In six months you'll lower MAX_BITRATE or add an AV1 rung, and you'll need to answer "which assets were passed through under the old rules?" Without the stamp, the answer is "re-process everything," which is the compute you built this to avoid.

ALTER TABLE assets
  ADD COLUMN ingest_action  text NOT NULL DEFAULT 'transcode',
  ADD COLUMN ingest_rules_version text NOT NULL DEFAULT 'legacy';

What's next

Two things to be honest about before you ship this.

Passthrough saves one rung, not the whole encode. ABR still needs 720p, 480p, 360p. What you're skipping is the top rung, which is the most expensive one, on some fraction of ingests. Real saving, smaller headline.

Measure before you build. Run steps 1 to 3 across a sample of what you've already ingested, skip the actions entirely, and just count verdicts:

for f in sample/*.mp4; do python3 -m gate.classify_only "$f"; done | \
  jq -r .action | sort | uniq -c

  41 transcode
  22 remux
  17 passthrough

If passthrough and remux are a rounding error for your traffic, you've spent an afternoon and learned something. If they aren't, you just found the cheapest infra win on your backlog.

From here, the next things worth adding are a real faststart check (ffprobe the moov atom position rather than trusting the extension), a mid-file keyframe sample, and rotation-metadata handling so sideways phone video doesn't get re-encoded just to be turned.