@@ -1,6 +1,8 @@
""" 视频封面抽帧工具 — 从视频中抽取帧作为封面，支持标题文字叠加。
统一封面管道：
统一封面管道（ ：
- 默认路径：本地 ffmpeg -ss 单帧 seek 抽取 + cv2 质量评分（清晰度/亮度/色彩），1-2s 完成
- 可选 MediaKit 路径：配置 MEDIAKIT_COVER_ENABLED=true 时启用火山 MediaKit SceneChange 抽帧
- 从已渲染视频抽帧：标题已通过 ASS 字幕烧进视频，帧天然带标题，无需再叠加。
- 从源素材抽帧（API E2 兜底）：源素材无标题，通过 Pillow 在帧上绘制标题文字。
"""
@@ -10,12 +12,10 @@ from __future__ import annotations
import logging
import tempfile
from pathlib import Path
from typing import Optional
logger = logging . getLogger ( __name__ )
# ── 标题叠加（Pillow）──────────────────────────────────────────────────────
# 实现统一放在 packages/shared/title_overlay.py，
def apply_title_overlay (
image_path : str ,
@@ -27,11 +27,7 @@ def apply_title_overlay(
margin_ratio : float = 0.06 ,
stroke_width_ratio : float = 0.04 ,
) - > str :
""" 在图片上绘制标题文字（指定颜色 + 黑色描边/阴影）。
委托给 packages.shared.title_overlay.apply_title_to_image，
保持 Worker 内调用方式不变。title_text 为空时直接返回原路径。
"""
""" 在图片上绘制标题文字（指定颜色 + 黑色描边/阴影）。 """
from packages . shared . title_overlay import apply_title_to_image
if not title_text or not title_text . strip ( ) :
@@ -56,26 +52,19 @@ def extract_first_frame(
height : int = - 1 ,
timeout : int = 30 ,
seek_ratio : float = 0.15 ,
seek_seconds : float | None = None ,
min_seek_seconds : float = 1.0 ,
) - > str :
""" 抽取视频封面帧（默认取视频时长 15 % 处的帧，避开片头纯色画面 ）。
因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。
""" 抽取视频封面帧（ffmpeg -ss 单帧 seek， ）。
Args:
video_path: 视频文件路径
output_path: 输出图片路径，不传则用临时文件
width: 输出宽度 （默认 -1， 保持原始分辨率）
height: 输出高度（默认 -1，保持原始分辨率 ）
timeout: 超时时间（秒）
seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15 % 处）
min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
Returns:
生成的封面帧文件路径
Raises:
RuntimeError: ffmpeg 执行失败或输出文件为空
width/height : 输出宽高 （默认保持原始分辨率）
timeout: 超时（秒 ）
seek_ratio: 抽帧位置占视频时长的比例
seek_seconds: 指定具体抽帧时间点（秒），优先于 seek_ratio
min_seek_seconds: 最小抽帧时间
"""
from video_processing . ffmpeg_utils import FFMPEG_BIN , probe_duration , run_ffmpeg
@@ -87,31 +76,25 @@ def extract_first_frame(
_is_temp_output = True
try :
# 计算抽帧时间点：取视频时长 * seek_ratio，最少 min_seek_seconds 秒
try :
duration = probe_duration ( video_path )
seek_time = max ( min_seek_seconds , duration * seek_ratio )
except Exception :
# probe 失败时 fallback 到第1秒
seek_time = min_seek_seconds
if seek_seconds is not None :
seek_time = max ( 0.0 , float ( seek_seconds ) )
else :
try :
duration = probe_duration ( video_path )
seek_time = max ( min_seek_seconds , duration * seek_ratio )
except Exception :
seek_time = min_seek_seconds
# 格式化为 HH:MM:SS.xx
seek_str = _format_seek_time ( seek_time )
# 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率。
# NOTE: scale_filter 在此处通过 if/else 分支赋值，之后不再被覆盖，
# 后续 cmd / cmd2 均复用同一变量，逻辑无变化。
if width > 0 or height > 0 :
w_str = str ( width ) if width > 0 else " -1 "
h_str = str ( height ) if height > 0 else " -1 "
scale_filter = f " scale= { w_str } : { h_str } :force_original_aspect_ratio=decrease,format=yuvj420p "
else :
# 保持原始分辨率，只确保格式兼容
scale_filter = " format=yuvj420p "
# -ss 放在 -i 前面（input seeking， 更 快）
# -vframes 1 只取一帧
# -q:v 2 jpeg 高质量
# -ss 放在 -i 前面（input seeking， 极 快），
cmd = [
FFMPEG_BIN ,
" -y " ,
@@ -154,7 +137,6 @@ def extract_first_frame(
return output_path
except Exception :
# 失败时清理自己创建的临时文件
if _is_temp_output and output_path :
try :
Path ( output_path ) . unlink ( missing_ok = True )
@@ -164,7 +146,6 @@ def extract_first_frame(
def _format_seek_time ( seconds : float ) - > str :
""" 将秒数格式化为 HH:MM:SS.xx 格式。 """
h = int ( seconds / / 3600 )
m = int ( ( seconds % 3600 ) / / 60 )
s = seconds % 60
@@ -177,19 +158,7 @@ def generate_and_upload_thumbnail(
* ,
seek_ratio : float = 0.15 ,
) - > str :
""" 从视频中提取一帧缩略图并上传到 OSS。
Args:
video_path: 视频文件路径
storage_key: OSS 存储 key
seek_ratio: 抽帧位置比例（默认 0.15）
Returns:
上传后的 URL 字符串
Raises:
RuntimeError: 抽帧或上传失败
"""
""" 从视频中提取一帧缩略图并上传到 OSS。 """
from video_processing . oss_helpers import upload_to_oss
tmp = tempfile . NamedTemporaryFile ( suffix = " .jpg " , delete = False )
@@ -204,24 +173,84 @@ def generate_and_upload_thumbnail(
Path ( tmp . name ) . unlink ( missing_ok = True )
def _compute_clip_boundary_seek_points (
duration : float ,
clip_boundaries : Optional [ list [ tuple [ float , float ] ] ] = None ,
num_frames : int = 5 ,
head_skip_ratio : float = 0.08 ,
tail_skip_ratio : float = 0.08 ,
) - > list [ float ] :
""" 基于clip分段边界计算抽帧时间点（取每段中间帧，效果比均匀抽更好）。
策略：
- 如果传入 clip_boundaries（每个元素是 (clip_start_in_timeline, clip_duration)），
取每个片段的中点作为抽帧候选点
- 候选点不足 num_frames 时，均匀补充
- 跳过片头 head_skip_ratio（ % ，避免片头黑屏/开场标题）和片尾 tail_skip_ratio（ % ）
- 返回按时间排序的 num_frames 个抽帧点（秒）
"""
if duration < = 0 :
# 无法probe，均匀分布兜底
return [ max ( 1.0 , duration * ( 0.1 + 0.8 * i / max ( num_frames - 1 , 1 ) ) ) for i in range ( num_frames ) ]
head_skip = duration * head_skip_ratio
tail_skip = duration * tail_skip_ratio
valid_start = head_skip
valid_end = max ( valid_start + 1.0 , duration - tail_skip )
candidates : list [ float ] = [ ]
if clip_boundaries :
# 累加timeline start，取每clip中点
cur = 0.0
for _clip_start , clip_dur in clip_boundaries :
if clip_dur < = 0 :
continue
mid = cur + clip_dur / 2.0
if valid_start < = mid < = valid_end :
candidates . append ( mid )
cur + = clip_dur
# 去重+排序
candidates = sorted ( set ( round ( c , 3 ) for c in candidates ) )
# 如果候选点不足，均匀补充
if len ( candidates ) < num_frames :
needed = num_frames - len ( candidates )
existing = set ( round ( c , 1 ) for c in candidates )
for i in range ( needed * 3 ) :
ratio = 0.1 + 0.8 * ( i + 0.5 ) / ( needed * 3 )
t = valid_start + ( valid_end - valid_start ) * ratio
if round ( t , 1 ) not in existing :
candidates . append ( t )
existing . add ( round ( t , 1 ) )
if len ( candidates ) > = num_frames :
break
# 如果还不够，强制均匀
while len ( candidates ) < num_frames :
idx = len ( candidates )
ratio = 0.1 + 0.8 * idx / max ( num_frames - 1 , 1 )
candidates . append ( valid_start + ( valid_end - valid_start ) * ratio )
candidates . sort ( )
# 如果超过num_frames，均匀选取
if len ( candidates ) > num_frames :
step = len ( candidates ) / num_frames
candidates = [ candidates [ int ( i * step ) ] for i in range ( num_frames ) ]
return [ round ( t , 3 ) for t in candidates [ : num_frames ] ]
def _extract_frames_via_mediakit (
video_path : str ,
plan_id : str ,
num_frames : int ,
) - > list [ dict ] | None :
""" 使用 MediaKit 智能抽帧 API 提取封面帧。
Args:
video_path: 本地视频文件路径
plan_id: 编辑计划 ID
num_frames: 需要的帧数
Returns:
帧列表 [ { " image_url " : str, " timestamp " : float}, ...]，失败返回 None
"""
""" 使用 MediaKit 智能抽帧 API 提取封面帧（fallback 路径，默认不启用）。 """
import uuid
from video_processing . oss_helpers import get_signed_download_url , upload_to_oss
from video_processing . oss_helpers import delete_from_oss , get_signed_download_url , upload_to_oss
from packages . shared . mediakit_client import get_mediakit_client
@@ -231,47 +260,37 @@ def _extract_frames_via_mediakit(
return None
video_storage_key : str = " "
# 1. 上传视频到 OSS，并生成预签名下载 URL（bucket 私有读，公网 URL 会 403）
try :
video_storage_key = f " temp/ { plan_id } / { uuid . uuid4 ( ) . hex [ : 8 ] } _ { Path ( video_path ) . name } "
public_url = upload_to_oss ( video_path , video_storage_key )
if not public_url :
logger . warning ( " [thumbnail] 视频上传 OSS 失败，无法使用 MediaKit " )
return None
# MediaKit 从公网拉取视频，必须使用预签名 URL；签名 1h 足够完成抽帧
video_url = get_signed_download_url ( video_storage_key , expires_seconds = 3600 ) or public_url
logger . info ( " [thumbnail] 视频已上传 OSS 并生成签名 URL: key= %s " , video_storage_key [ : 80 ] )
except Exception as e :
logger . warning ( " [thumbnail] 视频上传 OSS 异常: %s ，降级到 ffmpeg " , e )
logger . warning ( " [thumbnail] 视频上传 OSS 异常: %s ，降级到本地 ffmpeg " , e )
return None
# 2. 调用 MediaKit 智能抽帧
try :
frames = client . extract_frames (
video_url = video_url ,
strategy = " SceneChange " ,
max_frames = num_frames * 2 , # 多取一些帧供选择
max_frames = num_frames * 2 ,
)
if not frames :
logger . warning ( " [thumbnail] MediaKit 抽帧返回空，降级到 ffmpeg " )
logger . warning ( " [thumbnail] MediaKit 抽帧返回空 " )
return None
# 选取最均匀的 num_frames 个帧
if len ( frames ) > num_frames :
step = len ( frames ) / / num_frames
frames = [ frames [ i * step ] for i in range ( num_frames ) ]
logger . info ( " [thumbnail] MediaKit 抽帧成功: %d 帧 " , len ( frames ) )
return frames
except Exception as e :
logger . warning ( " [thumbnail] MediaKit 抽帧异常: %s ，降级到 ffmpeg " , e )
logger . warning ( " [thumbnail] MediaKit 抽帧异常: %s " , e )
return None
finally :
# 清理临时视频文件
try :
from video_processing . oss_helpers import delete_from_oss
delete_from_oss ( video_storage_key )
except Exception :
pass
@@ -282,100 +301,96 @@ def extract_and_upload_cover_frames(
plan_id : str ,
* ,
task_id : str = " " ,
num_frames : int = 5 , # 抽 5 帧候选，通过质量评分选出最佳帧
num_frames : int = 5 ,
title_text : str = " " ,
title_color : str = " #ffffff " ,
title_position : str = " bottom " ,
title_font_size : int | None = None ,
clip_boundaries : Optional [ list [ tuple [ float , float ] ] ] = None ,
) - > list [ dict ] :
""" 从视频中抽取多帧作为封面候选，通过质量评分选出最佳帧，上传到 OSS。
流程：
1. 优先使用 MediaKit 智能抽帧（多抽一些供选择）
2. MediaKit 不足时降级到 ffmpeg 均匀抽帧
3. 对所有候选帧进行质量评分（清晰度/亮度/色彩丰富度）
4. 按分数从高到低排序返回
默认路径（P1优化）：本地 ffmpeg 单帧 seek 抽帧 + cv2 评分，预期 <2s 完成。
- 基于 clip 分段边界取各段中间帧（clip_boundaries 参数），效果优于均匀抽帧
- 无边界信息时均匀分布（10 % ~90 % 之间）
- 所有帧本地 cv2 清晰度/亮度/色彩三维评分，最高分自动选出
Fallback（
Args:
video_path: 视频文件路径
plan_id: 编辑计划 ID（用于生成 storage key）
task_id: 任务 ID（用于生成独立的 storage key，避免标题变更时封面冲突）
num_frames: 抽取候选帧数（默认 5，通过质量评分选出最佳帧）
title_text: 标题文字；非空时用 Pillow 叠加到每帧。
从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。
title_color: 标题字体颜色（#RRGGBB）
title_position: 标题位置 top/center/bottom
title_font_size: 标题字号，None 时自动计算
Returns:
封面候选列表（按质量分数降序），每项包含 { " url " : str, " position " : float, " score " : float}
clip_boundaries: 片段边界列表 [(clip_start, clip_duration), ...]，用于智能取点
"""
import time
import httpx
from video_processing . ffmpeg_utils import probe_duration
from video_processing . oss_helpers import upload_to_oss
from packages . shared . config import get_shared_settings
t0 = time . monotonic ( )
try :
duration = probe_duration ( video_path )
except Exception :
duration = 0.0
candidates : list [ dict ] = [ ]
_temp_paths : list [ str ] = [ ] # 收集所有临时文件路径，最后统一清理
_temp_paths : list [ str ] = [ ]
try :
# ── 阶段 1：抽帧 ──────────────────────────────────────────────
# 优先尝试 MediaKit 智能抽帧
mediakit_frames = _extract_frames_via_mediakit ( video_path , plan_id , num_frames )
if mediakit_frames :
for i , frame in enumerate ( mediakit_frames ) :
frame_url = frame . get ( " image_url " )
if not frame_url :
continue
tmp = tempfile . NamedTemporaryFile ( suffix = " .jpg " , delete = False )
tmp . close ( )
_temp_paths . append ( tmp . name )
try :
# 下载 MediaKit 返回的帧图
resp = httpx . get ( frame_url , timeout = 30 , follow_redirects = True )
resp . raise_for_status ( )
with open ( tmp . name , " wb " ) as f :
f . write ( resp . content )
settings = get_shared_settings ( )
use_mediakit = getattr ( settings , " mediakit_cover_enabled " , False )
# 叠加标题文字（如需要）
if title_text and title_text . strip ( ) :
apply_title_overlay (
tmp . name ,
title_text ,
color = title_color ,
position = title_position ,
font_size = title_font_size ,
)
if use_mediakit :
logger . info ( " [thumbnail] MEDIAKIT_COVER_ENABLED=true，走 MediaKit 路径 " )
mediakit_frames = _extract_frames_via_mediakit ( video_path , plan_id , num_frames )
if mediakit_frames :
for i , frame in enumerate ( mediakit_frames ) :
frame_url = frame . get ( " image_url " )
if not frame_url :
continue
tmp = tempfile . NamedTemporaryFile ( suffix = " .jpg " , delete = False )
tmp . close ( )
_temp_paths . append ( tmp . name )
try :
resp = httpx . get ( frame_url , timeout = 30 , follow_redirects = True )
resp . raise_for_status ( )
with open ( tmp . name , " wb " ) as f :
f . write ( resp . content )
if title_text and title_text . strip ( ) :
apply_title_overlay (
tmp . name ,
title_text ,
color = title_color ,
position = title_position ,
font_size = title_font_size ,
)
storage_key = f " covers/ { plan_id } / { task_id } /mediakit_frame_ { i } .jpg "
url = upload_to_oss ( tmp . name , storage_key )
if url :
candidates . append (
{
" url " : url ,
" position " : round ( frame . get ( " timestamp " , 0.0 ) , 2 ) ,
" image_path " : tmp . name ,
}
)
except Exception as e :
logger . warning ( " [thumbnail] MediaKit 帧 %d 处理失败: %s " , i , e )
if len ( candidates ) > = num_frames :
logger . info ( " [thumbnail] MediaKit 抽帧完成: %d 帧 " , len ( candidates ) )
storage_key = f " covers/ { plan_id } / { task_id } /mediakit_frame_ { i } .jpg "
url = upload_to_oss ( tmp . name , storage_key )
if url :
seek_time = frame . get ( " timestamp " , 0.0 )
candidates . append (
{
" url " : url ,
" position " : round ( seek_time , 2 ) ,
" image_path " : tmp . name ,
}
)
except Exception as e :
logger . warning ( " [thumbnail] MediaKit 帧 %d 处理失败: %s " , i , e )
if len ( candidates ) > = num_frames :
logger . info ( " [thumbnail] MediaKit 智能抽帧完成: %d 帧 " , len ( candidates ) )
else :
logger . warning ( " [thumbnail] MediaKit 抽帧不足 %d 帧，降级到 ffmpeg " , num_frames )
# Fallback: ffmpeg 直接抽帧（仅当 MediaKit 不足时）
# ── 默认路径：本地 ffmpeg 单帧 seek ───────────────────────────
if len ( candidates ) < num_frames :
logger . info ( " [thumbnail] 使用 ffmpeg 抽帧补充 " )
# 均匀分布抽帧点：从 10% 到 90%
for i in range ( num_frames ) :
ratio = 0.1 + 0.8 * i / max ( num_frames - 1 , 1 )
if candidates :
logger . info ( " [thumbnail] MediaKit 不足 %d 帧，本地 ffmpeg 补充 " , num_frames )
else :
logger . info ( " [thumbnail] 使用本地 ffmpeg 抽帧（num= %d , duration= %.1f s） " , num_frames , duration )
seek_points = _compute_clip_boundary_seek_points ( duration , clip_boundaries , num_frames )
for i , seek_t in enumerate ( seek_points ) :
tmp = tempfile . NamedTemporaryFile ( suffix = " .jpg " , delete = False )
tmp . close ( )
_temp_paths . append ( tmp . name )
@@ -383,10 +398,9 @@ def extract_and_upload_cover_frames(
frame_path = extract_first_frame (
video_path ,
output_path = tmp . name ,
seek_ratio = ratio ,
seek_seconds = seek_t ,
min_seek_seconds = 0.5 ,
)
# 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
if title_text and title_text . strip ( ) :
apply_title_overlay (
frame_path ,
@@ -398,11 +412,10 @@ def extract_and_upload_cover_frames(
storage_key = f " covers/ { plan_id } / { task_id } /frame_ { i } .jpg "
url = upload_to_oss ( frame_path , storage_key )
if url :
seek_time = max ( 0.5 , duration * ratio ) if duration > 0 else 0.0
candidates . append (
{
" url " : url ,
" position " : round ( seek_time , 2 ) ,
" position " : seek_t ,
" image_path " : tmp . name ,
}
)
@@ -415,28 +428,23 @@ def extract_and_upload_cover_frames(
from packages . shared . cover_frame_scorer import score_frames
candidates = score_frames ( candidates )
elapsed = time . monotonic ( ) - t0
logger . info (
" [thumbnail] 封面帧质量 评分完成: plan_id= %s count= %d best_score= %.1f " ,
" [thumbnail] 封面帧评分完成: plan_id= %s count= %d best_score= %.1f elapsed= %.2f s path= %s " ,
plan_id ,
len ( candidates ) ,
candidates [ 0 ] . get ( " score " , 0.0 ) if candidates else 0.0 ,
elapsed ,
" mediakit " if use_mediakit else " local " ,
)
except Exception :
logger . warning (
" [thumbnail] 封面帧质量评分失败，保持原始顺序: plan_id= %s " ,
plan_id ,
exc_info = True ,
)
logger . warning ( " [thumbnail] 封面帧质量评分失败，保持原始顺序 " , exc_info = True )
# ── 阶段 3：清理临时文件 ────────────────────────────────────────
# 移除 image_path（不再需要），但临时文件统一清理
for c in candidates :
c . pop ( " image_path " , None )
return candidates
finally :
# 统一清理所有临时文件
for path in _temp_paths :
try :
Path ( path ) . unlink ( missing_ok = True )