apps/worker/worker_app/tasks/ingest.py

@@ -1,3 +1,4 @@
+import json
 import shutil
 import subprocess
 import tempfile
@@ -150,6 +151,214 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]:
     return metadata, success
 
 
+# ── HEVC 自动转码辅助函数（模块级，便于单元测试）─────────────────────────
+HEVC_CODECS = ("hevc", "h265", "hvh1")
+# 转码目标：长边封顶 1920（只缩不放，与 validate 的 max_long_edge 一致），
+# 竖屏/横屏/超宽屏统一按长边等比缩放，短边自动按比例（-2 保证偶数）。
+TRANSCODE_MAX_LONG_EDGE = 1920
+TRANSCODE_TIMEOUT_SECONDS = 900
+# ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义，
+# 否则逗号被当作 filter 分隔符解析，报 "No such filter" / Invalid size。
+# subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。
+# 横屏（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:
+    """判断编码是否为 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 时表示竖屏拍摄。
+
+    注意：这只覆盖"存储横屏 + 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() -> str:
+    """构建转码视频滤镜（竖屏/横屏统一，按显示长边封顶 1920、只缩不放）。
+
+    依赖 ffmpeg 内置 autorotate（默认开启）按 display matrix 物理旋转画面，
+    输出自动剥离 rotation side data；滤镜只做等比缩放，方向无关：
+    横屏限宽、竖屏限高，短边 -2 自适应偶数，min() 保证小视频不放大。
+
+    旧实现的问题：
+    - 显式 transpose=1 与 autorotate 叠加，竖屏被二次旋转成横屏；
+    - 竖屏沿用按高缩放表达式，1080x1920 被错误缩成 608x1080；
+    - 仅按短边 1080 触发缩放，超宽屏（如 4000x1000）长边超 1920 会被
+      validate 拦截误降级，用户拿到浏览器无法播放的 HEVC 原文件。
+    """
+    return _TRANSCODE_VF
+
+
+def probe_dimensions(path: str) -> 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 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,
+    max_long_edge: int = TRANSCODE_MAX_LONG_EDGE,
+) -> 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 +454,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 +471,21 @@ 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",
-                            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 2: 方向检测（按显示方向判定竖/横屏）──────────────
+                    # 不能只看 rotation side data：Android 等设备的物理竖屏视频
+                    # 存储维度已是 h>w 且 rotation=0/None，只看 rotation 会误判横屏、
+                    # 套用横屏滤镜把 1080x1920 压成 608x1080，转码产物校验失败降级，
+                    # 用户拿到 HEVC 原文件浏览器仍黑帧。
+                    _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",
+                        _src_w,
+                        _src_h,
+                        _rotation,
+                        _is_portrait,
+                        job_id,
+                    )
 
                     # ── Step 3: ffmpeg 转码（独立 try/except）──
                     try:
@@ -315,11 +493,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()
 
                         _cmd = [
                             "ffmpeg",
@@ -333,7 +511,7 @@ def ingest_asset(job_id: str) -> dict:
                             "-crf",
                             "18",
                             "-vf",
-                            _vf + ",format=yuv420p",
+                            _vf,
                             "-colorspace",
                             "bt709",
                             "-color_primaries",
@@ -344,54 +522,62 @@ 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,
                             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:
-                            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(

tests/unit/test_ingest_hevc_transcode.py

@@ -1,255 +1,560 @@
 """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,
+    is_portrait_video,
+    probe_dimensions,
+    probe_rotation,
+    probe_video_info,
+    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 TestIsPortraitVideo:
+    """按显示方向判定竖屏（存储维度 + rotation 互换）。"""
 
-        # 模拟转码成功
-        transcode_success = True
+    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
 
-        # 模拟元数据提取失败
-        new_metadata = {}
-        new_extract_success = False
+    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
 
-        # 修复后的逻辑：先提取元数据，确认成功后再更新
-        if transcode_success:
-            if new_extract_success:
-                job_storage_key = new_storage_key
-                metadata = new_metadata
-            # 如果元数据提取失败，不更新 job_storage_key
+    def test_landscape_normal(self):
+        assert is_portrait_video(1920, 1080, None) is False
+        assert is_portrait_video(1920, 1080, 0) is False
 
