111 lines
3.5 KiB
Python
Executable File
111 lines
3.5 KiB
Python
Executable File
"""媒体文件有效性校验与元数据解析工具。
|
||
|
||
从 worker ingest 任务中抽取的纯逻辑模块，包含：
|
||
- FPS 解析：从分数格式字符串（如 30000/1001）安全解析帧率
|
||
- 媒体有效性校验：根据元数据判断视频/音频/图片文件是否有效
|
||
- 常量定义：最小文件大小、支持的视频编码白名单
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
# 最小有效文件大小（字节）：小于此值的直接判为无效，避免文本/空文件伪装成媒体
|
||
MIN_VIDEO_FILE_SIZE = 1024 # 1KB
|
||
MIN_AUDIO_FILE_SIZE = 100 # 100B
|
||
MIN_IMAGE_FILE_SIZE = 100 # 100B
|
||
|
||
# 支持的视频编码格式（白名单，尽可能放宽）
|
||
# 渲染引擎会在 concat 前统一转码为 h264，因此只要 ffprobe 能识别的视频编码都允许 ingested
|
||
SUPPORTED_VIDEO_CODECS: frozenset[str] = frozenset(
|
||
{
|
||
"h264",
|
||
"avc1",
|
||
"avc", # H.264 / AVC
|
||
"hevc",
|
||
"h265",
|
||
"hev1",
|
||
"hvc1", # H.265 / HEVC
|
||
"vp9",
|
||
"vp09", # VP9
|
||
"av1",
|
||
"av01", # AV1
|
||
"vp8",
|
||
"vp08", # VP8
|
||
"mpeg4",
|
||
"mp4v", # MPEG-4
|
||
"mpeg2video",
|
||
"mpg2", # MPEG-2
|
||
"wmv2",
|
||
"wmv1",
|
||
"vc1", # WMV / VC-1
|
||
"flv1",
|
||
"flv",
|
||
"vp6f", # Flash / FLV
|
||
"theora",
|
||
"ogg", # Theora
|
||
"prores",
|
||
"prores_ks",
|
||
"apcn",
|
||
"apch",
|
||
"apco",
|
||
"apcs",
|
||
"ap4h",
|
||
"ap4x", # Apple ProRes
|
||
"dnxhd",
|
||
"dnxhr", # DNxHD / DNxHR
|
||
}
|
||
)
|
||
|
||
|
||
def safe_parse_fps(fps_str: str) -> float:
|
||
"""Safely parse fps from a fraction string like "30/1" or "30000/1001".
|
||
|
||
Args:
|
||
fps_str: FPS 字符串，支持小数格式（"30.0"）或分数格式（"30000/1001"）
|
||
|
||
Returns:
|
||
解析得到的帧率浮点数；解析失败或分母为0时返回 0.0
|
||
"""
|
||
try:
|
||
if "/" in fps_str:
|
||
num, den = fps_str.split("/", 1)
|
||
den_val = float(den)
|
||
if den_val == 0:
|
||
return 0.0
|
||
return float(num) / den_val
|
||
return float(fps_str)
|
||
except (ValueError, ZeroDivisionError):
|
||
return 0.0
|
||
|
||
|
||
def is_valid_media(metadata: dict, media_type: str) -> bool:
|
||
"""根据元数据判断文件是否为有效媒体文件。
|
||
|
||
Args:
|
||
metadata: 媒体元数据字典，可能包含 size_bytes / duration / codec / width / height 等
|
||
media_type: 媒体类型（video / audio / image）
|
||
|
||
Returns:
|
||
True 表示文件有效
|
||
"""
|
||
size = int(metadata.get("size_bytes", 0))
|
||
|
||
if media_type == "video":
|
||
duration = float(metadata.get("duration", 0))
|
||
if size < MIN_VIDEO_FILE_SIZE or duration <= 0:
|
||
return False
|
||
# 编码格式校验：只排除明确非视频的编码格式，只要 ffprobe 能识别的视频编码都允许
|
||
# 渲染引擎会在 concat 前统一转码为 h264 yuv420p，ingest 层不再做严格的编码拦截
|
||
codec = str(metadata.get("codec", "")).lower()
|
||
if codec and codec not in SUPPORTED_VIDEO_CODECS:
|
||
# 非白名单编码仍允许通过，仅记录日志（调用方负责日志）
|
||
pass
|
||
return True
|
||
if media_type == "audio":
|
||
duration = float(metadata.get("duration", 0))
|
||
return size >= MIN_AUDIO_FILE_SIZE and duration > 0
|
||
if media_type == "image":
|
||
width = int(metadata.get("width", 0))
|
||
height = int(metadata.get("height", 0))
|
||
return size >= MIN_IMAGE_FILE_SIZE and width > 0 and height > 0
|
||
return False
|