apps/worker/video_processing/ffmpeg_utils.py

@@ -304,3 +304,127 @@ def normalize_video(
     ]
     run_ffmpeg(command)
     return {"width": width, "height": height, "path": output_path}
+
+
+def random_edge_crop(
+    input_path: str | Path,
+    output_path: str | Path | None = None,
+    *,
+    min_crop_pct: float = 0.02,
+    max_crop_pct: float = 0.05,
+) -> Path:
+    """对视频四边做随机裁剪再缩放回原分辨率，用于改变 pHash 指纹。
+
+    Args:
+        input_path: 输入视频路径
+        output_path: 输出路径；为 None 时写入 input_path 同目录的临时文件，
+                     成功后覆盖原文件
+        min_crop_pct: 每边最小裁剪比例（默认 2%）
+        max_crop_pct: 每边最大裁剪比例（默认 5%）
+
+    Returns:
+        输出文件路径（Path 对象）
+
+    Raises:
+        subprocess.CalledProcessError: ffmpeg 执行失败时抛出
+    """
+    import random
+    import shutil
+    import tempfile
+
+    input_path = Path(input_path)
+
+    # 获取原始分辨率
+    info = probe_video_info(str(input_path))
+    W = info["width"]
+    H = info["height"]
+
+    if W <= 0 or H <= 0:
+        logger.warning("无法获取视频分辨率 (W=%d H=%d)，跳过裁剪: %s", W, H, input_path)
+        return input_path
+
+    # 四边各自随机裁剪 2%~5%
+    crop_top = int(H * random.uniform(min_crop_pct, max_crop_pct))
+    crop_bottom = int(H * random.uniform(min_crop_pct, max_crop_pct))
+    crop_left = int(W * random.uniform(min_crop_pct, max_crop_pct))
+    crop_right = int(W * random.uniform(min_crop_pct, max_crop_pct))
+
+    # 裁剪后尺寸（确保至少 2 像素）
+    new_w = max(W - crop_left - crop_right, 2)
+    new_h = max(H - crop_top - crop_bottom, 2)
+    x_offset = crop_left
+    y_offset = crop_top
+
+    # 确保裁剪尺寸为偶数（ffmpeg 编码器常要求偶数尺寸）
+    new_w = new_w if new_w % 2 == 0 else new_w - 1
+    new_h = new_h if new_h % 2 == 0 else new_h - 1
+    if new_w < 2:
+        new_w = 2
+    if new_h < 2:
+        new_h = 2
+
+    # 输出分辨率必须与原始一致
+    out_w = W if W % 2 == 0 else W + 1
+    out_h = H if H % 2 == 0 else H + 1
+
+    vf = f"crop={new_w}:{new_h}:{x_offset}:{y_offset},scale={out_w}:{out_h}"
+
+    logger.info(
+        "随机边缘裁剪: %s → crop(%d,%d,%d,%d)=%dx%d scale→%dx%d",
+        input_path.name,
+        crop_top,
+        crop_bottom,
+        crop_left,
+        crop_right,
+        new_w,
+        new_h,
+        out_w,
+        out_h,
+    )
+
+    # 确定输出路径
+    if output_path is None:
+        temp_fd, temp_path = tempfile.mkstemp(suffix=".mp4", dir=input_path.parent)
+        import os
+
+        os.close(temp_fd)
+        temp_output = Path(temp_path)
+        replace_original = True
+    else:
+        temp_output = Path(output_path)
+        replace_original = False
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        str(input_path),
+        "-vf",
+        vf,
+        "-c:v",
+        "libx264",
+        "-preset",
+        "fast",
+        "-crf",
+        "18",
+        "-c:a",
+        "copy",
+        "-movflags",
+        "+faststart",
+        str(temp_output),
+    ]
+
+    try:
+        run_ffmpeg(command)
+    except Exception:
+        # 裁剪失败时清理临时文件
+        if temp_output.exists() and replace_original:
+            temp_output.unlink(missing_ok=True)
+        raise
+
+    # 成功 → 覆盖原文件
+    if replace_original:
+        shutil.move(str(temp_output), str(input_path))
+        return input_path
+
+    return temp_output

apps/worker/worker_app/tasks/generation.py

@@ -716,6 +716,28 @@ def generate_video(self, task_id: str) -> dict:
 
                     _update_task_progress(task_id, 80, "渲染完成")
 