-        # 验证：storage_key 保持原值，没有脏数据
-        assert job_storage_key == original_storage_key
-        assert metadata["codec"] == "hevc"  # 保持原始元数据
+    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_metadata_success_updates_storage_key(self):
-        """验证元数据提取成功时正确更新 storage_key"""
-        original_storage_key = "uploads/test/video.mp4"
-        new_storage_key = "uploads/test/video_h264.mp4"
+    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
 
-        job_storage_key = original_storage_key
-        metadata = {"codec": "hevc", "width": 3840, "height": 2160}
 
-        # 模拟转码成功
-        transcode_success = True
+class TestBuildTranscodeVF:
+    """统一转码滤镜：长边封顶 1920、只缩不放、方向无关。"""
 
-        # 模拟元数据提取成功
-        new_metadata = {"codec": "h264", "width": 1920, "height": 1080}
-        new_extract_success = True
+    def test_comma_escaped_with_backslash(self):
+        r"""scale 表达式内的逗号必须 \, 转义（否则报 Invalid size / No such filter）。"""
+        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
 
-        # 修复后的逻辑
-        if transcode_success:
-            if new_extract_success:
-                job_storage_key = new_storage_key
-                metadata = new_metadata
+    def test_no_transpose_filter(self):
+        """不能显式 transpose：ffmpeg autorotate 已按 side data 物理旋转，
+        再加 transpose 会双重旋转把竖屏转成横屏。"""
+        assert "transpose" not in build_transcode_vf()
 
-        # 验证：storage_key 和 metadata 都更新为新值
-        assert job_storage_key == new_storage_key
-        assert metadata["codec"] == "h264"
-        assert metadata["width"] == 1920
+    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 封顶"
 
-    @patch("subprocess.run")
-    def test_ffmpeg_timeout_degradation(self, mock_subprocess):
-        """验证 FFmpeg 超时降级使用原始文件"""
-        mock_subprocess.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300)
+    def test_format_yuv420p_suffix(self):
+        assert build_transcode_vf().endswith(",format=yuv420p")
 
-        # 模拟降级逻辑
-        transcode_success = False
-        try:
-            raise subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300)
-        except subprocess.TimeoutExpired:
-            transcode_success = False
 
-        assert not transcode_success, "超时应该导致转码失败"
+class TestProbeVideoInfo:
+    """probe_video_info 合并探测：维度 + rotation 一次 ffprobe。"""
 
-    @patch("subprocess.run")
-    def test_ffmpeg_error_degradation(self, mock_subprocess):
-        """验证 FFmpeg 执行失败降级使用原始文件"""
-        mock_subprocess.return_value = MagicMock(
-            returncode=1,
-            stderr="Error: Invalid data found when processing input",
+    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
 
-        result = mock_subprocess.return_value
-        transcode_success = result.returncode == 0
+    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)
 
-        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 视频应该被缩放"
+class TestValidateTranscodeOutput:
+    def _probe_dimensions_called_with(self, monkeypatch, w, h, rotation=None):
+        monkeypatch.setattr("worker_app.tasks.ingest.probe_dimensions", lambda p: (w, h))
+        monkeypatch.setattr("worker_app.tasks.ingest.probe_rotation", lambda p: rotation)
 
-    def test_scale_filter_logic_1080p_video(self):
-        """验证 1080p 视频不会被缩放"""
-        ih = 1080
-        should_scale = ih > 1080
-        assert not should_scale, "1080p 视频不应该被缩放"
+    def test_portrait_output_ok(self, monkeypatch):
+        self._probe_dimensions_called_with(monkeypatch, 1080, 1920, None)
+        assert validate_transcode_output("/tmp/fake.mp4", expected_portrait=True)
 
-    def test_scale_filter_logic_720p_video(self):
-        """验证 720p 视频不会被缩放"""
-        ih = 720
-        should_scale = ih > 1080
-        assert not should_scale, "720p 视频不应该被缩放"
+    def test_landscape_output_ok(self, monkeypatch):
+        self._probe_dimensions_called_with(monkeypatch, 1920, 1080, None)
+        assert validate_transcode_output("/tmp/fake.mp4", expected_portrait=False)
 
