From 18aed0c34b4a4b3f54eb80b12a7cebd6e0145020 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 30 Aug 2026 19:57:46 +0800 Subject: [PATCH 1/5] =?UTF-8?q?fix(ingest):=20=E4=BF=AE=E5=A4=8D=E7=AB=96?= =?UTF-8?q?=E5=B1=8F=20HEVC=20=E8=BD=AC=E7=A0=81=E6=96=B9=E5=90=91?= =?UTF-8?q?=E9=94=99=E8=AF=AF=EF=BC=88=E5=8F=8C=E9=87=8D=E6=97=8B=E8=BD=AC?= =?UTF-8?q?/ffprobe=20=E5=8F=96=E9=94=99=E8=A1=8C/=E9=80=97=E5=8F=B7?= =?UTF-8?q?=E8=BD=AC=E4=B9=89=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 竖屏 iPhone HEVC 素材转码后方向错乱（竖屏变横屏），实测定位三个根因： 1. 双重旋转：ffmpeg 默认 autorotate 已按 display matrix 物理旋转画面， 代码又显式追加 transpose=1，导致 r90 竖屏被转成 1920x1080 横屏。 旋转全部交给 autorotate，滤镜只做 scale。 2. ffprobe 取错行：探测命令同时输出 side_data=rotation 和 stream_tags=rotate 两行（r90 文件输出 270/90 两行），split("\n")[0] 取到 270，方向判定反了。改为只读 side_data 单行并 int 化。 3. scale 表达式逗号未转义：if(gt(ih,1080),...) 中的裸逗号被当作 filter 分隔，报 Invalid size。统一用 \, 转义（raw string 字面量）。 附带修复： - 竖屏按宽、横屏按高缩放（旋转后竖屏 ww、横屏 w>=h、长边 <=1920、无残留 rotation side data；校验失败 打 error 日志并降级使用原始文件、不上传 OSS，杜绝横屏文件覆盖。 测试： - test_ingest_hevc_transcode.py 重写（22 用例）：纯逻辑断言 + 真实 ffmpeg 端到端（r90/r270/横屏/4K 素材造 display matrix， 断言 1080x1920 / 1920x1080 / h264 / 无 side data / 缩略图竖版）。 - test_ingest_hevc_transcode_task.py 新增（4 用例）：任务级 mock 验证成功改写 storage_key、校验失败降级+error 日志、ffmpeg 非零 降级、非 HEVC 不转码。 --- apps/worker/worker_app/tasks/ingest.py | 272 +++++++--- tests/unit/test_ingest_hevc_transcode.py | 502 +++++++++++------- tests/unit/test_ingest_hevc_transcode_task.py | 266 ++++++++++ 3 files changed, 753 insertions(+), 287 deletions(-) create mode 100644 tests/unit/test_ingest_hevc_transcode_task.py diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 46c7d3394..48938ba15 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -150,6 +150,130 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: return metadata, success +# ── HEVC 自动转码辅助函数（模块级，便于单元测试）───────────────────────── +HEVC_CODECS = ("hevc", "h265", "hvh1") +# ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义， +# 否则逗号被当作 filter 分隔符解析，报 "No such filter: '1080)' / Invalid size"。 +# subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。 +_PORTRAIT_VF = r"scale=if(gt(iw\,1080)\,1080\,iw):-2,format=yuv420p" +_LANDSCAPE_VF = r"scale=-2:if(gt(ih\,1080)\,1080\,ih),format=yuv420p" + + +def is_hevc_codec(codec: str | None) -> bool: + """判断编码是否为 HEVC（不区分大小写）。""" + return (codec or "").lower() in HEVC_CODECS + + +def probe_rotation(path: str) -> int | None: + """ffprobe 读取视频旋转角度（display matrix side data）。 + + 返回 0/90/-90/180 等整数；无 side data 或探测失败返回 None。 + + 注意：旧实现同时请求 side_data 和 stream_tags 且取输出第一行， + iOS 文件会输出两行（如 "270\\n90"）导致取到错误值，现仅读 side_data。 + """ + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "side_data=rotation", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + first_line = (result.stdout or "").strip().split("\n")[0].strip() + if not first_line: + return None + return int(float(first_line)) + except (subprocess.TimeoutExpired, ValueError, OSError): + return None + + +def is_portrait_rotation(rotation: int | None) -> bool: + """rotation side data 为 ±90/270 时表示竖屏拍摄。""" + return rotation in (90, 270, -90) + + +def build_transcode_vf(is_portrait: bool) -> str: + """构建转码视频滤镜。 + + 依赖 ffmpeg 内置 autorotate（默认开启）按 display matrix 物理旋转画面， + 输出自动剥离 rotation side data；这里只做"只缩不放"的 1080p 等比缩放： + - 竖屏（旋转后 w tuple[int | None, int | None]: + """ffprobe 读取视频宽高（像素维度）。""" + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height", + "-of", + "csv=p=0:s=x", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + text = (result.stdout or "").strip().split("\n")[0].strip() + width_str, height_str = text.split("x") + return int(width_str), int(height_str) + except (subprocess.TimeoutExpired, ValueError, OSError): + return None, None + + +def validate_transcode_output( + output_path: str, + expected_portrait: bool, + max_long_edge: int = 1920, +) -> bool: + """校验转码产物方向与维度。 + + - 竖屏源：产物必须 height > width，且仍有 rotation side data 视为失败 + （播放器会二次旋转成横屏） + - 横屏源：产物必须 width >= height + - 长边不得超过 max_long_edge（只缩不放） + 校验失败时调用方应降级使用原始文件，不允许产出方向错误的文件覆盖。 + """ + width, height = probe_dimensions(output_path) + if not width or not height: + return False + if expected_portrait and height <= width: + return False + if not expected_portrait and width < height: + return False + if max(width, height) > max_long_edge: + return False + # 产物仍带 rotation side data 说明方向没有物理固化，播放器会再次旋转 + if probe_rotation(output_path) is not None: + return False + return True + + @celery_app.task(name="worker.ingest_asset") def ingest_asset(job_id: str) -> dict: """ @@ -245,15 +369,13 @@ def ingest_asset(job_id: str) -> dict: # 浏览器 WebCodecs 硬件解码 HEVC 输出黑帧，上传时自动转码 # 失败时降级使用原始文件，不阻塞上传流程 if media_type == "video" and local_file and local_file.exists(): - codec = (metadata.get("codec") or "").lower() - if codec in ("hevc", "h265", "hvh1"): + if is_hevc_codec(metadata.get("codec")): logger.info( "检测到 HEVC 编码 (codec=%s)，启动转码: job_id=%s", - codec, + metadata.get("codec"), job_id, ) _tc_tmp = None - _needs_rotation = False # ── Step 1: 磁盘空间检查（独立 try/except，失败仍尝试转码）── try: @@ -264,50 +386,16 @@ def ingest_asset(job_id: str) -> dict: except Exception as _disk_err: logger.warning("磁盘检查失败，仍尝试转码: job_id=%s err=%s", job_id, _disk_err) - # ── Step 2: ffprobe 旋转检测（独立 try/except，失败不阻塞转码）── - try: - _probe_cmd = [ - "ffprobe", - "-v", - "error", - "-select_streams", - "v:0", - "-show_entries", - "side_data=rotation", - "-show_entries", - "stream_tags=rotate", - "-of", - "default=noprint_wrappers=1:nokey=1", - str(local_file), - ] - _probe_result = subprocess.run( - _probe_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - timeout=60, # 大文件在容器 overlay 文件系统上解析可能较慢 - ) - _rotation_str = (_probe_result.stdout or "").strip().split("\n")[0] - if _rotation_str in ("90", "270", "-90"): - _needs_rotation = True - logger.info( - "检测到竖屏视频 (rotation=%s)，将物理旋转画面: job_id=%s", - _rotation_str, - job_id, - ) - except subprocess.TimeoutExpired: - logger.warning( - "ffprobe 旋转检测超时(60s)，跳过旋转继续转码: job_id=%s", + # ── Step 2: ffprobe 旋转检测（失败按横屏处理，后续方向校验兜底）── + _rotation = probe_rotation(str(local_file)) + _is_portrait = is_portrait_rotation(_rotation) + if _rotation is not None: + logger.info( + "检测到视频 rotation=%s (portrait=%s): job_id=%s", + _rotation, + _is_portrait, job_id, ) - _needs_rotation = False - except Exception as _probe_err: - logger.warning( - "ffprobe 旋转检测异常，跳过旋转继续转码: job_id=%s err=%s", - job_id, - _probe_err, - ) - _needs_rotation = False # ── Step 3: ffmpeg 转码（独立 try/except）── try: @@ -315,11 +403,11 @@ def ingest_asset(job_id: str) -> dict: _tc_tmp = Path(_tc_tmp_file.name) _tc_tmp_file.close() # 关闭文件描述符，ffmpeg 会自己打开 - # 构建 video filter：竖屏先旋转再缩放 - if _needs_rotation: - _vf = "transpose=1,scale='if(gt(ih,1080),-2,iw)':'if(gt(ih,1080),1080,ih)'" - else: - _vf = "scale='if(gt(ih,1080),-2,iw)':'if(gt(ih,1080),1080,ih)'" + # 旋转交给 ffmpeg 内置 autorotate（按 display matrix 物理旋转， + # 输出自动剥离 side data）；滤镜只做 1080p 等比"只缩不放"。 + # 注意不能再加 transpose：旧逻辑 autorotate + transpose 双重旋转， + # 竖屏被转成横屏；scale 表达式内逗号必须 \, 转义（见 build_transcode_vf）。 + _vf = build_transcode_vf(_is_portrait) _cmd = [ "ffmpeg", @@ -333,7 +421,7 @@ def ingest_asset(job_id: str) -> dict: "-crf", "18", "-vf", - _vf + ",format=yuv420p", + _vf, "-colorspace", "bt709", "-color_primaries", @@ -344,21 +432,14 @@ def ingest_asset(job_id: str) -> dict: "yuv420p", "-level", "4.2", + "-c:a", + "aac", + "-b:a", + "128k", + "-movflags", + "+faststart", + str(_tc_tmp), ] - # 竖屏视频：清除旋转元数据 - if _needs_rotation: - _cmd.extend(["-metadata:s:v:0", "rotate=0"]) - _cmd.extend( - [ - "-c:a", - "aac", - "-b:a", - "128k", - "-movflags", - "+faststart", - str(_tc_tmp), - ] - ) _proc = subprocess.run( _cmd, stdout=subprocess.DEVNULL, @@ -367,31 +448,46 @@ def ingest_asset(job_id: str) -> dict: timeout=900, ) if _proc.returncode == 0 and _tc_tmp.exists() and _tc_tmp.stat().st_size > 0: - from video_processing.oss_helpers import upload_to_oss - - _p = Path(job.storage_key) - _new_key = str(_p.parent / (_p.stem + "_h264" + _p.suffix)) - _url = upload_to_oss(_tc_tmp, _new_key) - if _url: - # 先提取元数据，确认成功后再更新 storage_key（避免脏数据） - _new_metadata, _new_extract_success = extract_media_metadata( - str(_tc_tmp), - media_type, - ) - if _new_extract_success: - job.storage_key = _new_key - metadata = _new_metadata - extract_success = _new_extract_success - logger.info( - "HEVC→H.264 转码完成: job_id=%s key=%s", + # ── Step 4: 方向/维度校验，不符则降级，杜绝横屏文件覆盖 ── + if not validate_transcode_output(str(_tc_tmp), _is_portrait): + _w, _h = probe_dimensions(str(_tc_tmp)) + _rot = probe_rotation(str(_tc_tmp)) + logger.error( + "转码产物方向/维度校验失败，降级使用原始文件: " + "job_id=%s source_rotation=%s portrait=%s out=%sx%s out_rotation=%s", job_id, - _new_key[:80], + _rotation, + _is_portrait, + _w, + _h, + _rot, ) else: - logger.warning( - "转码文件上传 OSS 失败，使用原始文件: job_id=%s", - job_id, - ) + from video_processing.oss_helpers import upload_to_oss + + _p = Path(job.storage_key) + _new_key = str(_p.parent / (_p.stem + "_h264" + _p.suffix)) + _url = upload_to_oss(_tc_tmp, _new_key) + if _url: + # 先提取元数据，确认成功后再更新 storage_key（避免脏数据） + _new_metadata, _new_extract_success = extract_media_metadata( + str(_tc_tmp), + media_type, + ) + if _new_extract_success: + job.storage_key = _new_key + metadata = _new_metadata + extract_success = _new_extract_success + logger.info( + "HEVC→H.264 转码完成: job_id=%s key=%s", + job_id, + _new_key[:80], + ) + else: + logger.warning( + "转码文件上传 OSS 失败，使用原始文件: job_id=%s", + job_id, + ) else: _tail = _proc.stderr[-300:] if _proc.stderr else "" logger.warning( diff --git a/tests/unit/test_ingest_hevc_transcode.py b/tests/unit/test_ingest_hevc_transcode.py index 34daa9741..520ce0861 100644 --- a/tests/unit/test_ingest_hevc_transcode.py +++ b/tests/unit/test_ingest_hevc_transcode.py @@ -1,255 +1,359 @@ """HEVC 自动转码逻辑单元测试 (ingest.py) -测试覆盖： -- HEVC 编码检测逻辑 -- 转码后文件命名规则 -- 元数据提取失败时的脏数据防护 -- FFmpeg 超时/错误降级策略 -- 安全修复（tempfile、subprocess） -- Scale filter 逻辑 +测试覆盖（全部调用生产代码真实函数）： +- HEVC 编码检测（is_hevc_codec） +- ffprobe 旋转探测（probe_rotation）：单行 side_data 读取，不再取错 stream_tags 行 +- 竖屏判定（is_portrait_rotation） +- 转码滤镜构建（build_transcode_vf）：逗号 \\, 转义、无 transpose（避免与 + ffmpeg 内置 autorotate 双重旋转）、竖屏按宽/横屏按高的 1080p 只缩不放 +- 转码产物方向/维度校验（validate_transcode_output） +- ffmpeg 端到端（有 ffmpeg 时）：竖屏 r90/r270 → 1080x1920 h264 无 side data； + 横屏 → 1920x1080；缩略图方向为竖版 """ from __future__ import annotations +import shutil import subprocess +import sys from pathlib import Path -from unittest.mock import MagicMock, patch import pytest +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) -class TestHEVCAutoTranscode: - """测试 ingest_asset 中的 HEVC 自动转码逻辑""" +from worker_app.tasks.ingest import ( # noqa: E402 + HEVC_CODECS, + build_transcode_vf, + is_hevc_codec, + is_portrait_rotation, + probe_dimensions, + probe_rotation, + validate_transcode_output, +) - def test_hevc_detection_keywords(self): - """验证 HEVC 编码的所有关键词""" - hevc_keywords = ("hevc", "h265", "hvh1") +# ── 纯逻辑测试（不依赖 ffmpeg）────────────────────────────────────────────── - assert "hevc" in hevc_keywords - assert "h265" in hevc_keywords - assert "hvh1" in hevc_keywords - assert "h264" not in hevc_keywords - assert "avc1" not in hevc_keywords - def test_h264_not_detected_as_hevc(self): - """H.264 视频不应触发转码""" - codec = "h264" - hevc_keywords = ("hevc", "h265", "hvh1") - assert codec not in hevc_keywords, "H.264 不应触发转码" +class TestHEVCCodecDetection: + def test_hevc_keywords_detected(self): + for codec in ("hevc", "h265", "hvh1", "HEVC", "H265", "Hevc", "HVH1"): + assert is_hevc_codec(codec), f"{codec} 应该被识别为 HEVC" - def test_transcode_storage_key_naming(self): - """验证转码后文件命名规则""" - original_key = "uploads/video_123/test.mp4" - p = Path(original_key) - new_key = str(p.parent / (p.stem + "_h264" + p.suffix)) + def test_non_hevc_codecs_not_detected(self): + for codec in ("h264", "avc1", "vp9", "av1", "mpeg4", "", None): + assert not is_hevc_codec(codec), f"{codec} 不应被识别为 HEVC" - assert new_key == "uploads/video_123/test_h264.mp4" + def test_hevc_keywords_constant(self): + assert HEVC_CODECS == ("hevc", "h265", "hvh1") - def test_transcode_storage_key_naming_complex_path(self): - """验证复杂路径的命名规则""" - original_key = "uploads/2026/08/20/abc123/video_4k.mov" - p = Path(original_key) - new_key = str(p.parent / (p.stem + "_h264" + p.suffix)) - assert new_key == "uploads/2026/08/20/abc123/video_4k_h264.mov" +class TestPortraitRotation: + def test_portrait_rotations(self): + for rotation in (90, 270, -90): + assert is_portrait_rotation(rotation), f"rotation={rotation} 应为竖屏" - def test_metadata_failure_no_dirty_data(self): - """验证元数据提取失败时不更新 storage_key（避免脏数据） + def test_non_portrait_rotations(self): + for rotation in (0, 180, -180, None): + assert not is_portrait_rotation(rotation), f"rotation={rotation} 不应判定为竖屏" - 这是 AI Code Review 发现的 BUG 修复： - - 旧逻辑：先更新 storage_key，再提取元数据 → 可能产生脏数据 - - 新逻辑：先提取元数据，确认成功后再更新 storage_key - """ - original_storage_key = "uploads/test/video.mp4" - new_storage_key = "uploads/test/video_h264.mp4" - # 初始状态 - job_storage_key = original_storage_key - metadata = {"codec": "hevc", "width": 3840, "height": 2160} +class TestBuildTranscodeVF: + def test_comma_escaped_with_backslash(self): + r"""scale 表达式内的逗号必须 \, 转义（否则报 Invalid size / No such filter）。""" + for vf in (build_transcode_vf(True), build_transcode_vf(False)): + assert "\\," in vf + # 不应存在未转义的裸逗号（filter 分隔）出现在 if 表达式内 + assert "gt(ih,1080)" not in vf + assert "gt(iw,1080)" not in vf - # 模拟转码成功 - transcode_success = True + def test_no_transpose_filter(self): + """不能显式 transpose：ffmpeg autorotate 已按 side data 物理旋转， + 再加 transpose 会双重旋转把竖屏转成横屏。""" + assert "transpose" not in build_transcode_vf(True) + assert "transpose" not in build_transcode_vf(False) - # 模拟元数据提取失败 - new_metadata = {} - new_extract_success = False + def test_portrait_scales_by_width(self): + """竖屏（旋转后 w bool: + if not FFMPEG: + return False + out = subprocess.run([FFMPEG, "-hide_banner", "-encoders"], capture_output=True, text=True).stdout + return "libx265" in out + + +pytestmark = pytest.mark.skipif( + not (FFMPEG and FFPROBE and _has_x265()), + reason="ffmpeg/ffprobe/libx265 不可用，跳过端到端转码测试", +) + +FONT = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" + + +@pytest.fixture +def hevc_sources(tmp_path): + """构造带方向标记的 HEVC 测试素材（横屏存储 + rotation side data，模拟 iPhone）。""" + base = tmp_path / "base_landscape_h264.mp4" + draw = ( + "drawbox=x=0:y=0:w=1920:h=200:color=red:t=fill," + f"drawtext=fontfile={FONT}:text='TOP':fontsize=160:fontcolor=black:" + "x=(w-tw)/2:y=30" + ) + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-f", + "lavfi", + "-i", + "color=c=white:s=1920x1080:d=2:r=30", + "-vf", + draw, + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + str(base), + ], + check=True, + ) + base_hevc = tmp_path / "base_landscape_hevc.mp4" + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(base), + "-c:v", + "libx265", + "-tag:v", + "hvc1", + "-pix_fmt", + "yuv420p", + "-an", + str(base_hevc), + ], + check=True, + ) + + def tag_rotate(src: Path, rotation: int) -> Path: + out = tmp_path / f"portrait_r{rotation}_hevc.mp4" + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(src), + "-c", + "copy", + "-metadata:s:v:0", + f"rotate={rotation}", + str(out), + ], + check=True, ) + return out - result = mock_subprocess.return_value - transcode_success = result.returncode == 0 + yield { + "portrait_r90": tag_rotate(base_hevc, 90), + "portrait_r270": tag_rotate(base_hevc, 270), + "landscape": base_hevc, + "tmp_path": tmp_path, + } - assert not transcode_success, "FFmpeg 返回非零退出码应该导致转码失败" - def test_scale_filter_logic_4k_video(self): - """验证 4K 视频会被缩放到 1080p""" - ih = 2160 - should_scale = ih > 1080 - assert should_scale, "4K 视频应该被缩放" - - def test_scale_filter_logic_1080p_video(self): - """验证 1080p 视频不会被缩放""" - ih = 1080 - should_scale = ih > 1080 - assert not should_scale, "1080p 视频不应该被缩放" - - def test_scale_filter_logic_720p_video(self): - """验证 720p 视频不会被缩放""" - ih = 720 - should_scale = ih > 1080 - assert not should_scale, "720p 视频不应该被缩放" - - def test_tempfile_security_fix(self): - """验证使用 NamedTemporaryFile 替代 mktemp（安全修复） - - AI Code Review 发现的安全漏洞： - - tempfile.mktemp 存在 TOCTOU 竞态条件 - - 应该使用 NamedTemporaryFile(delete=False) - """ - import tempfile - - with patch("tempfile.NamedTemporaryFile") as mock_ntf: - mock_file = MagicMock() - mock_file.name = "/tmp/test_h264.mp4" - mock_ntf.return_value = mock_file - - # 新代码的调用方式 - _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4") - _tc_tmp = Path(_tc_tmp_file.name) - _tc_tmp_file.close() - - # 验证使用了 NamedTemporaryFile - mock_ntf.assert_called_once_with(delete=False, suffix="_h264.mp4") - - def test_subprocess_output_handling(self): - """验证 subprocess 输出处理（避免内存溢出） - - AI Code Review 发现的稳定性风险： - - capture_output=True 会将所有输出加载到内存 - - 应该使用 stdout=DEVNULL, stderr=PIPE - """ - import subprocess as sp - - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) - - # 新代码的调用方式 - sp.run( - ["ffmpeg", "-i", "input.mp4", "output.mp4"], - stdout=sp.DEVNULL, - stderr=sp.PIPE, - text=True, - timeout=300, - ) - - # 验证使用了 stdout=DEVNULL, stderr=PIPE - call_kwargs = mock_run.call_args[1] - assert call_kwargs.get("stdout") == sp.DEVNULL - assert call_kwargs.get("stderr") == sp.PIPE - assert call_kwargs.get("timeout") == 300 - - def test_ffmpeg_command_parameters(self): - """验证 FFmpeg 命令参数正确性""" - expected_params = [ +def _transcode_like_production(src: Path, dst: Path) -> tuple[bool, int | None]: + """按生产代码相同方式执行转码，返回 (is_portrait, rotation)。""" + rotation = probe_rotation(str(src)) + is_portrait = is_portrait_rotation(rotation) + vf = build_transcode_vf(is_portrait) + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(src), "-c:v", "libx264", "-preset", "fast", "-crf", "18", + "-vf", + vf, "-pix_fmt", "yuv420p", - "-c:a", - "aac", - "-b:a", - "128k", + "-an", "-movflags", "+faststart", - ] + str(dst), + ], + check=True, + ) + return is_portrait, rotation - # 验证所有关键参数都在命令中 - cmd = ["ffmpeg", "-y", "-i", "input.mp4"] - cmd.extend(expected_params) - cmd.append("output.mp4") - assert "-c:v" in cmd - assert "libx264" in cmd - assert "-crf" in cmd - assert "18" in cmd - assert "-pix_fmt" in cmd - assert "yuv420p" in cmd - assert "-movflags" in cmd - assert "+faststart" in cmd +class TestTranscodeEndToEnd: + def test_portrait_r90_hevc(self, hevc_sources): + """竖屏 HEVC(rotation=90) 转码后：h264、1080x1920（h>w）、无 rotation side data。""" + src = hevc_sources["portrait_r90"] + dst = hevc_sources["tmp_path"] / "out_r90.mp4" + is_portrait, rotation = _transcode_like_production(src, dst) + assert rotation == 90 + assert is_portrait is True - def test_hevc_codec_case_insensitive(self): - """验证 HEVC 检测不区分大小写""" - test_cases = ["hevc", "HEVC", "Hevc", "h265", "H265", "hvh1", "HVH1"] - hevc_keywords = ("hevc", "h265", "hvh1") + width, height = probe_dimensions(str(dst)) + assert (width, height) == (1080, 1920) + assert height > width + assert probe_rotation(str(dst)) is None + assert validate_transcode_output(str(dst), True) is True - for codec in test_cases: - assert codec.lower() in hevc_keywords, f"{codec} 应该被检测为 HEVC" + codec = subprocess.run( + [ + FFPROBE, + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=codec_name", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(dst), + ], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + assert codec == "h264" - def test_non_hevc_codecs(self): - """验证非 HEVC 编码不会触发转码""" - non_hevc_codecs = ["h264", "avc1", "vp9", "av1", "mpeg4", ""] - hevc_keywords = ("hevc", "h265", "hvh1") + def test_portrait_r270_hevc(self, hevc_sources): + """竖屏 HEVC(rotation=270→side_data 显示 -90) 同样转为 1080x1920。""" + src = hevc_sources["portrait_r270"] + dst = hevc_sources["tmp_path"] / "out_r270.mp4" + is_portrait, rotation = _transcode_like_production(src, dst) + assert rotation == -90 # ffprobe side_data 规范化为 -90 + assert is_portrait is True - for codec in non_hevc_codecs: - assert codec.lower() not in hevc_keywords, f"{codec} 不应触发转码" + width, height = probe_dimensions(str(dst)) + assert (width, height) == (1080, 1920) + assert height > width + assert probe_rotation(str(dst)) is None + assert validate_transcode_output(str(dst), True) is True + + def test_landscape_hevc(self, hevc_sources): + """横屏 HEVC 转码后：h264、1920x1080（w>h）、无 rotation side data。""" + src = hevc_sources["landscape"] + dst = hevc_sources["tmp_path"] / "out_land.mp4" + is_portrait, rotation = _transcode_like_production(src, dst) + assert rotation is None + assert is_portrait is False + + width, height = probe_dimensions(str(dst)) + assert (width, height) == (1920, 1080) + assert width > height + assert probe_rotation(str(dst)) is None + assert validate_transcode_output(str(dst), False) is True + + def test_thumbnail_portrait_direction(self, hevc_sources): + """缩略图（thumbnail_generator 同款 ffmpeg 抽帧，依赖 autorotate）竖屏源→竖版图。""" + src = hevc_sources["portrait_r90"] + thumb = hevc_sources["tmp_path"] / "thumb.jpg" + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-ss", + "0.1", + "-i", + str(src), + "-vframes", + "1", + "-vf", + "scale=640:-1:force_original_aspect_ratio=decrease,format=yuvj420p", + "-q:v", + "2", + str(thumb), + ], + check=True, + ) + width, height = probe_dimensions(str(thumb)) + assert height > width, f"竖屏缩略图应为竖版，实际 {width}x{height}" + + def test_probe_rotation_reads_side_data_not_tag_line(self, hevc_sources): + """回归：旧实现同时请求 side_data+stream_tags 输出两行且取第一行， + r90 文件会取到 "270"；新实现只读 side_data，r90→90、r270→-90。""" + assert probe_rotation(str(hevc_sources["portrait_r90"])) == 90 + assert probe_rotation(str(hevc_sources["portrait_r270"])) == -90 + assert probe_rotation(str(hevc_sources["landscape"])) is None diff --git a/tests/unit/test_ingest_hevc_transcode_task.py b/tests/unit/test_ingest_hevc_transcode_task.py new file mode 100644 index 000000000..cfb82b7e8 --- /dev/null +++ b/tests/unit/test_ingest_hevc_transcode_task.py @@ -0,0 +1,266 @@ +"""ingest_asset 任务中 HEVC 转码主流程的任务级单元测试。 + +通过 mock subprocess / repository / OSS，验证： +- 转码成功 + 方向校验通过 → storage_key 改写为 *_h264 +- 方向校验失败（竖屏转出横屏）→ 降级原文件，storage_key 不变，error 日志 +- ffmpeg 非零退出 → 降级原文件 +- 非 HEVC 编码 → 不触发转码 +""" + +from __future__ import annotations + +import sys +import tempfile +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +# 在 import worker_app 模块前 mock 掉数据库连接和 celery（同 test_ingest_validation.py） +_mock_db_module = MagicMock() +_mock_db_module.SessionLocal = MagicMock() +sys.modules["worker_app.db"] = _mock_db_module +sys.modules["worker_app.core.config"] = MagicMock() + +_mock_celery_module = MagicMock() + + +def _passthrough_decorator(*args, **kwargs): + if len(args) == 1 and callable(args[0]): + return args[0] + return lambda f: f + + +_mock_celery_module.celery_app.task = MagicMock(side_effect=_passthrough_decorator) +sys.modules["worker_app.celery_app"] = _mock_celery_module + +# mock video_processing 子模块（主流程会 import 它们） +_oss_helpers_mock = MagicMock() +_thumbnail_mock = MagicMock() +sys.modules["video_processing.oss_helpers"] = _oss_helpers_mock +sys.modules["video_processing.thumbnail_generator"] = _thumbnail_mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) + +import pytest # noqa: E402 +from worker_app.tasks import ingest as ingest_mod # noqa: E402 + + +class _FakeJobRepo: + def __init__(self, db): + self.initial_job = SimpleNamespace( + id="job-1", + project_id="proj-1", + library_id="lib-1", + storage_key="uploads/proj/IMG_2281.MOV", + file_hash="hash-1", + status=None, + error_message=None, + result_asset_id=None, + updated_at=None, + ) + self.updated_job = None + + def get(self, job_id): + return self.initial_job + + def update(self, job): + # 生产代码在同一 job 对象上原地修改属性后传入 update； + # 这里捕获引用，断言时读最终状态 + self.updated_job = job + + @property + def final_job(self): + return self.updated_job or self.initial_job + + +class _FakeAssetRepo: + def __init__(self, db): + self.created = None + + def create(self, asset): + self.created = asset + + +def _video_metadata(codec="hevc"): + return { + "codec": codec, + "width": 1920, + "height": 1080, + "duration": 10.0, + "size_bytes": 5 * 1024 * 1024, + } + + +@pytest.fixture +def task_env(tmp_path): + """统一构造 ingest_asset 主流程的 mock 环境。返回控制句柄。 + + 测试中用 mocks = _start_patches(control) 启动，断言必须用 + mocks["upload"] 等 start() 返回的 mock；不能在 stop() 后读模块 + 属性（stop 后属性恢复为原 auto-mock，调用记录为 0）。 + """ + local_file = tmp_path / "local_hevc.MOV" + local_file.write_bytes(b"fake-hevc-source") + tc_out = tmp_path / "transcode_out_h264.mp4" + + control = { + "rotation_source": 90, # 源文件 rotation；None=横屏无 side data + "transcode_rc": 0, + "transcode_produces_file": True, + "validate_ok": True, + "upload_url": "https://oss.example.com/x_h264.MOV", + "codec": "hevc", + "tc_out": tc_out, + "local_file": local_file, + } + + def fake_probe_rotation(path): + if Path(path).name == tc_out.name: + return None # 产物无 side data + return control["rotation_source"] + + def fake_probe_dimensions(path): + if Path(path).name == tc_out.name: + return (1080, 1920) if control["validate_ok"] else (1920, 1080) + return (1920, 1080) + + control["subprocess_calls"] = [] + + def fake_subprocess_run(cmd, **kwargs): + control["subprocess_calls"].append(list(cmd[:3])) + if cmd and cmd[0] == "ffmpeg" and "libx264" in cmd: + if control["transcode_rc"] != 0: + return SimpleNamespace(returncode=control["transcode_rc"], stderr="boom") + if control["transcode_produces_file"]: + Path(cmd[-1]).write_bytes(b"fake-h264-output") + return SimpleNamespace(returncode=0, stderr="") + return SimpleNamespace(returncode=0, stdout="", stderr="") + + def fake_ntf(*args, **kwargs): + mock_file = MagicMock() + mock_file.name = str(tc_out) if kwargs.get("suffix") == "_h264.mp4" else str(local_file) + mock_file.close = MagicMock() + # with ... as tmp: 让 __enter__ 返回自身，tmp.name 才是上面设置的路径 + mock_file.__enter__.return_value = mock_file + mock_file.__exit__.return_value = False + return mock_file + + job_repo = _FakeJobRepo(db=None) + asset_repo = _FakeAssetRepo(db=None) + + control["patchers"] = { + "session": patch.object(ingest_mod, "SessionLocal", return_value=MagicMock()), + "job_repo": patch.object(ingest_mod, "SQLAlchemyIngestJobRepository", return_value=job_repo), + "asset_repo": patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo), + "download": patch.object(ingest_mod, "download_asset", return_value=True), + "upload": patch.object( + sys.modules["video_processing.oss_helpers"], + "upload_to_oss", + return_value=control["upload_url"], + ), + "metadata": patch.object( + ingest_mod, + "extract_media_metadata", + side_effect=lambda path, mt: ( + (_video_metadata("h264"), True) + if Path(path).name == tc_out.name + else (_video_metadata(control["codec"]), True) + ), + ), + "rotation": patch.object(ingest_mod, "probe_rotation", side_effect=fake_probe_rotation), + "dimensions": patch.object(ingest_mod, "probe_dimensions", side_effect=fake_probe_dimensions), + "validate": patch.object( + ingest_mod, + "validate_transcode_output", + side_effect=lambda p, portrait: control["validate_ok"], + ), + "subprocess": patch.object(ingest_mod.subprocess, "run", side_effect=fake_subprocess_run), + "ntf": patch.object(tempfile, "NamedTemporaryFile", side_effect=fake_ntf), + # 缩略图生成跳过 + "thumb": patch( + "video_processing.thumbnail_generator.extract_first_frame", + side_effect=RuntimeError("skip thumb"), + ), + } + + control["job_repo"] = job_repo + control["asset_repo"] = asset_repo + return control + + +def _start_patches(control): + """启动全部 patcher，返回具名 mock dict（调用记录都在这些 mock 上）。""" + return {name: p.start() for name, p in control["patchers"].items()} + + +def _stop_patches(control): + for p in control["patchers"].values(): + p.stop() + + +class TestIngestHEVCTranscodeFlow: + def test_success_rewrites_storage_key(self, task_env): + """竖屏 HEVC 转码+校验通过 → storage_key 改写为 *_h264.MOV，asset READY 入库。""" + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281_h264.MOV" + assert task_env["asset_repo"].created is not None + # 转码产物上传 OSS 恰好一次，且上传的是 *_h264.MOV 新 key + mocks["upload"].assert_called_once() + uploaded_path, uploaded_key = mocks["upload"].call_args.args + assert uploaded_key == "uploads/proj/IMG_2281_h264.MOV" + assert str(uploaded_path).endswith("_h264.mp4") + + def test_validation_failure_keeps_original_file(self, task_env): + """竖屏转出横屏（校验失败）→ 降级原文件，storage_key 不变，打 error 日志。""" + task_env["validate_ok"] = False + mocks = _start_patches(task_env) + error_mock = MagicMock() + try: + with patch.object(ingest_mod.logger, "error", error_mock): + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + assert error_mock.called + assert "方向/维度校验失败" in error_mock.call_args[0][0] + # 校验失败：转码产物不得上传 OSS，杜绝横屏文件覆盖 + mocks["upload"].assert_not_called() + + def test_ffmpeg_nonzero_keeps_original(self, task_env): + """ffmpeg 返回非零 → 降级原文件，storage_key 不变。""" + task_env["transcode_rc"] = 1 + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + mocks["upload"].assert_not_called() + + def test_non_hevc_no_transcode(self, task_env): + """非 HEVC 编码（h264）→ 不触发 ffmpeg 转码。""" + task_env["codec"] = "h264" + task_env["rotation_source"] = None + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + mocks["upload"].assert_not_called() + # 所有 subprocess 调用都不应是 ffmpeg 转码 + for call in mocks["subprocess"].call_args_list: + cmd = call.args[0] if call.args else call.kwargs.get("cmd", []) + assert "libx264" not in cmd -- 2.54.0 From 5e3ee08d26d285b65ee8456210f0644f476b813a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 30 Aug 2026 20:24:32 +0800 Subject: [PATCH 2/5] =?UTF-8?q?test(ingest):=20=E7=AB=AF=E5=88=B0=E7=AB=AF?= =?UTF-8?q?=E7=B4=A0=E6=9D=90=E5=85=BC=E5=AE=B9=20ffmpeg=207.x=EF=BC=88-di?= =?UTF-8?q?splay=5Frotation=20=E4=BC=98=E5=85=88=EF=BC=8C=E8=80=81?= =?UTF-8?q?=E5=BC=8F=20rotate=20metadata=20=E5=9B=9E=E9=80=80=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI 环境为 ffmpeg 7.1.5（Debian 13），本地为 4.4.2。7.x 的 mp4 muxer 在 stream copy 时不再把老式 -metadata:s:v:0 rotate=N 写入 display matrix，导致竖屏测试素材不带旋转信息、probe_rotation 返回 None。 tag_rotate 改为双路径：优先 -display_rotation（5.1+ 支持），失败回退 -metadata rotate（4.x）；造完用 probe_rotation 自检，均失败则返回 None，4 个端到端用例对 None 源 pytest.skip，不再误报失败。 --- tests/unit/test_ingest_hevc_transcode.py | 56 ++++++++++++++++++++---- 1 file changed, 48 insertions(+), 8 deletions(-) diff --git a/tests/unit/test_ingest_hevc_transcode.py b/tests/unit/test_ingest_hevc_transcode.py index 520ce0861..3839081fe 100644 --- a/tests/unit/test_ingest_hevc_transcode.py +++ b/tests/unit/test_ingest_hevc_transcode.py @@ -199,8 +199,33 @@ def hevc_sources(tmp_path): ) def tag_rotate(src: Path, rotation: int) -> Path: + """给横屏 HEVC 素材打上旋转 display matrix，模拟 iPhone 竖屏。 + + ffmpeg 版本差异： + - 5.1+（含 CI 的 7.x）：mp4 muxer 支持 -display_rotation 输出选项， + 老式 -metadata rotate 在 stream copy 时不再写入 display matrix； + - 4.x：不识别 -display_rotation，仍用 -metadata:s:v:0 rotate=。 + 两条路径都试，造完用 probe_rotation 自检；都失败则返回 None， + 消费方跳过依赖旋转信息的断言（skip）。 + """ out = tmp_path / f"portrait_r{rotation}_hevc.mp4" - subprocess.run( + attempts = [ + # ffmpeg 5.1+: display_rotation 输出选项 + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(src), + "-c", + "copy", + "-display_rotation", + str(rotation), + str(out), + ], + # ffmpeg 4.x: 老式 rotate metadata [ FFMPEG, "-y", @@ -215,13 +240,19 @@ def hevc_sources(tmp_path): f"rotate={rotation}", str(out), ], - check=True, - ) - return out + ] + for cmd in attempts: + proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) + if proc.returncode == 0 and out.exists() and probe_rotation(str(out)) is not None: + return out + return None # 当前 ffmpeg 无法造出带 display matrix 的素材 + + r90 = tag_rotate(base_hevc, 90) + r270 = tag_rotate(base_hevc, 270) yield { - "portrait_r90": tag_rotate(base_hevc, 90), - "portrait_r270": tag_rotate(base_hevc, 270), + "portrait_r90": r90, + "portrait_r270": r270, "landscape": base_hevc, "tmp_path": tmp_path, } @@ -265,6 +296,8 @@ class TestTranscodeEndToEnd: def test_portrait_r90_hevc(self, hevc_sources): """竖屏 HEVC(rotation=90) 转码后：h264、1080x1920（h>w）、无 rotation side data。""" src = hevc_sources["portrait_r90"] + if src is None: + pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") dst = hevc_sources["tmp_path"] / "out_r90.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation == 90 @@ -298,6 +331,8 @@ class TestTranscodeEndToEnd: def test_portrait_r270_hevc(self, hevc_sources): """竖屏 HEVC(rotation=270→side_data 显示 -90) 同样转为 1080x1920。""" src = hevc_sources["portrait_r270"] + if src is None: + pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") dst = hevc_sources["tmp_path"] / "out_r270.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation == -90 # ffprobe side_data 规范化为 -90 @@ -326,6 +361,8 @@ class TestTranscodeEndToEnd: def test_thumbnail_portrait_direction(self, hevc_sources): """缩略图（thumbnail_generator 同款 ffmpeg 抽帧，依赖 autorotate）竖屏源→竖版图。""" src = hevc_sources["portrait_r90"] + if src is None: + pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") thumb = hevc_sources["tmp_path"] / "thumb.jpg" subprocess.run( [ @@ -354,6 +391,9 @@ class TestTranscodeEndToEnd: def test_probe_rotation_reads_side_data_not_tag_line(self, hevc_sources): """回归：旧实现同时请求 side_data+stream_tags 输出两行且取第一行， r90 文件会取到 "270"；新实现只读 side_data，r90→90、r270→-90。""" - assert probe_rotation(str(hevc_sources["portrait_r90"])) == 90 - assert probe_rotation(str(hevc_sources["portrait_r270"])) == -90 + r90, r270 = hevc_sources["portrait_r90"], hevc_sources["portrait_r270"] + if r90 is None or r270 is None: + pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") + assert probe_rotation(str(r90)) == 90 + assert probe_rotation(str(r270)) == -90 assert probe_rotation(str(hevc_sources["landscape"])) is None -- 2.54.0 From b03d19cf3278f73ab4420426477a1cef73f5257b Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 30 Aug 2026 20:41:33 +0800 Subject: [PATCH 3/5] =?UTF-8?q?fix(ingest):=20=E6=96=B9=E5=90=91=E5=88=A4?= =?UTF-8?q?=E5=AE=9A=E6=94=B9=E4=B8=BA=E6=8C=89=E6=98=BE=E7=A4=BA=E6=96=B9?= =?UTF-8?q?=E5=90=91=EF=BC=88=E7=89=A9=E7=90=86=E7=AB=96=E5=B1=8F=20rotati?= =?UTF-8?q?on=3D0=20=E4=B8=8D=E5=86=8D=E8=AF=AF=E5=88=A4=E6=A8=AA=E5=B1=8F?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code Review 阻塞问题修复： 旧逻辑仅以 rotation side data 判定竖屏（is_portrait_rotation），对 "已物理旋转、存储即 h>w、rotation=0/None"的竖屏视频（Android 拍摄常见） 会误判为横屏：套用横屏按高缩放滤镜把 1080x1920 压成 608x1080，产物被 validate_transcode_output 拦截后降级使用原始 HEVC 文件——此类视频在 浏览器仍无法播放（HEVC 黑帧），转码链路对这类素材完全失效。 修复： - 新增 is_portrait_video(stored_w, stored_h, rotation)：rotation 为 90/270/-90 时显示宽高相对存储维度互换，否则显示方向即存储维度，按 显示 h>w 判定竖屏；维度探测失败时退回仅看 rotation。 - 主流程 Step 2 同时 probe 源文件维度与 rotation，按显示方向选滤镜。 - is_portrait_rotation 保留并补充注释（仅覆盖 iOS 风格 side data）。 测试补充： - TestIsPortraitVideo：iOS 风格（1920x1080+rotation90）、物理竖屏 （1080x1920 无 rotation）、横屏、rotation=180 不互换、探测失败回退。 - 端到端新增 test_physical_portrait_hevc_no_rotation：真实 ffmpeg 造 1080x1920 无 side data 的 HEVC，断言转码后仍为 1080x1920 h264。 - 任务级新增 test_physical_portrait_no_rotation_still_transcodes： 断言物理竖屏也走转码、改写 storage_key、使用按宽缩放滤镜。 - FONT 改为多路径候选查找，缺失时 drawtext 降级（仅红条标记）， 测试不再硬依赖 Debian 字体路径。 注：转码临时文件的 finally 清理逻辑在 #1449 中已存在（ingest.py 转码 try 块 finally 段 unlink _tc_tmp），本次审查该条为窗口外误判。 --- apps/worker/worker_app/tasks/ingest.py | 53 +++++++-- tests/unit/test_ingest_hevc_transcode.py | 101 ++++++++++++++++-- tests/unit/test_ingest_hevc_transcode_task.py | 26 ++++- 3 files changed, 162 insertions(+), 18 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 48938ba15..65b2b8db7 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -200,10 +200,37 @@ def probe_rotation(path: str) -> int | None: def is_portrait_rotation(rotation: int | None) -> bool: - """rotation side data 为 ±90/270 时表示竖屏拍摄。""" + """rotation side data 为 ±90/270 时表示竖屏拍摄。 + + 注意：这只覆盖"存储横屏 + display matrix 旋转"的 iOS 风格视频； + 物理竖屏视频（Android 常见，存储即 h>w、rotation=None/0）不会命中， + 方向判定请用 is_portrait_video()。 + """ return rotation in (90, 270, -90) +def is_portrait_video( + stored_width: int | None, + stored_height: int | None, + rotation: int | None, +) -> bool: + """按显示方向判断是否竖屏（显示高度 > 显示宽度）。 + + - rotation 为 90/270/-90 时，显示方向的宽高相对存储维度互换； + - rotation 为 0/180/None 时，显示方向即存储维度。 + + 这样两类竖屏都能正确识别： + - iOS：存储 1920x1080 + rotation=90 → 显示 1080x1920 竖屏 + - Android/物理竖屏：存储 1080x1920、无 rotation → 显示 1080x1920 竖屏 + 探测失败（维度为 None）时退回仅看 rotation，保证调用链不中断。 + """ + if not stored_width or not stored_height: + return is_portrait_rotation(rotation) + if is_portrait_rotation(rotation): + return stored_width > stored_height + return stored_height > stored_width + + def build_transcode_vf(is_portrait: bool) -> str: """构建转码视频滤镜。 @@ -386,16 +413,22 @@ def ingest_asset(job_id: str) -> dict: except Exception as _disk_err: logger.warning("磁盘检查失败，仍尝试转码: job_id=%s err=%s", job_id, _disk_err) - # ── Step 2: ffprobe 旋转检测（失败按横屏处理，后续方向校验兜底）── + # ── Step 2: 方向检测（按显示方向判定竖/横屏）────────────── + # 不能只看 rotation side data：Android 等设备的物理竖屏视频 + # 存储维度已是 h>w 且 rotation=0/None，只看 rotation 会误判横屏、 + # 套用横屏滤镜把 1080x1920 压成 608x1080，转码产物校验失败降级， + # 用户拿到 HEVC 原文件浏览器仍黑帧。 _rotation = probe_rotation(str(local_file)) - _is_portrait = is_portrait_rotation(_rotation) - if _rotation is not None: - logger.info( - "检测到视频 rotation=%s (portrait=%s): job_id=%s", - _rotation, - _is_portrait, - job_id, - ) + _src_w, _src_h = probe_dimensions(str(local_file)) + _is_portrait = is_portrait_video(_src_w, _src_h, _rotation) + logger.info( + "视频方向检测: stored=%sx%s rotation=%s portrait=%s: job_id=%s", + _src_w, + _src_h, + _rotation, + _is_portrait, + job_id, + ) # ── Step 3: ffmpeg 转码（独立 try/except）── try: diff --git a/tests/unit/test_ingest_hevc_transcode.py b/tests/unit/test_ingest_hevc_transcode.py index 3839081fe..07a8ddd57 100644 --- a/tests/unit/test_ingest_hevc_transcode.py +++ b/tests/unit/test_ingest_hevc_transcode.py @@ -27,6 +27,7 @@ from worker_app.tasks.ingest import ( # noqa: E402 build_transcode_vf, is_hevc_codec, is_portrait_rotation, + is_portrait_video, probe_dimensions, probe_rotation, validate_transcode_output, @@ -58,6 +59,36 @@ class TestPortraitRotation: assert not is_portrait_rotation(rotation), f"rotation={rotation} 不应判定为竖屏" +class TestIsPortraitVideo: + """按显示方向判定竖屏（存储维度 + rotation 互换）。""" + + def test_ios_style_stored_landscape_with_rotation90(self): + # iOS：存储 1920x1080 + rotation=90 → 显示 1080x1920 竖屏 + assert is_portrait_video(1920, 1080, 90) is True + assert is_portrait_video(1920, 1080, -90) is True + assert is_portrait_video(1920, 1080, 270) is True + + def test_physical_portrait_no_rotation(self): + # Android/物理竖屏：存储 1080x1920、无 rotation → 竖屏（旧逻辑误判横屏） + assert is_portrait_video(1080, 1920, None) is True + assert is_portrait_video(1080, 1920, 0) is True + + def test_landscape_normal(self): + assert is_portrait_video(1920, 1080, None) is False + assert is_portrait_video(1920, 1080, 0) is False + + def test_rotation180_no_swap(self): + # 180 度不互换宽高 + assert is_portrait_video(1920, 1080, 180) is False + assert is_portrait_video(1080, 1920, 180) is True + + def test_dimensions_unknown_falls_back_to_rotation(self): + # 探测失败（None）退回仅看 rotation，不抛异常 + assert is_portrait_video(None, None, 90) is True + assert is_portrait_video(None, None, None) is False + assert is_portrait_video(0, 0, 270) is True + + class TestBuildTranscodeVF: def test_comma_escaped_with_backslash(self): r"""scale 表达式内的逗号必须 \, 转义（否则报 Invalid size / No such filter）。""" @@ -143,18 +174,25 @@ pytestmark = pytest.mark.skipif( reason="ffmpeg/ffprobe/libx265 不可用，跳过端到端转码测试", ) -FONT = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" +# 字体候选路径（Debian/Ubuntu/Alpine/macOS），找不到则省略 drawtext， +# 仅靠顶部红条表达方向，测试断言不依赖文字。 +_FONT_CANDIDATES = ( + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", + "/usr/share/fonts/dejavu/DejaVuSans-Bold.ttf", + "/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", + "/System/Library/Fonts/Supplemental/Arial Bold.ttf", +) +FONT = next((f for f in _FONT_CANDIDATES if Path(f).exists()), None) @pytest.fixture def hevc_sources(tmp_path): """构造带方向标记的 HEVC 测试素材（横屏存储 + rotation side data，模拟 iPhone）。""" base = tmp_path / "base_landscape_h264.mp4" - draw = ( - "drawbox=x=0:y=0:w=1920:h=200:color=red:t=fill," - f"drawtext=fontfile={FONT}:text='TOP':fontsize=160:fontcolor=black:" - "x=(w-tw)/2:y=30" - ) + # 顶部红条标记画面方向；有字体时叠加 TOP 文字（仅人眼校验用，断言不依赖） + draw = "drawbox=x=0:y=0:w=1920:h=200:color=red:t=fill" + if FONT: + draw += f",drawtext=fontfile={FONT}:text='TOP':fontsize=160:fontcolor=black:" "x=(w-tw)/2:y=30" subprocess.run( [ FFMPEG, @@ -250,9 +288,41 @@ def hevc_sources(tmp_path): r90 = tag_rotate(base_hevc, 90) r270 = tag_rotate(base_hevc, 270) + # 物理竖屏素材（Android 风格）：直接生成 1080x1920 HEVC，无 rotation side data。 + # 红条画在存储帧的顶部（短边 1080 一侧），方向断言只看维度。 + physical = tmp_path / "portrait_physical_hevc.mp4" + phys_draw = "drawbox=x=0:y=0:w=1080:h=120:color=red:t=fill" + if FONT: + phys_draw += f",drawtext=fontfile={FONT}:text='TOP':fontsize=120:fontcolor=black:" "x=(w-tw)/2:y=20" + subprocess.run( + [ + FFMPEG, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-f", + "lavfi", + "-i", + "color=c=white:s=1080x1920:d=2:r=30", + "-vf", + phys_draw, + "-c:v", + "libx265", + "-tag:v", + "hvc1", + "-pix_fmt", + "yuv420p", + "-an", + str(physical), + ], + check=True, + ) + yield { "portrait_r90": r90, "portrait_r270": r270, + "portrait_physical": physical, "landscape": base_hevc, "tmp_path": tmp_path, } @@ -261,7 +331,8 @@ def hevc_sources(tmp_path): def _transcode_like_production(src: Path, dst: Path) -> tuple[bool, int | None]: """按生产代码相同方式执行转码，返回 (is_portrait, rotation)。""" rotation = probe_rotation(str(src)) - is_portrait = is_portrait_rotation(rotation) + width, height = probe_dimensions(str(src)) + is_portrait = is_portrait_video(width, height, rotation) vf = build_transcode_vf(is_portrait) subprocess.run( [ @@ -344,6 +415,22 @@ class TestTranscodeEndToEnd: assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), True) is True + def test_physical_portrait_hevc_no_rotation(self, hevc_sources): + """物理竖屏 HEVC（存储 1080x1920、无 rotation side data，Android 风格） + → 必须识别为竖屏，转出 1080x1920 h264；旧逻辑只看 rotation 会误判横屏、 + 套横屏滤镜压成 608x1080 并被 validate 拦截降级，用户拿到不能播放的 HEVC。""" + src = hevc_sources["portrait_physical"] + dst = hevc_sources["tmp_path"] / "out_physical.mp4" + is_portrait, rotation = _transcode_like_production(src, dst) + assert rotation is None + assert is_portrait is True + + width, height = probe_dimensions(str(dst)) + assert (width, height) == (1080, 1920) + assert height > width + assert probe_rotation(str(dst)) is None + assert validate_transcode_output(str(dst), True) is True + def test_landscape_hevc(self, hevc_sources): """横屏 HEVC 转码后：h264、1920x1080（w>h）、无 rotation side data。""" src = hevc_sources["landscape"] diff --git a/tests/unit/test_ingest_hevc_transcode_task.py b/tests/unit/test_ingest_hevc_transcode_task.py index cfb82b7e8..f6726f917 100644 --- a/tests/unit/test_ingest_hevc_transcode_task.py +++ b/tests/unit/test_ingest_hevc_transcode_task.py @@ -110,6 +110,7 @@ def task_env(tmp_path): "validate_ok": True, "upload_url": "https://oss.example.com/x_h264.MOV", "codec": "hevc", + "source_dims": (1920, 1080), "tc_out": tc_out, "local_file": local_file, } @@ -122,7 +123,7 @@ def task_env(tmp_path): def fake_probe_dimensions(path): if Path(path).name == tc_out.name: return (1080, 1920) if control["validate_ok"] else (1920, 1080) - return (1920, 1080) + return control["source_dims"] control["subprocess_calls"] = [] @@ -247,6 +248,29 @@ class TestIngestHEVCTranscodeFlow: assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" mocks["upload"].assert_not_called() + def test_physical_portrait_no_rotation_still_transcodes(self, task_env): + """物理竖屏（存储 1080x1920、rotation=None，Android 风格）也必须判定竖屏 + 并转码改写 storage_key——回归旧逻辑只看 rotation 误判横屏的 bug。""" + task_env["source_dims"] = (1080, 1920) + task_env["rotation_source"] = None + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281_h264.MOV" + mocks["upload"].assert_called_once() + # 竖屏滤镜按宽缩放（表达式引用 iw 判断），不应是横屏的按高缩放 + cmds = [] + for call in mocks["subprocess"].call_args_list: + cmd = call.args[0] if call.args else call.kwargs.get("cmd", []) + cmds.append(cmd) + vfs = [str(c) for c in cmds if c and c[0] == "ffmpeg" and "libx264" in c] + assert vfs, "应执行 libx264 转码" + assert any("gt(iw" in vf for vf in vfs), f"竖屏应使用按宽缩放滤镜: {vfs[0]}" + def test_non_hevc_no_transcode(self, task_env): """非 HEVC 编码（h264）→ 不触发 ffmpeg 转码。""" task_env["codec"] = "h264" -- 2.54.0 From 0eba41332c802be525c16d6e11f1af17327bd478 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 30 Aug 2026 20:53:21 +0800 Subject: [PATCH 4/5] =?UTF-8?q?refactor(ingest):=20=E8=BD=AC=E7=A0=81?= =?UTF-8?q?=E9=AD=94=E6=B3=95=E6=95=B0=E5=AD=97=E6=8F=90=E5=8F=96=E4=B8=BA?= =?UTF-8?q?=E5=B8=B8=E9=87=8F=EF=BC=881080/1920/900s=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 响应 review 建议：TRANSCODE_TARGET_EDGE=1080、TRANSCODE_MAX_LONG_EDGE =1920、TRANSCODE_TIMEOUT_SECONDS=900，vf 字符串改 rf-string 引用常量， 渲染结果不变。 --- apps/worker/worker_app/tasks/ingest.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 65b2b8db7..e39334f0b 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -152,11 +152,15 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: # ── HEVC 自动转码辅助函数（模块级，便于单元测试）───────────────────────── HEVC_CODECS = ("hevc", "h265", "hvh1") +# 转码目标：短边/长边封顶 1080p（只缩不放），产物长边不得超过 1920 +TRANSCODE_TARGET_EDGE = 1080 +TRANSCODE_MAX_LONG_EDGE = 1920 +TRANSCODE_TIMEOUT_SECONDS = 900 # ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义， # 否则逗号被当作 filter 分隔符解析，报 "No such filter: '1080)' / Invalid size"。 # subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。 -_PORTRAIT_VF = r"scale=if(gt(iw\,1080)\,1080\,iw):-2,format=yuv420p" -_LANDSCAPE_VF = r"scale=-2:if(gt(ih\,1080)\,1080\,ih),format=yuv420p" +_PORTRAIT_VF = rf"scale=if(gt(iw\,{TRANSCODE_TARGET_EDGE})\,{TRANSCODE_TARGET_EDGE}\,iw):-2,format=yuv420p" +_LANDSCAPE_VF = rf"scale=-2:if(gt(ih\,{TRANSCODE_TARGET_EDGE})\,{TRANSCODE_TARGET_EDGE}\,ih),format=yuv420p" def is_hevc_codec(codec: str | None) -> bool: @@ -276,7 +280,7 @@ def probe_dimensions(path: str) -> tuple[int | None, int | None]: def validate_transcode_output( output_path: str, expected_portrait: bool, - max_long_edge: int = 1920, + max_long_edge: int = TRANSCODE_MAX_LONG_EDGE, ) -> bool: """校验转码产物方向与维度。 @@ -478,7 +482,7 @@ def ingest_asset(job_id: str) -> dict: stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, - timeout=900, + timeout=TRANSCODE_TIMEOUT_SECONDS, ) if _proc.returncode == 0 and _tc_tmp.exists() and _tc_tmp.stat().st_size > 0: # ── Step 4: 方向/维度校验，不符则降级，杜绝横屏文件覆盖 ── -- 2.54.0 From e68801eb593d7f14fce6fbde63f3b9ed00d9727e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 30 Aug 2026 21:09:43 +0800 Subject: [PATCH 5/5] =?UTF-8?q?fix(ingest):=20=E8=BD=AC=E7=A0=81=E6=BB=A4?= =?UTF-8?q?=E9=95=9C=E6=94=B9=E6=8C=89=E9=95=BF=E8=BE=B9=201920=20?= =?UTF-8?q?=E5=B0=81=E9=A1=B6=EF=BC=88=E4=BF=AE=E5=A4=8D=E8=B6=85=E5=AE=BD?= =?UTF-8?q?=E5=B1=8F=E8=AF=AF=E9=99=8D=E7=BA=A7=EF=BC=89+=20=E5=90=88?= =?UTF-8?q?=E5=B9=B6=20ffprobe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code Review 阻塞问题修复 + 建议采纳： 1. 缩放规则与校验规则对齐（真 bug）： 旧滤镜仅按短边 1080 触发缩放（横屏看 ih、竖屏看 iw），对超宽屏 （如 4000x1000）短边不超 1080 完全不缩放，产物长边 4000 超过 validate_transcode_output 的 1920 上限，转码被判定失败而降级用 原始 HEVC——浏览器仍无法播放。改为统一滤镜： scale=w=if(gte(iw\,ih)\,min(1920\,iw)\,-2): h=if(gt(ih\,iw)\,min(1920\,ih)\,-2),format=yuv420p 横屏限宽、竖屏限高、短边 -2 自适应，min() 保证小视频不放大； build_transcode_vf 不再需要 is_portrait 参数（方向无关）。 ffmpeg 实测：1920x1080→1920x1080、4K竖屏→1080x1920、 4000x1000→1920x480、1080x2400→864x1920、640x360 不放大，均无 side data。 2. 合并 ffprobe（性能建议）：新增 probe_video_info(path) 一次 -show_streams -of json 同时解析 width/height 与 Display Matrix rotation（用全量 JSON 输出，规避 show_entries 嵌套 section 名在 ffmpeg 4.x/7.x 不一致的问题）；主流程两次探测合并为一次。 probe_rotation/probe_dimensions 保留供测试与 validate 复用。 3. 临时文件清理：转码 try 块 finally 中 _tc_tmp.unlink() 在 #1449 即已存在（develop 基线代码），本轮 review 该条为窗口外误判， 已在 PR 评论说明。 测试： - TestBuildTranscodeVF 重写为统一滤镜断言（逗号转义、无 transpose、 横竖双分支 min(1920)、yuv420p 后缀）。 - 新增 TestProbeVideoInfo（真实 ffprobe 320x240 无 rotation； 不存在文件返回三元 None）。 - 端到端新增 test_ultra_wide_long_edge_capped（4000x1000→1920x480 且 validate 通过）；_transcode_like_production 改用 probe_video_info + 无参 build_transcode_vf，与生产完全一致。 - 任务级竖屏用例滤镜特征断言更新为 min(1920)。 --- apps/worker/worker_app/tasks/ingest.py | 85 ++++++++++--- tests/unit/test_ingest_hevc_transcode.py | 116 ++++++++++++++---- tests/unit/test_ingest_hevc_transcode_task.py | 5 +- 3 files changed, 167 insertions(+), 39 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index e39334f0b..c9ac3f79e 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -1,3 +1,4 @@ +import json import shutil import subprocess import tempfile @@ -152,15 +153,20 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: # ── HEVC 自动转码辅助函数（模块级，便于单元测试）───────────────────────── HEVC_CODECS = ("hevc", "h265", "hvh1") -# 转码目标：短边/长边封顶 1080p（只缩不放），产物长边不得超过 1920 -TRANSCODE_TARGET_EDGE = 1080 +# 转码目标：长边封顶 1920（只缩不放，与 validate 的 max_long_edge 一致）， +# 竖屏/横屏/超宽屏统一按长边等比缩放，短边自动按比例（-2 保证偶数）。 TRANSCODE_MAX_LONG_EDGE = 1920 TRANSCODE_TIMEOUT_SECONDS = 900 # ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义， -# 否则逗号被当作 filter 分隔符解析，报 "No such filter: '1080)' / Invalid size"。 +# 否则逗号被当作 filter 分隔符解析，报 "No such filter" / Invalid size。 # subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。 -_PORTRAIT_VF = rf"scale=if(gt(iw\,{TRANSCODE_TARGET_EDGE})\,{TRANSCODE_TARGET_EDGE}\,iw):-2,format=yuv420p" -_LANDSCAPE_VF = rf"scale=-2:if(gt(ih\,{TRANSCODE_TARGET_EDGE})\,{TRANSCODE_TARGET_EDGE}\,ih),format=yuv420p" +# 横屏（iw>=ih）限宽 min(1920,iw)、高 -2 自适应；竖屏（ih>iw）限高、宽自适应； +# min() 保证小视频不放大。与 validate_transcode_output 的"长边 <= 1920"规则对齐， +# 超宽屏（如 4000x1000）短边不触发旧的短边缩放、长边超限被误降级的问题由此消除。 +_TRANSCODE_VF = ( + rf"scale=w=if(gte(iw\,ih)\,min({TRANSCODE_MAX_LONG_EDGE}\,iw)\,-2):" + rf"h=if(gt(ih\,iw)\,min({TRANSCODE_MAX_LONG_EDGE}\,ih)\,-2),format=yuv420p" +) def is_hevc_codec(codec: str | None) -> bool: @@ -235,18 +241,20 @@ def is_portrait_video( return stored_height > stored_width -def build_transcode_vf(is_portrait: bool) -> str: - """构建转码视频滤镜。 +def build_transcode_vf() -> str: + """构建转码视频滤镜（竖屏/横屏统一，按显示长边封顶 1920、只缩不放）。 依赖 ffmpeg 内置 autorotate（默认开启）按 display matrix 物理旋转画面， - 输出自动剥离 rotation side data；这里只做"只缩不放"的 1080p 等比缩放： - - 竖屏（旋转后 w tuple[int | None, int | None]: @@ -277,6 +285,52 @@ def probe_dimensions(path: str) -> tuple[int | None, int | None]: return None, None +def probe_video_info(path: str) -> tuple[int | None, int | None, int | None]: + """一次 ffprobe 同时读取视频宽高与旋转角度（display matrix side data）。 + + 返回 (width, height, rotation)；探测失败对应位置为 None。 + 合并维度/角度两次探测，减少大文件、高并发下的 ffprobe 进程开销。 + rotation 仅取 stream side_data_list 的 Display Matrix（不读 tags.rotate， + 避免 iOS 文件 tag 值与 side data 双来源取错）。用 -show_streams 全量 JSON + 输出解析，兼容 ffmpeg 4.x/7.x（show_entries 嵌套 section 名跨版本不一致）。 + """ + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_streams", + "-of", + "json", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + data = json.loads(result.stdout or "{}") + streams = data.get("streams") or [] + if not streams: + return None, None, None + stream = streams[0] + width = int(stream["width"]) if stream.get("width") else None + height = int(stream["height"]) if stream.get("height") else None + rotation = None + for side in stream.get("side_data_list") or []: + if side.get("side_data_type") == "Display Matrix" and side.get("rotation") is not None: + deg = int(round(float(side["rotation"]))) % 360 + # ffprobe：顺时针 90 拍摄输出 90，逆时针 90 输出 -90（归一为 270） + rotation = {0: 0, 90: 90, 180: 180, 270: -90}.get(deg, deg if deg in (90, 180) else None) + break + return width, height, rotation + except (subprocess.TimeoutExpired, ValueError, OSError, json.JSONDecodeError, KeyError, TypeError): + return None, None, None + + def validate_transcode_output( output_path: str, expected_portrait: bool, @@ -422,8 +476,7 @@ def ingest_asset(job_id: str) -> dict: # 存储维度已是 h>w 且 rotation=0/None，只看 rotation 会误判横屏、 # 套用横屏滤镜把 1080x1920 压成 608x1080，转码产物校验失败降级， # 用户拿到 HEVC 原文件浏览器仍黑帧。 - _rotation = probe_rotation(str(local_file)) - _src_w, _src_h = probe_dimensions(str(local_file)) + _src_w, _src_h, _rotation = probe_video_info(str(local_file)) _is_portrait = is_portrait_video(_src_w, _src_h, _rotation) logger.info( "视频方向检测: stored=%sx%s rotation=%s portrait=%s: job_id=%s", @@ -444,7 +497,7 @@ def ingest_asset(job_id: str) -> dict: # 输出自动剥离 side data）；滤镜只做 1080p 等比"只缩不放"。 # 注意不能再加 transpose：旧逻辑 autorotate + transpose 双重旋转， # 竖屏被转成横屏；scale 表达式内逗号必须 \, 转义（见 build_transcode_vf）。 - _vf = build_transcode_vf(_is_portrait) + _vf = build_transcode_vf() _cmd = [ "ffmpeg", diff --git a/tests/unit/test_ingest_hevc_transcode.py b/tests/unit/test_ingest_hevc_transcode.py index 07a8ddd57..11b975923 100644 --- a/tests/unit/test_ingest_hevc_transcode.py +++ b/tests/unit/test_ingest_hevc_transcode.py @@ -30,6 +30,7 @@ from worker_app.tasks.ingest import ( # noqa: E402 is_portrait_video, probe_dimensions, probe_rotation, + probe_video_info, validate_transcode_output, ) @@ -90,33 +91,65 @@ class TestIsPortraitVideo: class TestBuildTranscodeVF: + """统一转码滤镜：长边封顶 1920、只缩不放、方向无关。""" + def test_comma_escaped_with_backslash(self): r"""scale 表达式内的逗号必须 \, 转义（否则报 Invalid size / No such filter）。""" - for vf in (build_transcode_vf(True), build_transcode_vf(False)): - assert "\\," in vf - # 不应存在未转义的裸逗号（filter 分隔）出现在 if 表达式内 - assert "gt(ih,1080)" not in vf - assert "gt(iw,1080)" not in vf + vf = build_transcode_vf() + assert "\\," in vf + # 不应存在未转义的裸逗号（filter 分隔）出现在 if 表达式内 + assert "gte(iw,ih)" not in vf + assert "gt(ih,iw)" not in vf + assert "min(1920,iw)" not in vf def test_no_transpose_filter(self): """不能显式 transpose：ffmpeg autorotate 已按 side data 物理旋转， 再加 transpose 会双重旋转把竖屏转成横屏。""" - assert "transpose" not in build_transcode_vf(True) - assert "transpose" not in build_transcode_vf(False) + assert "transpose" not in build_transcode_vf() - def test_portrait_scales_by_width(self): - """竖屏（旋转后 w tuple[bool, int | None]: """按生产代码相同方式执行转码，返回 (is_portrait, rotation)。""" - rotation = probe_rotation(str(src)) - width, height = probe_dimensions(str(src)) + # 与生产一致：合并探测 + 统一滤镜 + width, height, rotation = probe_video_info(str(src)) is_portrait = is_portrait_video(width, height, rotation) - vf = build_transcode_vf(is_portrait) + vf = build_transcode_vf() subprocess.run( [ FFMPEG, @@ -445,6 +505,20 @@ class TestTranscodeEndToEnd: assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), False) is True + def test_ultra_wide_long_edge_capped(self, hevc_sources): + """超宽屏 4000x1000：长边必须封顶 1920（→1920x480），validate 通过。 + 回归旧滤镜只按短边触发缩放、长边超 1920 被误降级的 bug。""" + src = hevc_sources["ultra_wide"] + dst = hevc_sources["tmp_path"] / "out_ultra.mp4" + is_portrait, rotation = _transcode_like_production(src, dst) + assert rotation is None + assert is_portrait is False + + width, height = probe_dimensions(str(dst)) + assert max(width, height) <= 1920, f"长边应封顶 1920，实际 {width}x{height}" + assert width == 1920 and height == 480 + assert validate_transcode_output(str(dst), False) is True + def test_thumbnail_portrait_direction(self, hevc_sources): """缩略图（thumbnail_generator 同款 ffmpeg 抽帧，依赖 autorotate）竖屏源→竖版图。""" src = hevc_sources["portrait_r90"] diff --git a/tests/unit/test_ingest_hevc_transcode_task.py b/tests/unit/test_ingest_hevc_transcode_task.py index f6726f917..a02a86bc1 100644 --- a/tests/unit/test_ingest_hevc_transcode_task.py +++ b/tests/unit/test_ingest_hevc_transcode_task.py @@ -262,14 +262,15 @@ class TestIngestHEVCTranscodeFlow: assert result["status"] == "completed" assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281_h264.MOV" mocks["upload"].assert_called_once() - # 竖屏滤镜按宽缩放（表达式引用 iw 判断），不应是横屏的按高缩放 + # 统一滤镜按长边 1920 封顶（横/竖分支都在），不应再出现按短边 1080 的旧表达式 cmds = [] for call in mocks["subprocess"].call_args_list: cmd = call.args[0] if call.args else call.kwargs.get("cmd", []) cmds.append(cmd) vfs = [str(c) for c in cmds if c and c[0] == "ffmpeg" and "libx264" in c] assert vfs, "应执行 libx264 转码" - assert any("gt(iw" in vf for vf in vfs), f"竖屏应使用按宽缩放滤镜: {vfs[0]}" + assert any("min(1920" in vf for vf in vfs), f"应使用长边1920封顶滤镜: {vfs[0]}" + assert all("gt(ih,1080)" not in vf for vf in vfs), "不应再用短边1080旧表达式" def test_non_hevc_no_transcode(self, task_env): """非 HEVC 编码（h264）→ 不触发 ffmpeg 转码。""" -- 2.54.0