diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py index c57342c..395721d 100644 --- a/backend/app/downloads/edit.py +++ b/backend/app/downloads/edit.py @@ -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: diff --git a/backend/tests/test_download_output.py b/backend/tests/test_download_output.py index c71ad80..d48bcad 100644 --- a/backend/tests/test_download_output.py +++ b/backend/tests/test_download_output.py @@ -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).