fix(ingest): 转码滤镜改按长边 1920 封顶（修复超宽屏误降级）+ 合并 ffprobe

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

@@ -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<h）：宽超过 1080 则缩到 1080（4K 竖屏 → 1080x1920）
-    - 横屏：高超过 1080 则缩到 1080（4K 横屏 → 1920x1080）
+    输出自动剥离 rotation side data；滤镜只做等比缩放，方向无关：
+    横屏限宽、竖屏限高，短边 -2 自适应偶数，min() 保证小视频不放大。
 
-    旧实现显式 transpose=1 与 autorotate 叠加导致竖屏被二次旋转成横屏，
-    且竖屏沿用按高缩放的表达式，1080x1920 会被错误缩小为 608x1080。
+    旧实现的问题：
+    - 显式 transpose=1 与 autorotate 叠加，竖屏被二次旋转成横屏；
+    - 竖屏沿用按高缩放表达式，1080x1920 被错误缩成 608x1080；
+    - 仅按短边 1080 触发缩放，超宽屏（如 4000x1000）长边超 1920 会被
+      validate 拦截误降级，用户拿到浏览器无法播放的 HEVC 原文件。
     """
-    return _PORTRAIT_VF if is_portrait else _LANDSCAPE_VF
+    return _TRANSCODE_VF
 
 
 def probe_dimensions(path: str) -> 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",

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<h）：宽超 1080 缩到 1080，高自适应 → 4K 竖屏 1080x1920。"""
-        vf = build_transcode_vf(True)
-        assert vf.startswith("scale=if(gt(iw\\,1080)\\,1080\\,iw):-2")
-
-    def test_landscape_scales_by_height(self):
-        """横屏：高超 1080 缩到 1080，宽自适应 → 4K 横屏 1920x1080。"""
-        vf = build_transcode_vf(False)
-        assert vf.startswith("scale=-2:if(gt(ih\\,1080)\\,1080\\,ih)")
+    def test_long_edge_capped_1920_orientation_agnostic(self):
+        """横屏限宽、竖屏限高，均 min(1920,...)，短边 -2 自适应。"""
+        vf = build_transcode_vf()
+        assert "gte(iw" in vf and "gt(ih" in vf, "横/竖分支都要在"
+        assert vf.count("min(1920") == 2, "宽高分支都按长边 1920 封顶"
 
     def test_format_yuv420p_suffix(self):
-        for vf in (build_transcode_vf(True), build_transcode_vf(False)):
-            assert vf.endswith(",format=yuv420p")
+        assert build_transcode_vf().endswith(",format=yuv420p")
+
+
+class TestProbeVideoInfo:
+    """probe_video_info 合并探测：维度 + rotation 一次 ffprobe。"""
+
+    def test_merges_dimensions_and_rotation(self, tmp_path):
+        """真实 ffprobe：r90 素材 → (1920, 1080, 90)；r270 → rotation=-90；横屏 None。"""
+        # 端到端素材由 hevc_sources fixture 构造，这里用独立小素材验证合并函数
+        base = tmp_path / "v.mp4"
+        subprocess.run(
+            [
+                FFMPEG,
+                "-y",
+                "-hide_banner",
+                "-loglevel",
+                "error",
+                "-f",
+                "lavfi",
+                "-i",
+                "color=c=white:s=320x240:d=1:r=15",
+                "-c:v",
+                "libx264",
+                "-pix_fmt",
+                "yuv420p",
+                str(base),
+            ],
+            check=True,
+        )
+        w, h, rot = probe_video_info(str(base))
+        assert (w, h) == (320, 240)
+        assert rot is None
+
+    def test_probe_failure_returns_triple_none(self):
+        w, h, rot = probe_video_info("/nonexistent/path/fake.mp4")
+        assert (w, h, rot) == (None, None, None)
 
 
 class TestValidateTranscodeOutput:
@@ -319,10 +352,37 @@ def hevc_sources(tmp_path):
         check=True,
     )
 
+    # 超宽屏 HEVC（4000x1000，无 rotation）：旧滤镜短边 1000<1080 不缩放，
+    # 长边 4000 超 validate 的 1920 上限被误降级；新滤镜长边封顶应缩到 1920x480。
+    ultra_wide = tmp_path / "ultra_wide_hevc.mp4"
+    subprocess.run(
+        [
+            FFMPEG,
+            "-y",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=red:s=4000x1000:d=1:r=30",
+            "-c:v",
+            "libx265",
+            "-tag:v",
+            "hvc1",
+            "-pix_fmt",
+            "yuv420p",
+            "-an",
+            str(ultra_wide),
+        ],
+        check=True,
+    )
+
     yield {
         "portrait_r90": r90,
         "portrait_r270": r270,
         "portrait_physical": physical,
+        "ultra_wide": ultra_wide,
         "landscape": base_hevc,
         "tmp_path": tmp_path,
     }
@@ -330,10 +390,10 @@ 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))
-    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"]

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 转码。"""