video_decode/examples/run_decode_script.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
批量运行解码工作流 (DecodeWorkflow)。

读取 examples/demo.json 中的视频列表，
逐一调用 DecodeWorkflow 进行处理，
并将结果输出到 examples/output_decode_result.json。
"""

import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List

# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from src.workflows.decode_workflow import DecodeWorkflow
from src.utils.logger import get_logger

logger = get_logger(__name__)


def load_json(path: Path) -> List[Dict[str, Any]]:
    """加载JSON文件"""
    if not path.exists():
        return []
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
        # 如果是字典且有 results 字段，提取 results
        if isinstance(data, dict) and "results" in data:
            return data["results"]
        # 如果是列表，直接返回
        elif isinstance(data, list):
            return data
        else:
            return []


def save_json(path: Path, data: Dict[str, Any]) -> None:
    """保存JSON文件（使用临时文件确保原子性）"""
    tmp_path = path.with_suffix(".tmp")
    with tmp_path.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    tmp_path.replace(path)


def build_decode_input(video_data: Dict[str, Any]) -> Dict[str, Any]:
    """根据视频数据构造 DecodeWorkflow 的输入结构"""
    return {
        "video_url": video_data.get("video_url", ""),
        "video_id": video_data.get("video_id", ""),
        "title": video_data.get("title", ""),
    }

def main() -> None:
    """主函数"""
    base_dir = Path(__file__).parent
    input_path = base_dir / "demo.json"
    output_path = base_dir / "output_decode_result.json"

    if not input_path.exists():
        raise FileNotFoundError(f"找不到输入文件: {input_path}")

    # 读取视频列表
    video_list = load_json(input_path)
    if not video_list:
        logger.warning(f"输入文件 {input_path} 中没有视频数据")
        return

    logger.info(f"共读取到 {len(video_list)} 个视频")

    # 读取已有的输出结果，支持增量追加
    output_data = {}
    if output_path.exists():
        try:
            with output_path.open("r", encoding="utf-8") as f:
                output_data = json.load(f)
        except Exception as e:
            logger.warning(f"读取已有输出文件失败，将创建新文件: {e}")
            output_data = {}

    if not output_data:
        output_data = {
            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
            "total": 0,
            "success_count": 0,
            "fail_count": 0,
            "results": [],
        }

    existing_results: List[Dict[str, Any]] = output_data.get("results", []) or []
    # 用 video_id + video URL 去重，避免重复处理（兼容旧字段名 channel_content_id）
    processed_keys = {
        f"{item.get('video_data', {}).get('video_id', '')}|"
        f"{item.get('video_data', {}).get('video_url', '')}"
        for item in existing_results
    }

    # 初始化工作流
    logger.info("初始化 DecodeWorkflow...")
    workflow = DecodeWorkflow()
    logger.info("DecodeWorkflow 初始化完成")

    # 处理每个视频
    for idx, video_data in enumerate(video_list, 1):
        video_url = video_data.get("video_url", "")
        video_id = video_data.get("video_id", "") or video_data.get("channel_content_id", "")  # 兼容旧字段名
        title = video_data.get("title", "")

        # 生成唯一键用于去重
        key = f"{video_id}|{video_url}"
        if key in processed_keys:
            logger.info(f"[{idx}/{len(video_list)}] 已处理过该视频，跳过: video_id={video_id}")
            continue

        logger.info(
            f"[{idx}/{len(video_list)}] 开始处理视频: "
            f"video_id={video_id}, title={title[:50]}..."
        )

        try:
            # 构建输入数据
            decode_input = build_decode_input(video_data)

            # 调用工作流
            decode_result = workflow.invoke(decode_input)

            # 检查workflow返回结果中是否包含错误
            if isinstance(decode_result, dict):
                # 检查是否有错误字段（支持多种错误字段名）
                error_msg = (
                    decode_result.get("error") or 
                    decode_result.get("错误") or
                    decode_result.get("workflow_error")
                )
                workflow_status = decode_result.get("workflow_status")
                
                # 如果返回了错误信息，视为失败
                if error_msg or workflow_status == "failed" or workflow_status == "incomplete":
                    error_msg = error_msg or "工作流执行失败"
                    logger.error(
                        f"[{idx}/{len(video_list)}] 处理失败: video_id={video_id}, error={error_msg}"
                    )
                    record = {
                        "video_data": video_data,
                        "what_deconstruction_result": None,
                        "script_result": None,
                        "success": False,
                        "error": error_msg,
                    }
                    output_data["fail_count"] = output_data.get("fail_count", 0) + 1
                    output_data["results"].append(record)
                    output_data["total"] = output_data.get("total", 0) + 1
                    save_json(output_path, output_data)
                    continue
                
                # 检查结果是否为空（可能表示失败）
                # 如果所有关键字段都为空，可能表示处理失败
                video_info = decode_result.get("视频信息", {})
                three_points = decode_result.get("三点解构", {})
                topic_understanding = decode_result.get("选题理解", {})
                script_understanding = decode_result.get("脚本理解", {})
                
                # 如果所有关键结果都为空，且没有明确的成功标志，视为失败
                if (not video_info and not three_points and 
                    not topic_understanding and not script_understanding):
                    error_msg = "工作流执行完成，但所有结果都为空"
                    logger.warning(
                        f"[{idx}/{len(video_list)}] 处理结果为空: video_id={video_id}"
                    )
                    # 这里可以选择记录为失败或警告，根据业务需求决定
                    # 暂时记录为失败
                    record = {
                        "video_data": video_data,
                        "what_deconstruction_result": None,
                        "script_result": None,
                        "success": False,
                        "error": error_msg,
                    }
                    output_data["fail_count"] = output_data.get("fail_count", 0) + 1
                    output_data["results"].append(record)
                    output_data["total"] = output_data.get("total", 0) + 1
                    save_json(output_path, output_data)
                    continue

            # 按照 output_demo_script.json 的格式组织结果
            # what_deconstruction_result: 包含视频信息、三点解构、选题理解
            what_deconstruction_result = {
                "视频信息": decode_result.get("视频信息", {}),
                "三点解构": decode_result.get("三点解构", {}),
                "选题理解": decode_result.get("选题理解", {}),
            }
            
            # script_result: 包含选题描述和脚本理解
            # 从选题理解中提取选题描述
            topic_understanding = decode_result.get("选题理解", {})
            selected_topic = {}
            if isinstance(topic_understanding, dict):
                if "选题" in topic_understanding:
                    selected_topic = topic_understanding.get("选题", {})
                else:
                    selected_topic = {
                        "主题": topic_understanding.get("主题", ""),
                        "描述": topic_understanding.get("描述", ""),
                    }
            
            script_result = {
                "选题描述": selected_topic,
                "脚本理解": decode_result.get("脚本理解", {}),
            }

            # 构造结果记录（参考 output_demo_script.json 格式）
            record = {
                "video_data": video_data,
                "what_deconstruction_result": what_deconstruction_result,
                "script_result": script_result,
                "success": True,
                "error": None,
            }

            output_data["success_count"] = output_data.get("success_count", 0) + 1
            logger.info(
                f"[{idx}/{len(video_list)}] 处理成功: video_id={video_id}"
            )

        except Exception as e:
            logger.error(
                f"[{idx}/{len(video_list)}] 处理失败: video_id={video_id}, error={e}",
                exc_info=True
            )
            record = {
                "video_data": video_data,
                "what_deconstruction_result": None,
                "script_result": None,
                "success": False,
                "error": str(e),
            }
            output_data["fail_count"] = output_data.get("fail_count", 0) + 1

        output_data["results"].append(record)
        output_data["total"] = output_data.get("total", 0) + 1

        # 处理完一条就保存一次，避免长任务中途失败导致全部丢失
        save_json(output_path, output_data)
        logger.info(f"结果已保存到 {output_path}")

    logger.info(
        f"\n{'='*60}\n"
        f"批量解码完成:\n"
        f"  总计: {output_data.get('total', 0)}\n"
        f"  成功: {output_data.get('success_count', 0)}\n"
        f"  失败: {output_data.get('fail_count', 0)}\n"
        f"  输出文件: {output_path}\n"
        f"{'='*60}"
    )


if __name__ == "__main__":
    main()