+                    # ── 3.5 随机边缘裁剪降重（#1664） ──────────────────────────
+                    from video_processing.ffmpeg_utils import random_edge_crop
+
+                    try:
+                        cropped_path = random_edge_crop(output_path)
+                        if cropped_path != output_path:
+                            output_path = cropped_path
+                        if gen_task:
+                            gen_task.append_log("边缘裁剪", "已应用随机 2-5% 边缘裁剪降重")
+                            _flush_logs(task_id, gen_task)
+                        logger.info("[task_id=%s] 随机边缘裁剪完成: %s", task_id, output_path)
+                    except Exception as crop_err:
+                        logger.warning(
+                            "[task_id=%s] 随机边缘裁剪失败，使用原始视频继续: %s",
+                            task_id,
+                            crop_err,
+                            exc_info=True,
+                        )
+                        if gen_task:
+                            gen_task.append_log("边缘裁剪", f"裁剪失败，使用原始视频: {crop_err}")
+                            _flush_logs(task_id, gen_task)
+
                     # ── 4. 上传 OSS + 查重记录 ───────────────────────────────
                     _update_task_progress(task_id, 85, "开始上传")
                     file_url, duration, file_size, video_count = _upload_and_record(

tests/unit/test_random_edge_crop.py

@@ -0,0 +1,207 @@
+"""#1664 随机边缘裁剪降重功能测试"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "worker"))
+sys.path.insert(0, str(ROOT / "apps" / "api"))
+sys.path.insert(0, str(ROOT / "packages"))
+
+
+from video_processing.ffmpeg_utils import random_edge_crop
+
+
+class TestRandomEdgeCropBasic:
+    """基本功能测试"""
+
+    def test_returns_input_path_when_output_none(self, tmp_path):
+        """output_path=None 时覆盖原文件并返回 input_path"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg"),
+        ):
+            result = random_edge_crop(input_file)
+
+        assert result == input_file
+
+    def test_returns_output_path_when_specified(self, tmp_path):
+        """指定 output_path 时返回该路径"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+        output_file = tmp_path / "output.mp4"
+
+        fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg"),
+        ):
+            result = random_edge_crop(input_file, output_file)
+
+        assert result == output_file
+
+    def test_skip_when_invalid_resolution(self, tmp_path):
+        """无法获取有效分辨率时跳过裁剪"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        fake_info = {"width": 0, "height": 0, "duration": 10, "fps": 30}
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
+        ):
+            result = random_edge_crop(input_file)
+
+        assert result == input_file
+        mock_ffmpeg.assert_not_called()
+
+
+class TestRandomEdgeCropFFmpeg:
+    """FFmpeg 调用参数验证"""
+
+    def test_ffmpeg_crop_and_scale_filter(self, tmp_path):
+        """生成的 ffmpeg 滤镜包含 crop + scale"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        # 固定随机值以便验证
+        fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
+
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
+            patch("random.uniform", side_effect=[0.03, 0.03, 0.03, 0.03]),
+        ):
+            random_edge_crop(input_file)
+
+        mock_ffmpeg.assert_called_once()
+        cmd = mock_ffmpeg.call_args[0][0]
+        # 找到 -vf 参数
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "crop=" in vf_value
+        assert "scale=1000:1000" in vf_value
+
+    def test_crop_amounts_within_range(self, tmp_path):
+        """裁剪量在 2%~5% 范围内"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
+
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
+            patch("random.uniform", side_effect=[0.02, 0.05, 0.02, 0.05]),
+        ):
+            random_edge_crop(input_file)
+
+        cmd = mock_ffmpeg.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        # crop_top=20, crop_bottom=50, crop_left=20, crop_right=50
+        # new_w = 1000-20-50 = 930, new_h = 1000-20-50 = 930
+        # x_offset = 20, y_offset = 20
+        assert "crop=930:930:20:20" in vf_value
+
+    def test_uses_libx264_codec(self, tmp_path):
+        """使用 libx264 编码"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
+        ):
+            random_edge_crop(input_file)
+
+        cmd = mock_ffmpeg.call_args[0][0]
+        assert "-c:v" in cmd
+        assert cmd[cmd.index("-c:v") + 1] == "libx264"
+
+
+class TestRandomEdgeCropErrorHandling:
+    """错误处理测试"""
+
+    def test_ffmpeg_failure_raises_exception(self, tmp_path):
+        """ffmpeg 失败时抛出异常"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch(
+                "video_processing.ffmpeg_utils.run_ffmpeg",
+                side_effect=subprocess.CalledProcessError(1, "ffmpeg"),
+            ),
+        ):
+            with pytest.raises(subprocess.CalledProcessError):
+                random_edge_crop(input_file)
+
+    def test_probe_failure_propagates(self, tmp_path):
+        """probe_video_info 失败时异常传播"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        with patch(
+            "video_processing.ffmpeg_utils.probe_video_info",
+            side_effect=RuntimeError("probe failed"),
+        ):
+            with pytest.raises(RuntimeError, match="probe failed"):
+                random_edge_crop(input_file)
+
+
+class TestRandomEdgeCropEvenDimensions:
+    """偶数尺寸处理测试"""
+
+    def test_odd_crop_dimensions_adjusted_to_even(self, tmp_path):
+        """裁剪后尺寸为奇数时自动调整为偶数"""
+        input_file = tmp_path / "input.mp4"
+        input_file.write_bytes(b"fake video data")
+
+        # 1000 - 3 (top) - 4 (bottom) = 993 → 调整为 992
+        # 1000 - 3 (left) - 4 (right) = 993 → 调整为 992
+        fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
+
+        with (
+            patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
+            patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
+            # side_effect 控制 uniform 返回值
+            # top: 0.003*1000=3, bottom: 0.004*1000=4, left: 0.003*1000=3, right: 0.004*1000=4
+        ):
+            # 使用自定义 uniform 返回特定值
+            def fake_uniform(low, high):
+                # 返回特定百分比使得裁剪后尺寸为奇数
+                # 我们需要 crop_top=3, crop_bottom=4, crop_left=3, crop_right=4
+                return 0.0035  # 近似值
+
+            # 更简单的方式：直接 mock int(H * random.uniform(...)) 的结果
+            # 但我们直接测试最终 crop 滤镜即可
+            with patch("random.uniform", side_effect=[0.021, 0.022, 0.021, 0.022]):
+                random_edge_crop(input_file)
+
+        cmd = mock_ffmpeg.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        # 提取 crop 参数并验证都是偶数
+        import re
+
+        crop_match = re.search(r"crop=(\d+):(\d+)", vf_value)
+        assert crop_match
+        crop_w = int(crop_match.group(1))
+        crop_h = int(crop_match.group(2))
+        assert crop_w % 2 == 0, f"crop width {crop_w} should be even"
+        assert crop_h % 2 == 0, f"crop height {crop_h} should be even"