-    def test_tempfile_security_fix(self):
-        """验证使用 NamedTemporaryFile 替代 mktemp（安全修复）
+    def test_portrait_source_but_landscape_output_rejected(self, monkeypatch):
+        """竖屏源转出横屏（双重旋转 bug 产物）必须判失败降级。"""
+        self._probe_dimensions_called_with(monkeypatch, 1920, 1080, None)
+        assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=True)
 
-        AI Code Review 发现的安全漏洞：
-        - tempfile.mktemp 存在 TOCTOU 竞态条件
-        - 应该使用 NamedTemporaryFile(delete=False)
+    def test_landscape_source_but_portrait_output_rejected(self, monkeypatch):
+        self._probe_dimensions_called_with(monkeypatch, 1080, 1920, None)
+        assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False)
+
+    def test_residual_rotation_side_data_rejected(self, monkeypatch):
+        """产物仍带 rotation side data 会被播放器二次旋转，必须判失败。"""
+        self._probe_dimensions_called_with(monkeypatch, 1080, 1920, rotation=90)
+        assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=True)
+
+    def test_long_edge_beyond_1920_rejected(self, monkeypatch):
+        """只缩不放：长边不得超过 1920。"""
+        self._probe_dimensions_called_with(monkeypatch, 3840, 2160, None)
+        assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False)
+
+    def test_missing_dimensions_rejected(self, monkeypatch):
+        self._probe_dimensions_called_with(monkeypatch, None, None, None)
+        assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False)
+
+
+# ── ffmpeg 端到端测试（无 ffmpeg/libx265 环境自动跳过）──────────────────────
+
+FFMPEG = shutil.which("ffmpeg")
+FFPROBE = shutil.which("ffprobe")
+
+
+def _has_x265() -> 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 不可用，跳过端到端转码测试",
+)
+
+# 字体候选路径（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"
+    # 顶部红条标记画面方向；有字体时叠加 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,
+            "-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:
+        """给横屏 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）。
         """
-        import tempfile
+        out = tmp_path / f"portrait_r{rotation}_hevc.mp4"
+        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",
+                "-hide_banner",
+                "-loglevel",
+                "error",
+                "-i",
+                str(src),
+                "-c",
+                "copy",
+                "-metadata:s:v:0",
+                f"rotate={rotation}",
+                str(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 的素材
 
-        with patch("tempfile.NamedTemporaryFile") as mock_ntf:
-            mock_file = MagicMock()
-            mock_file.name = "/tmp/test_h264.mp4"
-            mock_ntf.return_value = mock_file
+    r90 = tag_rotate(base_hevc, 90)
+    r270 = tag_rotate(base_hevc, 270)
 
-            # 新代码的调用方式
-            _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4")
-            _tc_tmp = Path(_tc_tmp_file.name)
-            _tc_tmp_file.close()
+    # 物理竖屏素材（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,
+    )
 
-            # 验证使用了 NamedTemporaryFile
-            mock_ntf.assert_called_once_with(delete=False, suffix="_h264.mp4")
+    # 超宽屏 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,
+    )
 
-    def test_subprocess_output_handling(self):
-        """验证 subprocess 输出处理（避免内存溢出）
+    yield {
+        "portrait_r90": r90,
+        "portrait_r270": r270,
+        "portrait_physical": physical,
+        "ultra_wide": ultra_wide,
+        "landscape": base_hevc,
+        "tmp_path": tmp_path,
+    }
 
-        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)。"""
+    # 与生产一致：合并探测 + 统一滤镜
+    width, height, rotation = probe_video_info(str(src))
+    is_portrait = is_portrait_video(width, height, rotation)
+    vf = build_transcode_vf()
+    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"]
+        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
+        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"]
+        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
+        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_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"]
+        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_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"]
+        if src is None:
+            pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材")
+        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。"""
+        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

tests/unit/test_ingest_hevc_transcode_task.py

@@ -0,0 +1,291 @@
+"""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",
+        "source_dims": (1920, 1080),
+        "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 control["source_dims"]
+
+    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_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()
+        # 统一滤镜按长边 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("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 转码。"""
+        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