video_decode/src/workflows/what_deconstruction_workflow.py

"""
What Deconstruction Workflow.

What解构主工作流：编排三点解构流程（灵感点、目的点、关键点）的执行顺序和流程逻辑。
流程（视频分析）：视频上传 → 灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 结果汇总
"""

from typing import Dict, Any
from langgraph.graph import StateGraph, END

from src.components.agents.base import BaseGraphAgent
from src.states.what_deconstruction_state import WhatDeconstructionState
from src.components.agents.topic_selection_understanding_agent import TopicSelectionUnderstandingAgent
from src.components.agents.search_keyword_agent import SearchKeywordAgent
from src.components.functions.result_aggregation_function import ResultAggregationFunction
from src.components.functions.video_upload_function import VideoUploadFunction
# 新增三点解构Agent
from src.components.agents.inspiration_points_agent import InspirationPointsAgent
from src.components.agents.purpose_point_agent import PurposePointAgent
from src.components.agents.key_points_agent import KeyPointsAgent
# 新增选题结构Agent V2
from src.components.agents.topic_agent_v2 import TopicAgentV2
# 新增结构化内容库Agent
from src.components.agents.structure_agent import StructureAgent
from src.utils.logger import get_logger

logger = get_logger(__name__)


class WhatDeconstructionWorkflow(BaseGraphAgent):
    """What解构主工作流（视频分析版本）

    功能：
    - 编排整个What解构流程（针对视频输入）
    - 支持条件分支：
      * 视频上传 → topic_selection_v2（选题结构分析V2，直接基于视频）→ 结束
      * 视频上传 → structure_agent（结构化内容库解构，直接基于视频）→ 结束
      * 视频上传 → 灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 搜索关键词 → 结果汇总 → 结束
    - 通过 state 中的 use_topic_agent_v2 或 use_structure_agent 标志手动控制分支选择（二选一）
    - 管理状态传递
    - 仅支持单视频输入

    实现方式：BaseGraphAgent (LangGraph)
    """

    def __init__(
        self,
        name: str = "what_deconstruction_workflow",
        description: str = "What解构主工作流（视频分析）",
        model_provider: str = "google_genai",
        max_depth: int = 10
    ):
        super().__init__(
            name=name,
            description=description,
            state_class=WhatDeconstructionState
        )

        self.max_depth = max_depth

        # 初始化视频上传Function
        self.video_upload_func = VideoUploadFunction()

        # 初始化选题理解Agent
        self.topic_selection_understanding_agent = TopicSelectionUnderstandingAgent(
            model_provider=model_provider
        )
        
        # 初始化搜索关键词Agent
        self.search_keyword_agent = SearchKeywordAgent(
            model_provider=model_provider
        )
        
        # 初始化结果汇总Function
        self.result_aggregation_func = ResultAggregationFunction()

        # 初始化新的三点解构Agent
        self.inspiration_points_agent = InspirationPointsAgent(
            model_provider=model_provider
        )
        self.purpose_point_agent = PurposePointAgent(
            model_provider=model_provider
        )
        self.key_points_agent = KeyPointsAgent(
            model_provider=model_provider
        )

        # 初始化选题结构Agent V2
        self.topic_agent_v2 = TopicAgentV2(
            model_provider=model_provider
        )

        # 初始化结构化内容库Agent
        self.structure_agent = StructureAgent(
            model_provider=model_provider
        )

        logger.info(f"WhatDeconstructionWorkflow（视频分析）初始化完成")

    def _build_graph(self) -> StateGraph:
        """构建工作流图

        新流程（视频分析）：
        START → 视频上传 → [条件分支]
          - 如果 use_topic_agent_v2=True: → topic_selection_v2 → END
          - 如果 use_structure_agent=True: → structure_agent → END
          - 否则: → 灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 搜索关键词 → 结果汇总 → END
        """
        workflow = StateGraph(dict)  # 使用dict作为状态类型

        # 添加节点
        workflow.add_node("video_upload", self._video_upload_node)
        workflow.add_node("topic_selection_v2", self._topic_selection_v2_node)
        workflow.add_node("structure_agent", self._structure_agent_node)
        workflow.add_node("inspiration_points_extraction", self._inspiration_points_node)
        workflow.add_node("purpose_point_extraction", self._purpose_point_node)
        workflow.add_node("key_points_extraction", self._key_points_node)
        workflow.add_node("topic_selection_understanding", self._topic_selection_understanding_node)
        workflow.add_node("search_keyword_extraction", self._search_keyword_node)
        workflow.add_node("result_aggregation", self._result_aggregation_node)

        # 定义流程的边：视频上传 → 条件分支 → topic_selection_v2/structure_agent（结束） 或 原流程
        workflow.set_entry_point("video_upload")
        # 条件分支：根据 use_topic_agent_v2 或 use_structure_agent 标志决定走哪个分支
        workflow.add_conditional_edges(
            "video_upload",
            self._route_after_upload,
            {
                "topic_selection_v2": "topic_selection_v2",
                "structure_agent": "structure_agent",
                "normal_flow": "inspiration_points_extraction"
            }
        )
        # topic_selection_v2 分支直接结束
        workflow.add_edge("topic_selection_v2", END)
        # structure_agent 分支直接结束
        workflow.add_edge("structure_agent", END)
        # 原流程继续
        workflow.add_edge("inspiration_points_extraction", "purpose_point_extraction")
        workflow.add_edge("purpose_point_extraction", "key_points_extraction")
        workflow.add_edge("key_points_extraction", "topic_selection_understanding")
        workflow.add_edge("topic_selection_understanding", "search_keyword_extraction")
        workflow.add_edge("search_keyword_extraction", "result_aggregation")
        workflow.add_edge("result_aggregation", END)

        logger.info("工作流图构建完成 - 视频分析流程：视频上传 → [条件分支: topic_selection_v2 / structure_agent / 三点解构流程]")

        return workflow

    def _video_upload_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：视频上传（第一步）- 下载视频并上传至Gemini"""
        logger.info("=== 执行节点：视频上传 ===")

        try:
            # 初始化Function
            if not self.video_upload_func.is_initialized:
                self.video_upload_func.initialize()

            # 执行视频上传
            result = self.video_upload_func.execute(state)

            # 更新状态
            state.update(result)

            video_uri = result.get("video_uploaded_uri")
            if video_uri:
                logger.info(f"视频上传完成 - URI: {video_uri}")
            else:
                error = result.get("video_upload_error", "未知错误")
                logger.warning(f"视频上传失败: {error}")

        except Exception as e:
            logger.error(f"视频上传失败: {e}", exc_info=True)
            state.update({
                "video_uploaded_uri": None,
                "video_upload_error": str(e)
            })

        return state

    def _route_after_upload(self, state: Dict[str, Any]) -> str:
        """条件分支函数：路由到不同的处理分支
        
        通过 state 中的标志来控制：
        - use_topic_agent_v2=True: 走 topic_selection_v2 分支，直接结束
        - use_structure_agent=True: 走 structure_agent 分支，直接结束
        - 否则: 走原来的正常流程
        
        Returns:
            "topic_selection_v2" / "structure_agent" / "normal_flow"
        """
        use_v2 = state.get("use_topic_agent_v2", False)
        use_structure = state.get("use_structure_agent", False)
        
        if use_v2:
            logger.info("检测到 use_topic_agent_v2=True，将使用 TopicAgentV2 分支")
            return "topic_selection_v2"
        elif use_structure:
            logger.info("检测到 use_structure_agent=True，将使用 StructureAgent 分支")
            return "structure_agent"
        else:
            logger.info("使用正常流程（三点解构 → 选题理解 → 搜索关键词）")
            return "normal_flow"

    def _topic_selection_v2_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：选题结构分析 V2（直接基于视频）"""
        logger.info("=== 执行节点：选题结构分析 V2 ===")

        try:
            # 初始化Agent
            if not self.topic_agent_v2.is_initialized:
                self.topic_agent_v2.initialize()

            # 执行Agent
            result = self.topic_agent_v2.process(state)

            # 更新状态
            state.update(result)

            topic_selection = result.get("topic_selection_v2", {})
            if "error" not in topic_selection:
                logger.info(f"选题结构分析 V2 完成 - topic_selection_v2: {topic_selection}")
            else:
                logger.warning(f"选题结构分析 V2 执行出错: {topic_selection.get('error')}")

        except Exception as e:
            logger.error(f"选题结构分析 V2 失败: {e}", exc_info=True)
            state.update({
                "video_script": "",
                "topic_selection_v2": {
                    "error": str(e),
                }
            })

        return state

    def _structure_agent_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：结构化内容库解构（直接基于视频）"""
        logger.info("=== 执行节点：结构化内容库解构 ===")

        try:
            # 初始化Agent
            if not self.structure_agent.is_initialized:
                self.structure_agent.initialize()

            # 执行Agent
            result = self.structure_agent.process(state)

            # 更新状态
            state.update(result)

            structure_data = result.get("structure_data", {})
            if "错误" not in structure_data:
                logger.info(f"结构化内容库解构完成 - structure_data: {structure_data}")
            else:
                logger.warning(f"结构化内容库解构执行出错: {structure_data.get('错误')}")

        except Exception as e:
            logger.error(f"结构化内容库解构失败: {e}", exc_info=True)
            state.update({
                "structure_data": {
                    "错误": str(e),
                }
            })

        return state

    def _topic_selection_understanding_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：选题理解"""
        logger.info("=== 执行节点：选题理解 ===")

        try:
            # 初始化Agent
            if not self.topic_selection_understanding_agent.is_initialized:
                self.topic_selection_understanding_agent.initialize()

            # 执行Agent
            result = self.topic_selection_understanding_agent.process(state)

            # 更新状态
            state.update(result)

            logger.info(f"选题理解完成 - result: {result}")

        except Exception as e:
            logger.error(f"选题理解失败: {e}", exc_info=True)
            state.update({
                "topic_selection_understanding": {}
            })

        return state

    def _search_keyword_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：搜索关键词提取"""
        logger.info("=== 执行节点：搜索关键词提取 ===")

        try:
            # 初始化Agent
            if not self.search_keyword_agent.is_initialized:
                self.search_keyword_agent.initialize()

            # 执行Agent
            result = self.search_keyword_agent.process(state)

            # 更新状态
            state.update(result)

            search_keywords_count = result.get("search_keywords", {}).get("总数", 0)
            logger.info(f"搜索关键词提取完成 - 共 {search_keywords_count} 个搜索词")

        except Exception as e:
            logger.error(f"搜索关键词提取失败: {e}", exc_info=True)
            state.update({
                "search_keywords": {
                    "搜索词列表": [],
                    "总数": 0,
                    "错误": str(e)
                }
            })

        return state

    def _inspiration_points_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：灵感点提取（新三点解构）"""
        logger.info("=== 执行节点：灵感点提取 ===")

        try:
            # 初始化Agent
            if not self.inspiration_points_agent.is_initialized:
                self.inspiration_points_agent.initialize()

            # 执行Agent
            result = self.inspiration_points_agent.process(state)

            # 更新状态
            state.update(result)

            # 安全地获取灵感点数量：total_count 在 metadata 中
            if isinstance(result, dict):
                metadata = result.get("metadata", {})
                inspiration_count = metadata.get("total_count", 0) if isinstance(metadata, dict) else 0
                # 如果 metadata 中没有，尝试从 inspiration_points 列表长度获取
                if inspiration_count == 0:
                    inspiration_points = result.get("inspiration_points", [])
                    if isinstance(inspiration_points, list):
                        inspiration_count = len(inspiration_points)
            else:
                # 如果 result 不是 dict（比如是列表），尝试获取长度
                inspiration_count = len(result) if isinstance(result, list) else 0
            
            logger.info(f"灵感点提取完成 - 共 {inspiration_count} 个灵感点")

        except Exception as e:
            logger.error(f"灵感点提取失败: {e}", exc_info=True)
            state.update({
                "inspiration_points": {
                    "error": str(e),
                    "points": [],
                    "total_count": 0
                }
            })

        return state

    def _purpose_point_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：目的点提取（新三点解构）"""
        logger.info("=== 执行节点：目的点提取 ===")

        try:
            # 初始化Agent
            if not self.purpose_point_agent.is_initialized:
                self.purpose_point_agent.initialize()

            # 执行Agent
            result = self.purpose_point_agent.process(state)

            # 更新状态
            state.update(result)

            main_purpose = result.get("purpose_point", {}).get("main_purpose", "未知")
            logger.info(f"目的点提取完成 - 主要目的: {main_purpose}")

        except Exception as e:
            logger.error(f"目的点提取失败: {e}", exc_info=True)
            state.update({
                "purpose_point": {
                    "error": str(e),
                    "main_purpose": "未知"
                }
            })

        return state

    def _key_points_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：关键点提取（新三点解构）"""
        logger.info("=== 执行节点：关键点提取 ===")

        try:
            # 初始化Agent
            if not self.key_points_agent.is_initialized:
                self.key_points_agent.initialize()

            # 执行Agent
            result = self.key_points_agent.process(state)

            # 更新状态
            state.update(result)

            total_key_points = result.get("key_points", {}).get("total_count", 0)
            logger.info(f"关键点提取完成 - 共 {total_key_points} 个关键点")

        except Exception as e:
            logger.error(f"关键点提取失败: {e}", exc_info=True)
            state.update({
                "key_points": {
                    "error": str(e),
                    "creator_perspective": {"key_points": [], "summary": ""},
                    "consumer_perspective": {"key_points": [], "summary": ""}
                }
            })

        return state

    def _result_aggregation_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：结果汇总"""
        logger.info("=== 执行节点：结果汇总 ===")

        try:
            # 初始化Function
            if not self.result_aggregation_func.is_initialized:
                self.result_aggregation_func.initialize()

            # 执行Function
            final_result = self.result_aggregation_func.execute(state)

            # 更新状态
            state["final_result"] = final_result

            logger.info("结果汇总完成")

        except Exception as e:
            logger.error(f"结果汇总失败: {e}", exc_info=True)
            state["final_result"] = {
                "帖子总结": {"错误": f"汇总失败: {str(e)}"},
                "帖子包含元素": []
            }

        return state

    def invoke(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """执行工作流（公共接口）- 视频分析版本

        Args:
            input_data: 输入数据，包含 video 字段（视频URL）
                格式参考：examples/56898272/视频详情.json
                {
                    "video": "http://...",
                    "title": "...",
                    "use_topic_agent_v2": False,  # 可选，控制是否使用 topic_agent_v2 分支
                    "use_structure_agent": True,  # 可选，控制是否使用 structure_agent 分支（与 use_topic_agent_v2 二选一），默认 True
                    ...
                }

        Returns:
            最终解构结果
        """
        logger.info("=== 开始执行 What 解构工作流（视频分析） ===")

        # 确保工作流已初始化
        if not self.is_initialized:
            self.initialize()

        # 初始化状态（仅视频输入）
        initial_state = {
            "video": input_data.get("video", ""),
            "channel_content_id": input_data.get("channel_content_id", ""),
            "text": {
                "title": input_data.get("title", ""),
                "body": input_data.get("body_text", ""),
                "hashtags": []
            },
            "current_depth": 0,
            "max_depth": self.max_depth,
            "use_topic_agent_v2": input_data.get("use_topic_agent_v2", False),
            "use_structure_agent": input_data.get("use_structure_agent", True)
        }

        # 执行工作流
        result = self.compiled_graph.invoke(initial_state)

        logger.info("=== What 解构工作流执行完成（视频分析） ===")

        # 如果走的是 topic_agent_v2 分支，返回 topic_selection_v2 结果
        if result.get("use_topic_agent_v2") and "topic_selection_v2" in result:
            return result.get("topic_selection_v2", {})
        
        # 如果走的是 structure_agent 分支，返回 structure_data 结果
        if result.get("use_structure_agent") and "structure_data" in result:
            return result.get("structure_data", {})

        return result.get("final_result", {})