xiaoxia-saas/tests/unit/test_mediakit_video_understand.py

"""#1209 MediaKit 视频理解 + AI 智能选片段集成测试.

覆盖范围：
1. MediaKitClient.analyze_videos — 正常流程、不可用降级、空输入
2. _call_ai_recommend_service 带 asset_analyses — prompt 注入验证
3. _build_asset_analyses — API 路由层集成逻辑
4. run_ai_recommend — asset_analyses 透传
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

# ── MediaKitClient.analyze_videos ─────────────────────────────────────────────


class TestMediaKitClientAnalyzeVideos:
    """MediaKitClient.analyze_videos 单元测试."""

    def _make_client(self, api_key: str = "test-key") -> object:
        from packages.shared.mediakit_client import MediaKitClient

        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value = MagicMock(
                mediakit_api_key=api_key,
                mediakit_base_url="https://mock.mediakit.com/api/v1",
                mediakit_timeout=30,
            )
            return MediaKitClient()

    def test_analyze_videos_not_available(self):
        """未配置 API Key 时返回 None."""
        client = self._make_client(api_key="")
        result = client.analyze_videos(
            video_urls=["https://example.com/video.mp4"],
            prompt="describe this video",
        )
        assert result is None

    def test_analyze_videos_empty_urls(self):
        """空 URL 列表返回 None."""
        client = self._make_client()
        result = client.analyze_videos(video_urls=[], prompt="describe")
        assert result is None

    @patch("packages.shared.mediakit_client.httpx.post")
    @patch("packages.shared.mediakit_client.httpx.get")
    def test_analyze_videos_success(self, mock_get, mock_post):
        """正常提交任务并获取结果."""
        # 提交任务返回 task_id
        mock_post.return_value = MagicMock(
            status_code=200,
            json=lambda: {"success": True, "task_id": "test-task-123"},
        )
        mock_post.return_value.raise_for_status = MagicMock()

        # 轮询返回 completed
        mock_get.return_value = MagicMock(
            status_code=200,
            json=lambda: {
                "success": True,
                "task_id": "test-task-123",
                "status": "completed",
                "result": {
                    "duration": 30.5,
                    "contents": [
                        "视频展示了城市日景，包含车流和行人，氛围繁忙。",
                        "视频展示了夜晚霓虹灯特写，色调偏暖。",
                    ],
                },
            },
        )
        mock_get.return_value.raise_for_status = MagicMock()

        client = self._make_client()
        result = client.analyze_videos(
            video_urls=["https://example.com/v1.mp4", "https://example.com/v2.mp4"],
            prompt="描述视频内容",
            level="Economy",
            poll_interval=0.01,  # 测试用快速轮询
        )

        assert result is not None
        assert len(result) == 2
        assert "城市日景" in result[0]
        assert "霓虹灯" in result[1]

        # 验证提交参数
        call_args = mock_post.call_args
        assert "/tools/video-understand-router" in call_args.args[0]
        payload = call_args.kwargs["json"]
        assert payload["video_urls"] == ["https://example.com/v1.mp4", "https://example.com/v2.mp4"]
        assert payload["prompt"] == "描述视频内容"
        assert payload["level"] == "Economy"

    @patch("packages.shared.mediakit_client.httpx.post")
    def test_analyze_videos_submit_failure(self, mock_post):
        """提交任务失败返回 None."""
        mock_post.side_effect = Exception("Network error")

        client = self._make_client()
        result = client.analyze_videos(
            video_urls=["https://example.com/v1.mp4"],
            prompt="describe",
            poll_interval=0.01,
        )
        assert result is None

    @patch("packages.shared.mediakit_client.httpx.post")
    @patch("packages.shared.mediakit_client.httpx.get")
    def test_analyze_videos_task_failed(self, mock_get, mock_post):
        """任务状态为 failed 时返回 None."""
        mock_post.return_value = MagicMock(
            status_code=200,
            json=lambda: {"task_id": "task-fail"},
        )
        mock_post.return_value.raise_for_status = MagicMock()

        mock_get.return_value = MagicMock(
            status_code=200,
            json=lambda: {"status": "failed", "error": "model timeout"},
        )
        mock_get.return_value.raise_for_status = MagicMock()

        client = self._make_client()
        result = client.analyze_videos(
            video_urls=["https://example.com/v.mp4"],
            prompt="describe",
            poll_interval=0.01,
        )
        assert result is None

    @patch("packages.shared.mediakit_client.httpx.post")
    @patch("packages.shared.mediakit_client.httpx.get")
    def test_analyze_videos_timeout(self, mock_get, mock_post):
        """超过最大轮询次数返回 None."""
        mock_post.return_value = MagicMock(
            status_code=200,
            json=lambda: {"task_id": "task-slow"},
        )
        mock_post.return_value.raise_for_status = MagicMock()

        # 一直返回 processing
        mock_get.return_value = MagicMock(
            status_code=200,
            json=lambda: {"status": "processing"},
        )
        mock_get.return_value.raise_for_status = MagicMock()

        client = self._make_client()
        result = client.analyze_videos(
            video_urls=["https://example.com/v.mp4"],
            prompt="describe",
            poll_interval=0.001,
            max_poll_attempts=3,
        )
        assert result is None


# ── _call_ai_recommend_service with asset_analyses ────────────────────────────


class TestCallAiRecommendWithAnalysis:
    """测试 _call_ai_recommend_service 带 asset_analyses 的行为."""

    @patch("packages.shared.ai_service.get_doubao_client")
    def test_analysis_included_in_prompt(self, mock_get_client):
        """有分析结果时，prompt 包含视频内容描述."""
        mock_client = MagicMock()
        mock_client.is_available = True
        mock_get_client.return_value = mock_client

        # 模拟豆包返回一个有效 JSON
        mock_client.chat_completion.return_value = (
            '{"clips": [{"clip_type": "intro", "order": 0, "text_content": "开场",'
            '"duration": 3.0, "transition_effect": "fade", "asset_id": "asset1",'
            '"start_time": 0.0, "config": {}}, {"clip_type": "outro", "order": 1,'
            '"text_content": "结尾", "duration": 3.0, "transition_effect": "fade",'
            '"asset_id": "", "start_time": 0.0, "config": {}}],'
            '"title": "测试视频", "confidence": 0.9}'
        )

        from packages.shared.ai_service import _call_ai_recommend_service

        result = _call_ai_recommend_service(
            plan_id="plan-1",
            template_id="tmpl-1",
            asset_ids=["asset1", "asset2"],
            editing_mode="one_take",
            target_duration=30.0,
            asset_analyses={
                "asset1": "室内场景，一位女性在桌前讲解产品，氛围轻松专业",
                "asset2": "室外公园，阳光充足，有孩子在玩耍",
            },
        )

        assert result is not None
        assert len(result["clips"]) == 2

        # 验证 prompt 中包含了视频分析内容
        messages = mock_client.chat_completion.call_args.kwargs["messages"]
        system_prompt = messages[0]["content"]
        assert "叙事连贯" in system_prompt  # 有分析时的额外指导

        user_prompt = messages[1]["content"]
        assert "室内场景" in user_prompt
        assert "室外公园" in user_prompt

    @patch("packages.shared.ai_service.get_doubao_client")
    def test_no_analysis_basic_prompt(self, mock_get_client):
        """无分析结果时，prompt 保持基本格式."""
        mock_client = MagicMock()
        mock_client.is_available = True
        mock_get_client.return_value = mock_client

        mock_client.chat_completion.return_value = (
            '{"clips": [{"clip_type": "showcase", "order": 0, "text_content": "展示",'
            '"duration": 5.0, "transition_effect": "cut", "asset_id": "a1",'
            '"start_time": 0.0, "config": {}}],'
            '"title": "简单视频", "confidence": 0.8}'
        )

        from packages.shared.ai_service import _call_ai_recommend_service

        result = _call_ai_recommend_service(
            plan_id="plan-2",
            template_id="tmpl-1",
            asset_ids=["a1"],
            editing_mode="one_take",
            target_duration=10.0,
            # 不传 asset_analyses
        )

        assert result is not None

        # 验证 prompt 中不含叙事连贯等指导
        messages = mock_client.chat_completion.call_args.kwargs["messages"]
        system_prompt = messages[0]["content"]
        assert "叙事连贯" not in system_prompt

    @patch("packages.shared.ai_service.get_doubao_client")
    def test_partial_analysis_only_some_assets(self, mock_get_client):
        """部分素材有分析结果时，只有被分析的素材包含内容描述."""
        mock_client = MagicMock()
        mock_client.is_available = True
        mock_get_client.return_value = mock_client

        mock_client.chat_completion.return_value = (
            '{"clips": [{"clip_type": "showcase", "order": 0, "text_content": "展示",'
            '"duration": 5.0, "transition_effect": "cut", "asset_id": "a1",'
            '"start_time": 0.0, "config": {}}],'
            '"title": "部分分析", "confidence": 0.85}'
        )

        from packages.shared.ai_service import _call_ai_recommend_service

        _call_ai_recommend_service(
            plan_id="plan-3",
            template_id="tmpl-1",
            asset_ids=["a1", "a2", "a3"],
            editing_mode="one_take",
            target_duration=15.0,
            asset_analyses={
                "a1": "海边日落，金色阳光",  # 只有 a1 有分析
            },
        )

        messages = mock_client.chat_completion.call_args.kwargs["messages"]
        user_prompt = messages[1]["content"]
        assert "海边日落" in user_prompt
        # a2 和 a3 应该只有 ID，没有内容描述
        assert "素材ID: a2" in user_prompt
        assert "素材ID: a3" in user_prompt


# ── run_ai_recommend with asset_analyses ──────────────────────────────────────


class TestRunAiRecommendWithAnalysis:
    """run_ai_recommend 透传 asset_analyses."""

    @patch("packages.shared.ai_service._call_ai_recommend_service")
    def test_passes_through_analysis(self, mock_call):
        """asset_analyses 正确传递给底层服务."""
        from packages.shared.ai_service import run_ai_recommend

        mock_call.return_value = {
            "clips": [
                {
                    "clip_type": "showcase",
                    "order": 0,
                    "text_content": "t",
                    "duration": 3.0,
                    "transition_effect": "cut",
                    "asset_id": "a1",
                    "start_time": 0.0,
                    "config": {},
                }
            ],
            "config": {"title": {"text": "test", "ai_auto": True}},
            "total_duration": 3.0,
            "confidence": 0.85,
        }

        analyses = {"a1": "海边日落场景"}
        run_ai_recommend(
            plan_id="plan-1",
            template_id="tmpl-1",
            asset_ids=["a1"],
            editing_mode="one_take",
            target_duration=10.0,
            asset_analyses=analyses,
        )

        # 验证传递
        call_kwargs = mock_call.call_args.kwargs
        assert call_kwargs["asset_analyses"] == analyses

    @patch("packages.shared.ai_service._call_ai_recommend_service")
    def test_default_no_analysis(self, mock_call):
        """默认不传 asset_analyses 时为 None."""
        from packages.shared.ai_service import run_ai_recommend

        mock_call.return_value = {
            "clips": [
                {
                    "clip_type": "showcase",
                    "order": 0,
                    "text_content": "t",
                    "duration": 3.0,
                    "transition_effect": "cut",
                    "asset_id": "a1",
                    "start_time": 0.0,
                    "config": {},
                }
            ],
            "config": {"title": {"text": "test", "ai_auto": True}},
            "total_duration": 3.0,
            "confidence": 0.85,
        }

        run_ai_recommend(
            plan_id="plan-1",
            template_id="tmpl-1",
            asset_ids=["a1"],
        )

        call_kwargs = mock_call.call_args.kwargs
        assert call_kwargs["asset_analyses"] is None


# ── _build_asset_analyses (API route helper) ─────────────────────────────────


class TestBuildAssetAnalyses:
    """_build_asset_analyses 集成逻辑测试."""

    def test_empty_asset_ids(self):
        """空素材列表返回空 dict."""
        from apps.api.app.api.routes.templates_editor.ai_features import (
            _build_asset_analyses,
        )

        result = _build_asset_analyses([], MagicMock())
        assert result == {}

    @patch("packages.shared.mediakit_client.get_mediakit_client")
    def test_mediakit_not_available(self, mock_get_client):
        """MediaKit 不可用时返回空 dict."""
        mock_client = MagicMock()
        mock_client.is_available = False
        mock_get_client.return_value = mock_client

        # 需要 patch 在 packages.shared.mediakit_client 层面，因为 _build_asset_analyses 内部 from import
        from apps.api.app.api.routes.templates_editor.ai_features import (
            _build_asset_analyses,
        )

        result = _build_asset_analyses(["asset1"], MagicMock())
        assert result == {}

    def test_exception_returns_empty(self):
        """任何异常都返回空 dict，不阻塞主流程."""
        from apps.api.app.api.routes.templates_editor.ai_features import (
            _build_asset_analyses,
        )

        # 传一个 mock db，让内部自然失败
        mock_db = MagicMock()
        mock_db.side_effect = None  # db 本身不抛异常，但内部操作会失败

        result = _build_asset_analyses(["nonexistent-asset"], mock_db)
        assert result == {}


# ── 长文本截断验证 ───────────────────────────────────────────────────────────


class TestAnalysisTruncation:
    """验证过长分析文本被截断以避免 token 爆炸."""

    @patch("packages.shared.ai_service.get_doubao_client")
    def test_long_analysis_truncated(self, mock_get_client):
        """超过 300 字的分析文本被截断并加 ...."""
        mock_client = MagicMock()
        mock_client.is_available = True
        mock_get_client.return_value = mock_client

        mock_client.chat_completion.return_value = (
            '{"clips": [{"clip_type": "showcase", "order": 0, "text_content": "t",'
            '"duration": 3.0, "transition_effect": "cut", "asset_id": "a1",'
            '"start_time": 0.0, "config": {}}],'
            '"title": "test", "confidence": 0.8}'
        )

        from packages.shared.ai_service import _call_ai_recommend_service

        long_analysis = "A" * 500  # 500 字符

        _call_ai_recommend_service(
            plan_id="plan-1",
            template_id="tmpl-1",
            asset_ids=["a1"],
            editing_mode="one_take",
            target_duration=10.0,
            asset_analyses={"a1": long_analysis},
        )

        messages = mock_client.chat_completion.call_args.kwargs["messages"]
        user_prompt = messages[1]["content"]
        # 截断后应该包含 ...
        assert "..." in user_prompt
        # 原始 500 字符不应完整出现
        assert "A" * 500 not in user_prompt