fix(downloads): probe codecs from ffprobe JSON, not its flat output

ffprobe prints codec_name BEFORE codec_type, so the line parser attributed every
name to the previous stream: a plain H.264+AAC mp4 probed as (aac, None) and the
browser-playability check re-encoded a perfectly good file on EVERY download.
Verified on localdev — the same source now stores without a re-encode pass.
This commit is contained in:
2026-07-23 22:56:42 +02:00
parent 4209f6bb24
commit 63a36db116
2 changed files with 56 additions and 15 deletions
+26 -14
View File
@@ -364,29 +364,41 @@ _BROWSER_SAFE_VCODEC = ("h264",)
_BROWSER_SAFE_ACODEC = ("aac", "mp3")
def parse_probe_streams(payload: str) -> tuple[str | None, str | None]:
"""(video_codec, audio_codec) out of ffprobe's JSON stream list, lowercased.
JSON, not the flat `default=nw=1` output: there the fields arrive in ffprobe's own struct
order, which puts `codec_name` BEFORE `codec_type`, so a line-by-line "remember the last
codec_type" parser attributes every name to the PREVIOUS stream. Measured on real downloads:
an ordinary H.264+AAC mp4 probed as `v=aac a=None` — the browser-playability check then
believed the video stream was AAC and re-encoded a perfectly fine file on every download."""
try:
streams = json.loads(payload or "{}").get("streams") or []
except ValueError:
return None, None
vcodec = acodec = None
for s in streams:
name = (s.get("codec_name") or "").lower() or None
ctype = (s.get("codec_type") or "").lower()
if ctype == "video" and vcodec is None:
vcodec = name
elif ctype == "audio" and acodec is None:
acodec = name
return vcodec, acodec
def probe_codecs(path: Path) -> tuple[str | None, str | None]:
"""Return (video_codec, audio_codec) lowercased via one ffprobe call; None for a missing
stream or if ffprobe fails (a failed probe is treated as 'leave it alone' upstream)."""
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "stream=codec_type,codec_name",
"-of", "default=nw=1", str(path)],
"-of", "json", str(path)],
capture_output=True, text=True, timeout=30, check=True,
).stdout.lower()
).stdout
except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError):
return None, None
vcodec = acodec = None
ctype = None
for line in out.splitlines():
if line.startswith("codec_type="):
ctype = line.split("=", 1)[1].strip()
elif line.startswith("codec_name="):
name = line.split("=", 1)[1].strip() or None
if ctype == "video" and vcodec is None:
vcodec = name
elif ctype == "audio" and acodec is None:
acodec = name
return vcodec, acodec
return parse_probe_streams(out)
def probe_duration(path: Path) -> int:
+30 -1
View File
@@ -5,7 +5,7 @@ so a missing filepath (an arbitrary-URL extractor) skipped the staging fallback
with "PosixPath('.') has an empty name" instead of downloading."""
import pytest
from app.downloads.edit import _concat_quote
from app.downloads.edit import _concat_quote, parse_probe_streams
from app.worker import _media_info, _produced_file
from pathlib import Path, PurePosixPath
@@ -62,6 +62,35 @@ class TestMediaInfo:
assert _media_info(info)["id"] == "vid"
class TestParseProbeStreams:
# ffprobe emits codec_name BEFORE codec_type, so the old line parser shifted every name onto
# the previous stream and an H.264+AAC mp4 came back as (aac, None) → a needless re-encode.
def test_reads_each_stream_by_its_own_type(self):
payload = '{"streams":[{"codec_name":"h264","codec_type":"video"},{"codec_name":"aac","codec_type":"audio"}]}'
assert parse_probe_streams(payload) == ("h264", "aac")
def test_audio_first_ordering_is_handled(self):
payload = '{"streams":[{"codec_name":"AAC","codec_type":"audio"},{"codec_name":"H264","codec_type":"video"}]}'
assert parse_probe_streams(payload) == ("h264", "aac")
def test_extra_streams_do_not_win_over_the_first_of_their_kind(self):
payload = (
'{"streams":[{"codec_name":"h264","codec_type":"video"},'
'{"codec_name":"vp9","codec_type":"video"},'
'{"codec_name":"mov_text","codec_type":"subtitle"},'
'{"codec_name":"opus","codec_type":"audio"}]}'
)
assert parse_probe_streams(payload) == ("h264", "opus")
def test_audio_only_and_video_only(self):
assert parse_probe_streams('{"streams":[{"codec_name":"opus","codec_type":"audio"}]}') == (None, "opus")
assert parse_probe_streams('{"streams":[{"codec_name":"av1","codec_type":"video"}]}') == ("av1", None)
def test_garbage_is_not_a_crash(self):
assert parse_probe_streams("") == (None, None)
assert parse_probe_streams("not json") == (None, None)
class TestConcatQuote:
# PurePosixPath so the escaping is asserted identically on a Windows dev box and in the
# Linux test container (a WindowsPath would eat the backslash as a separator).