diff --git a/app/eval_music.py b/app/eval_music.py new file mode 100644 index 0000000..86f28fa --- /dev/null +++ b/app/eval_music.py @@ -0,0 +1,573 @@ +""" +音乐生成质量评估 - 多维度评估框架 + +评估维度： +1. 节奏质量 (15%) - BPM 稳定性、节奏型 +2. 和弦进行 (10%) - 和弦丰富度、和谐度 +3. 听感质量 (20%) - 整体音质、混音平衡 +4. 情绪-歌词匹配 (20%) - 情绪与歌词主题一致性 +5. 指令匹配度 (15%) - 与生成 prompt 的匹配度 +6. 伴奏质量 (10%) - 伴奏清晰度、乐器质量 +7. 人声质量 (10%) - 人声清晰度、音准 + +工具： +- Demucs: 人声/伴奏分离 +- ffprobe: 音频元数据、响度分析 +- ffmpeg: 音频分析滤镜 (astats, loudnorm) +- LLM: 语义理解、情绪分析 +""" + +import asyncio +import json +import logging +import os +import tempfile +from typing import Dict, Tuple, Optional + +logger = logging.getLogger(__name__) + +# GPU 服务器配置（复用 ktv_adapter 的配置） +GPU_HOST = "ymq@opencomputing.net" +GPU_DEMUCS_VENV = "/data/ymq/demucs_venv" + + +async def analyze_audio_metadata(audio_path: str) -> Dict: + """ + 使用 ffprobe 提取音频元数据 + + 返回: + { + "duration": float, + "sample_rate": int, + "channels": int, + "bitrate": int, + "codec": str + } + """ + try: + proc = await asyncio.create_subprocess_exec( + "ffprobe", + "-v", "error", + "-show_entries", "stream=sample_rate,channels,codec_name,bit_rate", + "-show_entries", "format=duration,bit_rate", + "-of", "json", + audio_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + logger.warning(f"ffprobe failed: {stderr.decode()}") + return {} + + data = json.loads(stdout.decode()) + stream = data.get("streams", [{}])[0] + fmt = data.get("format", {}) + + return { + "duration": float(fmt.get("duration", 0)), + "sample_rate": int(stream.get("sample_rate", 0)), + "channels": int(stream.get("channels", 0)), + "bitrate": int(fmt.get("bit_rate", 0)), + "codec": stream.get("codec_name", "unknown") + } + except Exception as e: + logger.error(f"Failed to analyze audio metadata: {e}") + return {} + + +async def analyze_audio_loudness(audio_path: str) -> Dict: + """ + 使用 ffmpeg loudnorm 分析音频响度 + + 返回: + { + "integrated_loudness": float, # 综合响度 (LUFS) + "loudness_range": float, # 响度范围 (LU) + "true_peak": float # 真峰值 (dBTP) + } + """ + try: + proc = await asyncio.create_subprocess_exec( + "ffmpeg", + "-i", audio_path, + "-af", "loudnorm=print_format=json", + "-f", "null", + "-", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + # loudnorm 输出在 stderr + output = stderr.decode() + + # 提取 JSON 部分 + import re + json_match = re.search(r'\{[^}]+\}', output) + if json_match: + data = json.loads(json_match.group()) + return { + "integrated_loudness": float(data.get("input_i", -70)), + "loudness_range": float(data.get("input_lra", 0)), + "true_peak": float(data.get("input_tp", -100)) + } + + return {} + except Exception as e: + logger.error(f"Failed to analyze audio loudness: {e}") + return {} + + +async def separate_vocals_accompaniment( + audio_path: str, + output_dir: str +) -> Tuple[Optional[str], Optional[str]]: + """ + 使用 Demucs 分离人声和伴奏 + + 返回: + (vocals_path, accompaniment_path) or (None, None) + """ + try: + # 在 GPU 服务器上运行 Demucs + gpu_audio_path = f"/tmp/eval_{os.path.basename(audio_path)}" + + # 上传音频 + upload_cmd = f"scp '{audio_path}' {GPU_HOST}:{gpu_audio_path}" + upload_proc = await asyncio.create_subprocess_shell( + upload_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await upload_proc.communicate() + + # 运行 Demucs + demucs_cmd = ( + f"ssh {GPU_HOST} '" + f"source {GPU_DEMUCS_VENV}/bin/activate && " + f"cd {GPU_DEMUCS_VENV} && " + f"python -m demucs --two-stems vocals --out /tmp/demucs_out '{gpu_audio_path}'" + f"'" + ) + demucs_proc = await asyncio.create_subprocess_shell( + demucs_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await demucs_proc.communicate() + + # 下载分离结果 + # Demucs 输出格式: /tmp/demucs_out/htdemucs/{filename}/vocals.wav + filename = os.path.splitext(os.path.basename(audio_path))[0] + gpu_vocals = f"/tmp/demucs_out/htdemucs/{filename}/vocals.wav" + gpu_no_vocals = f"/tmp/demucs_out/htdemucs/{filename}/no_vocals.wav" + + local_vocals = os.path.join(output_dir, f"{filename}_vocals.wav") + local_no_vocals = os.path.join(output_dir, f"{filename}_no_vocals.wav") + + # 下载人声 + download_vocals = f"scp {GPU_HOST}:{gpu_vocals} '{local_vocals}'" + download_proc1 = await asyncio.create_subprocess_shell( + download_vocals, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await download_proc1.communicate() + + # 下载伴奏 + download_no_vocals = f"scp {GPU_HOST}:{gpu_no_vocals} '{local_no_vocals}'" + download_proc2 = await asyncio.create_subprocess_shell( + download_no_vocals, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await download_proc2.communicate() + + # 清理 GPU 临时文件 + cleanup_cmd = f"ssh {GPU_HOST} 'rm -f {gpu_audio_path} && rm -rf /tmp/demucs_out'" + cleanup_proc = await asyncio.create_subprocess_shell( + cleanup_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await cleanup_proc.communicate() + + vocals_exists = os.path.exists(local_vocals) + no_vocals_exists = os.path.exists(local_no_vocals) + + return ( + local_vocals if vocals_exists else None, + local_no_vocals if no_vocals_exists else None + ) + + except Exception as e: + logger.error(f"Failed to separate vocals: {e}") + return None, None + + +async def evaluate_rhythm_quality(audio_path: str) -> Tuple[float, str]: + """ + 评估节奏质量 + + 检查项： + - BPM 稳定性（通过节拍分析） + - 节奏型清晰度 + + 简化实现：基于音频时长和响度范围推断 + + 返回: (score 0-1, reason) + """ + metadata = await analyze_audio_metadata(audio_path) + loudness = await analyze_audio_loudness(audio_path) + + if not metadata: + return 0.0, "无法读取音频元数据" + + duration = metadata.get("duration", 0) + lra = loudness.get("loudness_range", 0) + + # 节奏质量推断 + # 响度范围适中 (5-15 LU) 通常表示节奏稳定 + if 5 <= lra <= 15: + score = 0.85 + reason = f"节奏稳定 (LRA={lra:.1f}LU)" + elif 3 <= lra < 5 or 15 < lra <= 20: + score = 0.7 + reason = f"节奏基本稳定 (LRA={lra:.1f}LU)" + else: + score = 0.5 + reason = f"节奏波动较大 (LRA={lra:.1f}LU)" + + # 时长加分 + if duration >= 60: + score = min(1.0, score + 0.1) + reason += f", 时长充足 ({duration:.0f}s)" + + return score, reason + + +async def evaluate_chord_progression(audio_path: str) -> Tuple[float, str]: + """ + 评估和弦进行 + + 简化实现：基于音频特征推断和声丰富度 + + 返回: (score 0-1, reason) + """ + # 和弦分析需要专业的音频分析库（如 librosa 的 chroma 特征） + # 这里使用简化评估 + + metadata = await analyze_audio_metadata(audio_path) + if not metadata: + return 0.0, "无法读取音频元数据" + + duration = metadata.get("duration", 0) + + # 时长较长的音乐通常有更丰富的和声进行 + if duration >= 180: # 3分钟+ + return 0.85, f"和声进行丰富 (时长 {duration:.0f}s)" + elif duration >= 120: + return 0.75, f"和声进行合理 (时长 {duration:.0f}s)" + elif duration >= 60: + return 0.65, f"和声进行基本 (时长 {duration:.0f}s)" + else: + return 0.5, f"时长过短，和声发展有限 ({duration:.0f}s)" + + +async def evaluate_audio_quality(audio_path: str) -> Tuple[float, str]: + """ + 评估听感质量 - 整体音质、混音平衡 + + 检查项： + - 响度水平 (-24 到 -14 LUFS 为佳) + - 真峰值 (< -1 dBTP) + - 采样率和位深 + + 返回: (score 0-1, reason) + """ + metadata = await analyze_audio_metadata(audio_path) + loudness = await analyze_audio_loudness(audio_path) + + if not metadata: + return 0.0, "无法读取音频元数据" + + score = 0.0 + reasons = [] + + # 采样率评分 + sample_rate = metadata.get("sample_rate", 0) + if sample_rate >= 44100: + score += 0.2 + reasons.append(f"高采样率 {sample_rate}Hz") + elif sample_rate >= 22050: + score += 0.15 + reasons.append(f"标准采样率 {sample_rate}Hz") + else: + reasons.append(f"采样率偏低 {sample_rate}Hz") + + # 响度评分 + integrated_loudness = loudness.get("integrated_loudness", -70) + if -24 <= integrated_loudness <= -14: + score += 0.4 + reasons.append(f"响度适中 ({integrated_loudness:.1f} LUFS)") + elif -30 <= integrated_loudness < -24 or -14 < integrated_loudness <= -10: + score += 0.3 + reasons.append(f"响度可接受 ({integrated_loudness:.1f} LUFS)") + else: + score += 0.1 + reasons.append(f"响度异常 ({integrated_loudness:.1f} LUFS)") + + # 真峰值评分 + true_peak = loudness.get("true_peak", -100) + if true_peak < -1: + score += 0.2 + reasons.append(f"无削波 (TP={true_peak:.1f} dBTP)") + elif true_peak < 0: + score += 0.15 + reasons.append(f"轻微削波风险 (TP={true_peak:.1f} dBTP)") + else: + reasons.append(f"明显削波 (TP={true_peak:.1f} dBTP)") + + # 位深/码率评分 + bitrate = metadata.get("bitrate", 0) + if bitrate >= 320000: # 320kbps+ + score += 0.2 + reasons.append(f"高码率 {bitrate//1000}kbps") + elif bitrate >= 192000: + score += 0.15 + reasons.append(f"标准码率 {bitrate//1000}kbps") + else: + reasons.append(f"码率偏低 {bitrate//1000}kbps") + + return score, "; ".join(reasons) + + +async def evaluate_emotion_lyrics_match( + audio_path: str, + lyrics: Optional[str] = None +) -> Tuple[float, str]: + """ + 评估情绪-歌词匹配度 + + 使用 LLM 分析： + 1. 音频的情绪特征（通过音频特征推断） + 2. 歌词的情感主题 + 3. 两者的一致性 + + 返回: (score 0-1, reason) + """ + if not lyrics: + return 0.6, "无歌词，跳过情绪-歌词匹配评估" + + # 分析音频情绪（简化：基于响度和时长） + loudness = await analyze_audio_loudness(audio_path) + integrated_loudness = loudness.get("integrated_loudness", -70) + + # 响度较高通常表示更强烈的情绪 + if integrated_loudness > -18: + audio_emotion = "强烈/激昂" + elif integrated_loudness > -24: + audio_emotion = "中等强度" + else: + audio_emotion = "柔和/舒缓" + + # TODO: 使用 LLM 分析歌词情感并与音频情绪比较 + # 这里返回模拟结果 + + return 0.8, f"情绪-歌词匹配评估（音频情绪: {audio_emotion}，待集成 LLM 歌词分析）" + + +async def evaluate_prompt_adherence( + audio_path: str, + prompt: Optional[str] = None +) -> Tuple[float, str]: + """ + 评估指令匹配度 - 与生成 prompt 的匹配度 + + 使用 LLM 分析： + 1. prompt 描述的风格、情绪、乐器等 + 2. 音频是否体现这些特征 + + 返回: (score 0-1, reason) + """ + if not prompt: + return 0.6, "无 prompt，跳过指令匹配度评估" + + # TODO: 使用 LLM 多模态分析音频与 prompt 的匹配度 + # 这里返回模拟结果 + + return 0.8, f"指令匹配度评估（待集成 LLM 分析）" + + +async def evaluate_accompaniment_quality( + accompaniment_path: Optional[str] +) -> Tuple[float, str]: + """ + 评估伴奏质量 + + 检查项： + - 伴奏清晰度 + - 乐器质量 + - 混音平衡 + + 返回: (score 0-1, reason) + """ + if not accompaniment_path or not os.path.exists(accompaniment_path): + return 0.5, "伴奏文件不存在，跳过伴奏质量评估" + + # 分析伴奏响度 + loudness = await analyze_audio_loudness(accompaniment_path) + integrated_loudness = loudness.get("integrated_loudness", -70) + + # 伴奏响度适中为佳 + if -22 <= integrated_loudness <= -16: + score = 0.85 + reason = f"伴奏混音平衡 ({integrated_loudness:.1f} LUFS)" + elif -28 <= integrated_loudness < -22 or -16 < integrated_loudness <= -12: + score = 0.7 + reason = f"伴奏混音可接受 ({integrated_loudness:.1f} LUFS)" + else: + score = 0.5 + reason = f"伴奏响度异常 ({integrated_loudness:.1f} LUFS)" + + return score, reason + + +async def evaluate_vocal_quality( + vocals_path: Optional[str] +) -> Tuple[float, str]: + """ + 评估人声质量 + + 检查项： + - 人声清晰度 + - 音准（简化评估） + - 响度平衡 + + 返回: (score 0-1, reason) + """ + if not vocals_path or not os.path.exists(vocals_path): + return 0.5, "人声文件不存在，跳过人声质量评估" + + # 分析人声响度 + loudness = await analyze_audio_loudness(vocals_path) + integrated_loudness = loudness.get("integrated_loudness", -70) + + # 人声响度通常高于伴奏 + if -20 <= integrated_loudness <= -12: + score = 0.85 + reason = f"人声清晰 ({integrated_loudness:.1f} LUFS)" + elif -26 <= integrated_loudness < -20 or -12 < integrated_loudness <= -8: + score = 0.7 + reason = f"人声可接受 ({integrated_loudness:.1f} LUFS)" + else: + score = 0.5 + reason = f"人声响度异常 ({integrated_loudness:.1f} LUFS)" + + return score, reason + + +async def evaluate_music_quality( + audio_path: str, + prompt: Optional[str] = None, + lyrics: Optional[str] = None, + use_demucs: bool = True +) -> Dict: + """ + 音乐质量综合评估 + + 参数: + audio_path: 音频文件路径 + prompt: 生成时的描述（可选） + lyrics: 歌词文本（可选） + use_demucs: 是否使用 Demucs 分离人声/伴奏 + + 返回: + { + "overall_score": float (0-1), + "overall_reason": str, + "dimensions": { + "rhythm_quality": {"score": float, "weight": 0.15, "reason": str}, + "chord_progression": {"score": float, "weight": 0.10, "reason": str}, + "audio_quality": {"score": float, "weight": 0.20, "reason": str}, + "emotion_lyrics_match": {"score": float, "weight": 0.20, "reason": str}, + "prompt_adherence": {"score": float, "weight": 0.15, "reason": str}, + "accompaniment_quality": {"score": float, "weight": 0.10, "reason": str}, + "vocal_quality": {"score": float, "weight": 0.10, "reason": str} + } + } + """ + # 基础检查 + if not os.path.exists(audio_path): + return { + "overall_score": 0.0, + "overall_reason": f"音频文件不存在: {audio_path}", + "dimensions": {} + } + + file_size = os.path.getsize(audio_path) + if file_size < 51200: # < 50KB + return { + "overall_score": 0.0, + "overall_reason": f"音频文件过小: {file_size} bytes", + "dimensions": {} + } + + # 分离人声和伴奏（如果启用） + vocals_path = None + accompaniment_path = None + + if use_demucs: + with tempfile.TemporaryDirectory() as tmpdir: + vocals_path, accompaniment_path = await separate_vocals_accompaniment( + audio_path, tmpdir + ) + + # 评估各维度 + rhythm_score, rhythm_reason = await evaluate_rhythm_quality(audio_path) + chord_score, chord_reason = await evaluate_chord_progression(audio_path) + quality_score, quality_reason = await evaluate_audio_quality(audio_path) + emotion_score, emotion_reason = await evaluate_emotion_lyrics_match(audio_path, lyrics) + prompt_score, prompt_reason = await evaluate_prompt_adherence(audio_path, prompt) + accomp_score, accomp_reason = await evaluate_accompaniment_quality(accompaniment_path) + vocal_score, vocal_reason = await evaluate_vocal_quality(vocals_path) + + else: + # 不使用 Demucs，跳过伴奏和人声评估 + rhythm_score, rhythm_reason = await evaluate_rhythm_quality(audio_path) + chord_score, chord_reason = await evaluate_chord_progression(audio_path) + quality_score, quality_reason = await evaluate_audio_quality(audio_path) + emotion_score, emotion_reason = await evaluate_emotion_lyrics_match(audio_path, lyrics) + prompt_score, prompt_reason = await evaluate_prompt_adherence(audio_path, prompt) + accomp_score, accomp_reason = 1.0, "未分离伴奏，跳过评估" + vocal_score, vocal_reason = 1.0, "未分离人声，跳过评估" + + # 构建维度评估 + dimensions = { + "rhythm_quality": {"score": rhythm_score, "weight": 0.15, "reason": rhythm_reason}, + "chord_progression": {"score": chord_score, "weight": 0.10, "reason": chord_reason}, + "audio_quality": {"score": quality_score, "weight": 0.20, "reason": quality_reason}, + "emotion_lyrics_match": {"score": emotion_score, "weight": 0.20, "reason": emotion_reason}, + "prompt_adherence": {"score": prompt_score, "weight": 0.15, "reason": prompt_reason}, + "accompaniment_quality": {"score": accomp_score, "weight": 0.10, "reason": accomp_reason}, + "vocal_quality": {"score": vocal_score, "weight": 0.10, "reason": vocal_reason}, + } + + # 加权计算总分 + overall_score = sum(d["score"] * d["weight"] for d in dimensions.values()) + + # 生成综合原因 + reasons = [] + for dim_name, dim_data in dimensions.items(): + if dim_data["score"] < 0.6: + reasons.append(f"{dim_name}: {dim_data['reason']}") + + overall_reason = "; ".join(reasons) if reasons else "各维度评估良好" + + return { + "overall_score": overall_score, + "overall_reason": overall_reason, + "dimensions": dimensions + } diff --git a/app/eval_video.py b/app/eval_video.py new file mode 100644 index 0000000..bfc5eea --- /dev/null +++ b/app/eval_video.py @@ -0,0 +1,409 @@ +""" +视频生成质量评估 - 多维度评估框架 + +评估维度： +1. 视觉质量 (25%) - 分辨率、清晰度、色彩 +2. 运动流畅度 (20%) - 帧率稳定性、运动伪影 +3. 语义一致性 (30%) - 视频内容与 prompt 描述的匹配度 +4. 角色一致性 (15%) - Ref2V 生成时角色特征保持度 +5. 时间连贯性 (10%) - 无闪烁、跳变 + +工具： +- ffprobe: 视频元数据、帧率 +- ffmpeg: 帧提取、质量分析 +- LLM 多模态: 语义理解、角色匹配 +""" + +import asyncio +import json +import logging +import os +import tempfile +from typing import Dict, Tuple, Optional + +logger = logging.getLogger(__name__) + + +async def analyze_video_metadata(video_path: str) -> Dict: + """ + 使用 ffprobe 提取视频元数据 + + 返回: + { + "width": int, + "height": int, + "fps": float, + "duration": float, + "codec": str, + "bitrate": int + } + """ + try: + proc = await asyncio.create_subprocess_exec( + "ffprobe", + "-v", "error", + "-select_streams", "v:0", + "-show_entries", "stream=width,height,r_frame_rate,codec_name,bit_rate", + "-show_entries", "format=duration", + "-of", "json", + video_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + logger.warning(f"ffprobe failed: {stderr.decode()}") + return {} + + data = json.loads(stdout.decode()) + stream = data.get("streams", [{}])[0] + fmt = data.get("format", {}) + + # 解析帧率 + fps_str = stream.get("r_frame_rate", "30/1") + if "/" in fps_str: + num, den = fps_str.split("/") + fps = float(num) / float(den) if float(den) > 0 else 30.0 + else: + fps = float(fps_str) + + return { + "width": stream.get("width", 0), + "height": stream.get("height", 0), + "fps": fps, + "duration": float(fmt.get("duration", 0)), + "codec": stream.get("codec_name", "unknown"), + "bitrate": int(stream.get("bit_rate", 0)) + } + except Exception as e: + logger.error(f"Failed to analyze video metadata: {e}") + return {} + + +async def extract_key_frames(video_path: str, num_frames: int = 5) -> list: + """ + 从视频中提取关键帧 + + 返回: 帧图片路径列表 + """ + try: + # 获取视频时长 + metadata = await analyze_video_metadata(video_path) + duration = metadata.get("duration", 0) + + if duration <= 0: + return [] + + # 均匀提取帧 + interval = duration / (num_frames + 1) + frames = [] + + with tempfile.TemporaryDirectory() as tmpdir: + for i in range(num_frames): + timestamp = interval * (i + 1) + output_path = os.path.join(tmpdir, f"frame_{i:03d}.jpg") + + proc = await asyncio.create_subprocess_exec( + "ffmpeg", + "-ss", str(timestamp), + "-i", video_path, + "-vframes", "1", + "-q:v", "2", + output_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + await proc.communicate() + + if os.path.exists(output_path): + frames.append(output_path) + + # 复制到持久化位置（如果需要） + return frames + except Exception as e: + logger.error(f"Failed to extract key frames: {e}") + return [] + + +async def evaluate_visual_quality(video_path: str) -> Tuple[float, str]: + """ + 评估视觉质量 + + 检查项： + - 分辨率 >= 720p (1280x720) + - 码率合理性（避免过度压缩） + - 帧率 >= 24fps + + 返回: (score 0-1, reason) + """ + metadata = await analyze_video_metadata(video_path) + + if not metadata: + return 0.0, "无法读取视频元数据" + + width = metadata.get("width", 0) + height = metadata.get("height", 0) + fps = metadata.get("fps", 0) + bitrate = metadata.get("bitrate", 0) + + score = 0.0 + reasons = [] + + # 分辨率评分 + pixels = width * height + if pixels >= 1920 * 1080: # 1080p+ + score += 0.4 + reasons.append(f"高分辨率 {width}x{height}") + elif pixels >= 1280 * 720: # 720p + score += 0.3 + reasons.append(f"标准分辨率 {width}x{height}") + elif pixels >= 640 * 480: # 480p + score += 0.2 + reasons.append(f"较低分辨率 {width}x{height}") + else: + reasons.append(f"分辨率过低 {width}x{height}") + + # 帧率评分 + if fps >= 29: + score += 0.3 + reasons.append(f"流畅帧率 {fps:.1f}fps") + elif fps >= 24: + score += 0.2 + reasons.append(f"标准帧率 {fps:.1f}fps") + else: + reasons.append(f"帧率过低 {fps:.1f}fps") + + # 码率评分（经验值） + if bitrate >= 2000000: # 2Mbps+ + score += 0.3 + reasons.append(f"码率充足 {bitrate//1000}kbps") + elif bitrate >= 1000000: + score += 0.2 + reasons.append(f"码率一般 {bitrate//1000}kbps") + else: + reasons.append(f"码率过低 {bitrate//1000}kbps") + + return score, "; ".join(reasons) + + +async def evaluate_motion_smoothness(video_path: str) -> Tuple[float, str]: + """ + 评估运动流畅度 + + 检查项： + - 帧率稳定性 + - 运动伪影（通过帧差分析） + + 返回: (score 0-1, reason) + """ + metadata = await analyze_video_metadata(video_path) + + if not metadata: + return 0.0, "无法读取视频元数据" + + fps = metadata.get("fps", 0) + + # 简单评估：帧率 >= 24fps 认为流畅 + if fps >= 29: + return 0.9, f"运动流畅 {fps:.1f}fps" + elif fps >= 24: + return 0.7, f"运动基本流畅 {fps:.1f}fps" + elif fps >= 20: + return 0.5, f"运动略显卡顿 {fps:.1f}fps" + else: + return 0.3, f"运动不流畅 {fps:.1f}fps" + + +async def evaluate_semantic_consistency( + video_path: str, + prompt: str, + frames: Optional[list] = None +) -> Tuple[float, str]: + """ + 评估语义一致性 - 视频内容与 prompt 描述的匹配度 + + 使用 LLM 多模态能力： + 1. 提取关键帧 + 2. 让 LLM 分析帧内容与 prompt 的匹配度 + + 返回: (score 0-1, reason) + """ + if not prompt: + return 0.5, "无 prompt，无法评估语义一致性" + + # 提取关键帧 + if not frames: + frames = await extract_key_frames(video_path, num_frames=3) + + if not frames: + return 0.5, "无法提取视频帧，跳过语义评估" + + # 构建 LLM prompt + llm_prompt = f"""分析以下视频帧与描述的匹配度： + +描述：{prompt} + +请评估： +1. 视频内容是否准确反映了描述中的场景、物体、动作 +2. 氛围和情绪是否匹配 +3. 整体视觉风格是否符合预期 + +返回 JSON 格式： +{{ + "match_score": 0-10, + "reason": "简要说明匹配情况" +}} +""" + + try: + # 调用 LLM 多模态（这里使用占位实现，实际应调用 vision_analyze 或类似工具） + # 暂时返回中性分数 + # TODO: 集成实际的 LLM 多模态调用 + + # 模拟 LLM 返回 + return 0.8, "语义一致性评估（待集成 LLM 多模态）" + + except Exception as e: + logger.warning(f"语义一致性评估失败: {e}") + return 0.5, f"语义评估异常: {e}" + + +async def evaluate_character_consistency( + video_path: str, + reference_image: Optional[str], + frames: Optional[list] = None +) -> Tuple[float, str]: + """ + 评估角色一致性 - Ref2V 生成时角色特征保持度 + + 使用 LLM 多模态： + 1. 比较参考图和视频帧中的人物 + 2. 评估面部特征、服装、姿态的一致性 + + 返回: (score 0-1, reason) + """ + if not reference_image: + return 1.0, "非 Ref2V 模式，跳过角色一致性评估" + + if not os.path.exists(reference_image): + return 0.5, f"参考图不存在: {reference_image}" + + # 提取关键帧 + if not frames: + frames = await extract_key_frames(video_path, num_frames=3) + + if not frames: + return 0.5, "无法提取视频帧，跳过角色一致性评估" + + try: + # 调用 LLM 多模态比较 + # TODO: 集成实际的 LLM 多模态调用 + + # 模拟返回 + return 0.85, "角色一致性评估（待集成 LLM 多模态）" + + except Exception as e: + logger.warning(f"角色一致性评估失败: {e}") + return 0.5, f"角色一致性评估异常: {e}" + + +async def evaluate_temporal_coherence(video_path: str) -> Tuple[float, str]: + """ + 评估时间连贯性 - 检测闪烁、跳变、伪影 + + 方法： + 1. 分析相邻帧的差异 + 2. 检测异常跳变 + + 返回: (score 0-1, reason) + """ + # 简单实现：假设大部分生成视频时间连贯性良好 + # 实际可以通过 ffmpeg 的 frame diff 分析实现 + + metadata = await analyze_video_metadata(video_path) + duration = metadata.get("duration", 0) + + # 视频时长越长，时间连贯性问题越容易暴露 + if duration >= 5: + return 0.8, "时间连贯性良好（基于时长推断）" + else: + return 0.7, "时间连贯性一般（视频较短）" + + +async def evaluate_scene_video_quality( + video_path: str, + prompt: str, + reference_image: Optional[str] = None +) -> Dict: + """ + 场景视频质量综合评估 + + 参数: + video_path: 视频文件路径 + prompt: 生成时的描述 + reference_image: Ref2V 的参考图（可选） + + 返回: + { + "overall_score": float (0-1), + "overall_reason": str, + "dimensions": { + "visual_quality": {"score": float, "weight": 0.25, "reason": str}, + "motion_smoothness": {"score": float, "weight": 0.20, "reason": str}, + "semantic_consistency": {"score": float, "weight": 0.30, "reason": str}, + "character_consistency": {"score": float, "weight": 0.15, "reason": str}, + "temporal_coherence": {"score": float, "weight": 0.10, "reason": str} + } + } + """ + # 基础检查 + if not os.path.exists(video_path): + return { + "overall_score": 0.0, + "overall_reason": f"视频文件不存在: {video_path}", + "dimensions": {} + } + + file_size = os.path.getsize(video_path) + if file_size < 10240: # < 10KB + return { + "overall_score": 0.0, + "overall_reason": f"视频文件过小: {file_size} bytes", + "dimensions": {} + } + + # 提取关键帧（用于多个维度） + frames = await extract_key_frames(video_path, num_frames=3) + + # 评估各维度 + visual_score, visual_reason = await evaluate_visual_quality(video_path) + motion_score, motion_reason = await evaluate_motion_smoothness(video_path) + semantic_score, semantic_reason = await evaluate_semantic_consistency(video_path, prompt, frames) + character_score, character_reason = await evaluate_character_consistency(video_path, reference_image, frames) + temporal_score, temporal_reason = await evaluate_temporal_coherence(video_path) + + # 加权计算总分 + dimensions = { + "visual_quality": {"score": visual_score, "weight": 0.25, "reason": visual_reason}, + "motion_smoothness": {"score": motion_score, "weight": 0.20, "reason": motion_reason}, + "semantic_consistency": {"score": semantic_score, "weight": 0.30, "reason": semantic_reason}, + "character_consistency": {"score": character_score, "weight": 0.15, "reason": character_reason}, + "temporal_coherence": {"score": temporal_score, "weight": 0.10, "reason": temporal_reason}, + } + + overall_score = sum(d["score"] * d["weight"] for d in dimensions.values()) + + # 生成综合原因 + reasons = [] + for dim_name, dim_data in dimensions.items(): + if dim_data["score"] < 0.6: + reasons.append(f"{dim_name}: {dim_data['reason']}") + + overall_reason = "; ".join(reasons) if reasons else "各维度评估良好" + + return { + "overall_score": overall_score, + "overall_reason": overall_reason, + "dimensions": dimensions + } diff --git a/app/ktv_adapter.py b/app/ktv_adapter.py index ae4dd0e..c34f199 100644 --- a/app/ktv_adapter.py +++ b/app/ktv_adapter.py @@ -17,6 +17,16 @@ import os import logging import tempfile import time +from functools import wraps + +from app.quality_gate import ( + eval_music_quality, + eval_character_design, + eval_character_image, + eval_storyboard, + eval_ktv_synthesis, + eval_scene_video_quality, +) logger = logging.getLogger("pipeline.handlers.ktv") @@ -915,6 +925,53 @@ async def handle_ktv_synthesizing(tenant_id, task_id, step_name, input_data, con return result +# ─── Quality Gate Wrappers ──────────────────────────────────────────── +# 步骤 9-10 (音乐), 11 (角色设计), 12 (角色图), 13 (分镜), 17 (合成) 带质量门控 + + +def _make_quality_handler(original_handler, eval_func): + """创建带质量门控的 wrapper handler。 + + 流程: handler → evaluator → 不达标重试(最多3次) → 仍不达标暂停等待人工 + """ + async def wrapper(tenant_id, task_id, step_name, input_data, config): + from app.quality_gate import with_quality_gate + + # 获取 version（用于人工任务记录） + version = 1 + if isinstance(config, dict): + version = config.get("version", 1) + if not version and isinstance(input_data, dict): + tp = input_data.get("task_params", {}) + if isinstance(tp, dict): + version = tp.get("version", 1) + + result = await with_quality_gate( + task_id=task_id, + step_name=step_name, + version=version, + handler=original_handler, + evaluator=eval_func, + tenant_id=tenant_id, + input_data=input_data, + config=config, + ) + return result + + wrapper.__name__ = f"quality_{original_handler.__name__}" + wrapper.__qualname__ = wrapper.__name__ + return wrapper + + +# 注册用的质量门控 handlers +quality_music_polling = _make_quality_handler(handle_music_polling, eval_music_quality) +quality_character_designing = _make_quality_handler(handle_character_designing, eval_character_design) +quality_character_image_generating = _make_quality_handler(handle_character_image_generating, eval_character_image) +quality_storyboard_generating = _make_quality_handler(handle_storyboard_generating, eval_storyboard) +quality_scene_video_evaluating = _make_quality_handler(handle_scene_video_evaluating, eval_scene_video_quality) +quality_ktv_synthesizing = _make_quality_handler(handle_ktv_synthesizing, eval_ktv_synthesis) + + # ─── Registration (adapter) ─────────────────────────────────────────── KTV_HANDLERS = { @@ -927,20 +984,24 @@ KTV_HANDLERS = { "lyric_generating": handle_lyric_generating, "lyric_evaluating": handle_lyric_evaluating, "music_generating": handle_music_generating, - "music_polling": handle_music_polling, - "character_designing": handle_character_designing, - "character_image_generating": handle_character_image_generating, - "storyboard_generating": handle_storyboard_generating, + "music_polling": quality_music_polling, # 质量门控 ✓ + "character_designing": quality_character_designing, # 质量门控 ✓ + "character_image_generating": quality_character_image_generating, # 质量门控 ✓ + "storyboard_generating": quality_storyboard_generating, # 质量门控 ✓ "scene_video_generating": handle_scene_video_generating, - "scene_video_evaluating": handle_scene_video_evaluating, + "scene_video_evaluating": quality_scene_video_evaluating, # 质量门控 ✓ "scene_video_concatenating": handle_scene_video_concatenating, - "ktv_synthesizing": handle_ktv_synthesizing, + "ktv_synthesizing": quality_ktv_synthesizing, # 质量门控 ✓ } def load_ktv_adapter(): """Register all KTV step handlers via pipeline_service handler registry.""" from pipeline_service.handler import register_handler + from app.quality_gate import ( + eval_music_quality, eval_character_design, eval_character_image, + eval_storyboard, eval_ktv_synthesis, + ) for step_type, fn in KTV_HANDLERS.items(): register_handler(step_type, fn) - logger.info(f"Registered {len(KTV_HANDLERS)} KTV handlers via adapter") + logger.info(f"Registered {len(KTV_HANDLERS)} KTV handlers via adapter (6 with quality gate)") diff --git a/app/quality_gate.py b/app/quality_gate.py new file mode 100644 index 0000000..191a010 --- /dev/null +++ b/app/quality_gate.py @@ -0,0 +1,449 @@ +"""质量门控：评估-重试-人工介入闭环。 + +用于 KTV 产线关键步骤的质量保障。 + +架构： +1. 执行 handler → 2. 评估结果 → 3. 不达标则重试(最多3次) → 4. 仍不达标则暂停等待人工决策 +""" + +import json +import logging +import os +from typing import Callable, Any, Optional, Dict, Tuple + +logger = logging.getLogger("pipeline.quality_gate") + +# 最大自动重试次数 +MAX_RETRIES = 3 + + +async def with_quality_gate( + task_id: str, + step_name: str, + version: int, + handler: Callable, + evaluator: Callable, + tenant_id: str = "", + input_data: Optional[dict] = None, + config: Optional[dict] = None, +) -> dict: + """带质量门控的 handler 执行器。 + + 参数: + handler: 原始步骤处理函数 + evaluator: 评估函数 async def evaluator(output_data) -> (passed: bool, score: float, reason: str) + + 流程: + 1. 执行 handler + 2. 调用 evaluator 评估结果 + 3. 不通过则重试（最多 MAX_RETRIES 次） + 4. 仍不通过则创建 human_task，暂停任务 + + 返回: + handler 的输出数据（含 _quality_meta 元信息） + """ + input_data = input_data or {} + config = config or {} + + last_output: Optional[dict] = None + last_reason = "" + + for attempt in range(1, MAX_RETRIES + 1): + logger.info(f"[QualityGate] {step_name} 第 {attempt}/{MAX_RETRIES} 次执行") + + try: + # 执行 handler + output = await handler(tenant_id, task_id, step_name, input_data, config) + last_output = output + + # 评估结果 + passed, score, reason = await evaluator(output) + + logger.info(f"[QualityGate] {step_name} 评估: passed={passed}, score={score:.2f}, reason={reason}") + + if passed: + # 通过，附加元信息返回 + output["_quality_meta"] = { + "attempts": attempt, + "score": score, + "reason": reason, + "passed": True, + } + return output + + last_reason = reason + + except Exception as e: + logger.warning(f"[QualityGate] {step_name} 第 {attempt} 次执行异常: {e}") + last_reason = f"执行异常: {e}" + + # 超过重试次数，创建人工任务 + logger.warning( + f"[QualityGate] {step_name} 经过 {MAX_RETRIES} 次尝试仍不达标，创建人工任务" + ) + + await _create_quality_review_task( + task_id=task_id, + step_name=step_name, + version=version, + attempts=MAX_RETRIES, + last_output=last_output, + last_reason=last_reason, + ) + + # 抛出异常让 executor 检测到 WAITING 状态并暂停任务 + raise QualityGatePausedError( + f"步骤 {step_name} 质量评估未通过，已暂停等待人工审核。原因: {last_reason}" + ) + + +async def _create_quality_review_task( + task_id: str, + step_name: str, + version: int, + attempts: int, + last_output: dict, + last_reason: str, +): + """创建质量审核人工任务。""" + from pipeline_service.storage import create_human_task + from pipeline_service.state import STATE_WAITING + from pipeline_service.storage import update_step_state + + # 构造审核表单 + form_schema = { + "title": f"步骤 {step_name} 质量审核", + "description": f"经过 {attempts} 次自动重试仍未通过质量评估", + "fields": [ + { + "name": "last_reason", + "label": "失败原因", + "type": "textarea", + "readonly": True, + "default": last_reason, + }, + { + "name": "last_output_summary", + "label": "最后一次输出摘要", + "type": "textarea", + "readonly": True, + "default": json.dumps(last_output, ensure_ascii=False, indent=2)[:2000] if last_output else "无", + }, + { + "name": "decision", + "label": "决策", + "type": "select", + "options": [ + {"value": "retry_manual", "label": "手动重试（调整参数后重新执行）"}, + {"value": "accept", "label": "接受当前结果（忽略质量要求）"}, + {"value": "abort", "label": "终止任务"}, + ], + "required": True, + }, + { + "name": "notes", + "label": "备注说明", + "type": "textarea", + "required": False, + }, + ], + } + + # 创建人工任务记录 + await create_human_task( + task_id=task_id, + step_name=step_name, + version=version, + task_type="quality_review", + assignee_role="admin", + form_schema=form_schema, + timeout_hours=24, + ) + + # 设置步骤状态为等待 + await update_step_state(task_id, step_name, STATE_WAITING) + + logger.info(f"[QualityGate] 已创建人工审核任务: {task_id}/{step_name}") + + +class QualityGatePausedError(Exception): + """质量门控暂停异常，用于通知 executor 暂停任务。""" + pass + + +# ─── 评估函数 ────────────────────────────────────────────────────────────── + +async def eval_music_quality(output: dict) -> Tuple[bool, float, str]: + """评估音乐生成结果（步骤 9-10）— 多维度评估。 + + 7 个评估维度： + 节奏质量 (15%) | 和弦进行 (10%) | 听感质量 (20%) + 情绪-歌词匹配 (20%) | 指令匹配度 (15%) + 伴奏质量 (10%) | 人声质量 (10%) + """ + music_path = output.get("music_path", "") + + if not music_path or not os.path.exists(music_path): + return False, 0.0, f"音乐文件不存在: {music_path}" + + file_size = os.path.getsize(music_path) + if file_size < 50 * 1024: + return False, 0.3, f"音乐文件过小 ({file_size/1024:.1f}KB < 50KB)" + + # 调用多维度评估 + try: + from app.eval_music import evaluate_music_quality + prompt = output.get("prompt", output.get("music_prompt", "")) + lyrics = output.get("lyrics", "") + + result = await evaluate_music_quality( + audio_path=music_path, + prompt=prompt, + lyrics=lyrics, + use_demucs=True, + ) + + overall_score = result["overall_score"] + overall_reason = result["overall_reason"] + passed = overall_score >= 0.6 + + # 将维度评分存入 output 供后续参考 + output["_eval_dimensions"] = result.get("dimensions", {}) + + return passed, overall_score, overall_reason + + except ImportError: + logger.warning("eval_music 模块未安装，降级为基础评估") + # 降级：仅检查时长 + import asyncio + proc = await asyncio.create_subprocess_shell( + f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{music_path}'", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + duration = float(stdout.decode().strip()) if stdout else 0 + + if duration < 30: + return False, 0.5, f"音乐时长不足 ({duration:.1f}s < 30s)" + + score = min(1.0, duration / 180) + return True, score, f"音乐文件正常 ({duration:.1f}s, {file_size/1024:.0f}KB)" + + +async def eval_character_design(output: dict) -> Tuple[bool, float, str]: + """评估角色设计结果（步骤 11）。 + + 评估标准： + 1. characters 列表非空 + 2. 每个角色包含 name, prompt/description, personality + 3. prompt 描述长度合理（> 50 字符） + """ + characters = output.get("characters", []) + + if not characters or not isinstance(characters, list): + return False, 0.0, "角色列表为空" + + issues = [] + for i, char in enumerate(characters): + if not isinstance(char, dict): + issues.append(f"角色 {i} 格式错误") + continue + + name = char.get("name", "") + prompt = char.get("prompt", char.get("description", "")) + personality = char.get("personality", "") + + if not name: + issues.append(f"角色 {i} 缺少 name") + if not prompt or len(prompt) < 50: + issues.append(f"角色 {i} 的 prompt 描述过短 ({len(prompt)} chars)") + if not personality: + issues.append(f"角色 {i} 缺少 personality") + + if issues: + return False, 0.3, "; ".join(issues[:3]) + + return True, 0.9, f"角色设计完整 ({len(characters)} 个角色)" + + +async def eval_character_image(output: dict) -> Tuple[bool, float, str]: + """评估角色图生成结果（步骤 12）。 + + 评估标准： + 1. character_images 列表非空 + 2. 每个图片文件存在且大小 > 10KB + 3. 图片数量匹配角色数量 + """ + char_images = output.get("character_images", []) + + if not char_images or not isinstance(char_images, list): + return False, 0.0, "角色图列表为空" + + missing = [] + too_small = [] + + for item in char_images: + path = item.get("image_path", "") + name = item.get("name", "unknown") + + if not path or not os.path.exists(path): + missing.append(name) + continue + + size = os.path.getsize(path) + if size < 10 * 1024: + too_small.append(f"{name}({size/1024:.1f}KB)") + + if missing: + return False, 0.3, f"图片缺失: {', '.join(missing)}" + if too_small: + return False, 0.5, f"图片过小: {', '.join(too_small)}" + + return True, 0.9, f"角色图生成正常 ({len(char_images)} 张)" + + +async def eval_storyboard(output: dict) -> Tuple[bool, float, str]: + """评估分镜脚本结果（步骤 13）。 + + 评估标准： + 1. storyboard 列表非空 + 2. 每个分镜包含 scene_id, start_time, end_time, description + 3. 时间轴覆盖完整（无大段空白） + 4. description 描述充分 + """ + storyboard = output.get("storyboard", []) + + if not storyboard or not isinstance(storyboard, list): + return False, 0.0, "分镜列表为空" + + if len(storyboard) < 3: + return False, 0.3, f"分镜数量过少 ({len(storyboard)} < 3)" + + issues = [] + prev_end = 0 + + for i, scene in enumerate(storyboard): + if not isinstance(scene, dict): + issues.append(f"分镜 {i} 格式错误") + continue + + # 必填字段检查 + for field in ["scene_id", "start_time", "end_time", "description"]: + if field not in scene: + issues.append(f"分镜 {i} 缺少 {field}") + + # 时间轴连续性 + start = scene.get("start_time", 0) + end = scene.get("end_time", 0) + if start > prev_end + 5: # 允许 5 秒间隔 + issues.append(f"分镜 {i} 时间轴有断层 ({prev_end}s → {start}s)") + prev_end = end + + # description 质量 + desc = scene.get("description", "") + if len(desc) < 20: + issues.append(f"分镜 {i} 描述过短 ({len(desc)} chars)") + + if len(issues) > len(storyboard) // 2: + return False, 0.4, "; ".join(issues[:3]) + + score = 0.9 if not issues else max(0.6, 0.9 - 0.1 * len(issues)) + return True, score, f"分镜脚本完整 ({len(storyboard)} 个分镜, {issues and '有'+str(len(issues))+'个问题' or '无问题'})" + + +async def eval_ktv_synthesis(output: dict) -> Tuple[bool, float, str]: + """评估 KTV 合成结果（步骤 17）。 + + 评估标准： + 1. ktv_path 文件存在且大小 > 1MB + 2. 视频时长 > 30 秒 + 3. 字幕文件存在 + 4. MTV 版本可选 + """ + ktv_path = output.get("ktv_path", "") + subtitle_path = output.get("subtitle_path", "") + + # 检查 KTV 视频 + if not ktv_path or not os.path.exists(ktv_path): + return False, 0.0, f"KTV 视频不存在: {ktv_path}" + + file_size = os.path.getsize(ktv_path) + if file_size < 1024 * 1024: + return False, 0.3, f"KTV 视频过小 ({file_size/1024/1024:.1f}MB < 1MB)" + + # 用 ffprobe 检查时长 + import asyncio + proc = await asyncio.create_subprocess_shell( + f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{ktv_path}'", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + duration = float(stdout.decode().strip()) if stdout else 0 + + if duration < 30: + return False, 0.5, f"KTV 视频时长不足 ({duration:.1f}s < 30s)" + + # 检查字幕 + if not subtitle_path or not os.path.exists(subtitle_path): + return False, 0.6, f"字幕文件缺失: {subtitle_path}" + + score = min(1.0, duration / 180) # 3分钟满分 + return True, score, f"KTV 合成正常 ({duration:.1f}s, {file_size/1024/1024:.1f}MB)" + + +async def eval_scene_video_quality(output: dict) -> Tuple[bool, float, str]: + """评估场景视频生成结果（步骤 14）— 多维度评估。 + + 5 个评估维度： + 视觉质量 (25%) | 运动流畅度 (20%) | 语义一致性 (30%) + 角色一致性 (15%) | 时间连贯性 (10%) + """ + scene_video_path = output.get("scene_video_path", "") + + if not scene_video_path or not os.path.exists(scene_video_path): + return False, 0.0, f"场景视频不存在: {scene_video_path}" + + file_size = os.path.getsize(scene_video_path) + if file_size < 1024 * 1024: + return False, 0.3, f"场景视频过小 ({file_size/1024/1024:.1f}MB < 1MB)" + + # 调用多维度评估 + try: + from app.eval_video import evaluate_scene_video_quality + prompt = output.get("prompt", output.get("scene_prompt", "")) + reference_image = output.get("reference_image", "") + + result = await evaluate_scene_video_quality( + video_path=scene_video_path, + prompt=prompt, + reference_image=reference_image, + ) + + overall_score = result["overall_score"] + overall_reason = result["overall_reason"] + passed = overall_score >= 0.6 + + # 将维度评分存入 output 供后续参考 + output["_eval_dimensions"] = result.get("dimensions", {}) + + return passed, overall_score, overall_reason + + except ImportError: + logger.warning("eval_video 模块未安装，降级为基础评估") + # 降级：仅检查时长 + import asyncio + proc = await asyncio.create_subprocess_shell( + f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{scene_video_path}'", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + duration = float(stdout.decode().strip()) if stdout else 0 + + if duration < 10: + return False, 0.5, f"场景视频时长不足 ({duration:.1f}s < 10s)" + + score = min(1.0, duration / 30) + return True, score, f"场景视频正常 ({duration:.1f}s, {file_size/1024/1024:.1f}MB)"