feat: 视频渲染后随机边缘裁剪 2-5% 降重 #1664 (#1682)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-09-04 11:56:18 +08:00
parent 0542654ca8
commit f523548eee
3 changed files with 353 additions and 0 deletions
@@ -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