"""P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验.

覆盖：
  P0-1: _download_library_assets 双模式查询（asset_library_id / project_id）
  P0-2: OSS 上传失败抛异常 + URL 可访问性校验
  P0-3: FFmpeg 失败时完整 stderr 日志
  P1:   template_id 存在性校验 + asset_ids 归属校验
"""

from __future__ import annotations

import logging
import subprocess
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# 添加 worker app 到 sys.path
_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
if str(_WORKER_ROOT) not in sys.path:
    sys.path.insert(0, str(_WORKER_ROOT))


# ── P0-1: _download_library_assets ────────────────────────────────────────────


class TestDownloadLibraryAssets:
    """P0-1: 素材下载双模式 + 错误处理."""

    def _make_asset(self, id_: str, file_url: str, project_id: str = "p1", library_id: str = "lib1"):
        mock = MagicMock()
        mock.id = id_
        mock.file_url = file_url
        mock.name = f"asset_{id_}"
        mock.project_id = project_id
        mock.asset_library_id = library_id
        return mock

    @patch("worker_app.tasks.generation.SessionLocal")
    @patch("worker_app.tasks.generation.download_asset")
    def test_asset_library_mode(self, mock_download, mock_session_factory):
        """素材库模式：按 asset_library_id 查询."""
        from worker_app.tasks.generation import _download_library_assets

        session = MagicMock()
        mock_session_factory.return_value = session
        query = MagicMock()
        session.query.return_value = query
        filter_result = MagicMock()
        query.filter.return_value = filter_result
        in_filter = MagicMock()
        filter_result.filter.return_value = in_filter
        assets = [self._make_asset("a1", "video/a1.mp4")]
        in_filter.order_by.return_value.all.return_value = assets

        mock_download.return_value = True

        with patch("worker_app.tasks.generation.AssetModel", create=True):
            result = _download_library_assets(
                Path("/tmp/test"),
                asset_library_id="lib1",
            )

        assert len(result) == 1
        mock_download.assert_called_once()

    @patch("worker_app.tasks.generation.SessionLocal")
    @patch("worker_app.tasks.generation.download_asset")
    def test_project_mode(self, mock_download, mock_session_factory):
        """项目级模式：asset_library_id 为空时按 project_id 查询."""
        from worker_app.tasks.generation import _download_library_assets

        session = MagicMock()
        mock_session_factory.return_value = session
        query = MagicMock()
        session.query.return_value = query
        filter_result = MagicMock()
        query.filter.return_value = filter_result
        proj_filter = MagicMock()
        filter_result.filter.return_value = proj_filter
        assets = [self._make_asset("a1", "video/a1.mp4", project_id="proj1")]
        proj_filter.order_by.return_value.all.return_value = assets

        mock_download.return_value = True

        result = _download_library_assets(
            Path("/tmp/test"),
            project_id="proj1",
        )

        assert len(result) == 1

    def test_both_empty_raises(self):
        """asset_library_id 和 project_id 都为空时抛 ValueError."""
        from worker_app.tasks.generation import _download_library_assets

        with pytest.raises(ValueError, match="至少需要提供一个"):
            _download_library_assets(Path("/tmp/test"))

    @patch("worker_app.tasks.generation.SessionLocal")
    def test_no_assets_found_raises(self, mock_session_factory):
        """查不到素材时抛 RuntimeError."""
        from worker_app.tasks.generation import _download_library_assets

        session = MagicMock()
        mock_session_factory.return_value = session
        query = MagicMock()
        session.query.return_value = query
        filter_result = MagicMock()
        query.filter.return_value = filter_result
        in_filter = MagicMock()
        filter_result.filter.return_value = in_filter
        in_filter.order_by.return_value.all.return_value = []

        with pytest.raises(RuntimeError, match="未找到视频素材"):
            _download_library_assets(
                Path("/tmp/test"),
                asset_library_id="lib1",
            )

    @patch("worker_app.tasks.generation.SessionLocal")
    @patch("worker_app.tasks.generation.download_asset")
    def test_all_asset_ids_fail_raises(self, mock_download, mock_session_factory):
        """指定 asset_ids 但全部下载失败时抛 RuntimeError."""
        from worker_app.tasks.generation import _download_library_assets

        session = MagicMock()
        mock_session_factory.return_value = session
        query = MagicMock()
        session.query.return_value = query
        filter_result = MagicMock()
        query.filter.return_value = filter_result
        id_filter = MagicMock()
        filter_result.filter.return_value = id_filter
        assets = [self._make_asset("a1", "video/a1.mp4")]
        id_filter.order_by.return_value.all.return_value = assets

        mock_download.return_value = False  # 全部下载失败

        with pytest.raises(RuntimeError, match="素材下载失败"):
            _download_library_assets(
                Path("/tmp/test"),
                asset_library_id="lib1",
                asset_ids=["a1"],
            )


# ── P0-2: OSS 上传 + URL 校验 ────────────────────────────────────────────────


class TestOSSUploadAndVerify:
    """P0-2: OSS 上传失败抛异常 + URL 可访问性校验."""

    def test_verify_url_accessible_success(self):
        """URL 可访问时返回 True."""
        from worker_app.tasks.generation import _verify_url_accessible

        mock_response = MagicMock()
        mock_response.status = 200
        mock_response.__enter__ = MagicMock(return_value=mock_response)
        mock_response.__exit__ = MagicMock(return_value=False)

        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
            assert _verify_url_accessible("https://example.com/test.mp4") is True

    def test_verify_url_accessible_failure(self):
        """URL 不可访问时返回 False."""
        from worker_app.tasks.generation import _verify_url_accessible

        with patch("urllib.request.OpenerDirector.open", side_effect=Exception("connection refused")):
            assert _verify_url_accessible("https://example.com/test.mp4") is False

    def test_verify_url_404(self):
        """URL 返回 404 时返回 False."""
        from worker_app.tasks.generation import _verify_url_accessible

        mock_response = MagicMock()
        mock_response.status = 404
        mock_response.__enter__ = MagicMock(return_value=mock_response)
        mock_response.__exit__ = MagicMock(return_value=False)

        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
            assert _verify_url_accessible("https://example.com/test.mp4") is False


# ── P0-3: FFmpeg stderr 日志 ─────────────────────────────────────────────────


class TestFFmpegStderrLogging:
    """P0-3: FFmpeg 失败时完整 stderr 打到日志."""

    def test_run_ffmpeg_logs_stderr_on_failure(self, caplog):
        """run_ffmpeg 失败时记录 stderr 到日志."""
        from video_processing.ffmpeg_utils import run_ffmpeg

        error = subprocess.CalledProcessError(
            returncode=183,
            cmd=["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"],
            output="",
            stderr="Error message from ffmpeg: filter graph error details here",
        )

        with patch("subprocess.run", side_effect=error):
            with caplog.at_level(logging.ERROR):
                with pytest.raises(subprocess.CalledProcessError):
                    run_ffmpeg(["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"])

        assert "FFmpeg 命令失败" in caplog.text
        assert "exit_code=183" in caplog.text
        assert "filter graph error" in caplog.text

    def test_run_ffmpeg_success(self):
        """run_ffmpeg 成功时正常返回."""
        from video_processing.ffmpeg_utils import run_ffmpeg

        mock_result = MagicMock()
        mock_result.stdout = "output"
        mock_result.stderr = ""

        with patch("subprocess.run", return_value=mock_result):
            stdout, stderr = run_ffmpeg(["ffmpeg", "-version"])
            assert stdout == "output"
            assert stderr == ""


# ── P1: template_id + asset_ids 校验 ─────────────────────────────────────────


class TestP1Validations:
    """P1: template_id 存在性校验 + asset_ids 归属校验."""

    def test_validate_template_exists_success(self):
        """模板存在时不抛异常."""
        from worker_app.tasks.generation import _validate_template_exists

        mock_template = MagicMock()
        mock_template.id = "tmpl_001"
        mock_template.name = "Test Template"
        mock_template.is_active = True
        mock_template.status = "active"
        mock_template.version = 1

        session = MagicMock()

        # EditTemplateModel 查询返回 None（走旧模板系统 fallback）
        edit_query = MagicMock()
        edit_filter = MagicMock()
        edit_query.filter.return_value = edit_filter
        edit_filter.first.return_value = None

        # TemplateModel 查询返回 mock_template
        old_query = MagicMock()
        old_filter = MagicMock()
        old_query.filter.return_value = old_filter
        old_filter.first.return_value = mock_template

        def _query_side_effect(model):
            # 根据 model 类型返回不同的 query mock
            name = getattr(model, "__name__", "")
            if "EditTemplate" in name:
                return edit_query
            return old_query

        session.query.side_effect = _query_side_effect

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            _validate_template_exists("tmpl_001")  # 不抛异常

    def test_validate_template_exists_not_found(self):
        """模板不存在时抛 ValueError."""
        from worker_app.tasks.generation import _validate_template_exists

        session = MagicMock()

        # 两个系统查询都返回 None
        def _query_side_effect(model):
            q = MagicMock()
            f = MagicMock()
            q.filter.return_value = f
            f.first.return_value = None
            return q

        session.query.side_effect = _query_side_effect

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            with pytest.raises(ValueError, match="模板不存在"):
                _validate_template_exists("tmpl_nonexistent")


# ── P1: 一键生成 clip 级效果层映射 ───────────────────────────────────────────


class TestTemplateClipEffectMapping:
    """P1: 模板 clip 级效果层映射到一键生成素材 clips."""

    def _make_virtual_clip(self, idx: int, clip_type: str = "main", config: dict | None = None):
        from dataclasses import dataclass, field

        _clip_type_val = clip_type

        @dataclass
        class FakeClip:
            id: str = f"vc_{idx:03d}"
            plan_id: str = "task_001"
            clip_type: str = _clip_type_val
            order: int = idx
            asset_id: str = f"asset_{idx}"
            duration: float = 5.0
            transition_effect: str = "cut"
            transition_duration: float = 0.0
            playback_speed: float = 1.0
            config: dict = field(default_factory=dict)

        return FakeClip(config=config or {})

    def _make_template_clip_config(self, clip_type: str = "main", transition: str = "cut", config: dict | None = None):
        mock = MagicMock()
        mock.clip_type = clip_type
        mock.transition_effect = transition
        mock.config = config or {}
        mock.default_duration = 3.0
        mock.text_template = ""
        return mock

    def test_transition_effect_mapped(self):
        """转场效果正确映射到素材 clips."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(i) for i in range(3)]
        clip_configs = [
            self._make_template_clip_config("main", transition="fade"),
            self._make_template_clip_config("main", transition="dissolve"),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        # 前两个按顺序映射，第三个用最后一个模板配置
        assert clips[0].transition_effect == "fade"
        assert clips[1].transition_effect == "dissolve"
        assert clips[2].transition_effect == "dissolve"  # 复用最后一个

    def test_color_grade_mapped(self):
        """滤镜配置正确映射到 clip.config.color_grade."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(i) for i in range(2)]
        clip_configs = [
            self._make_template_clip_config(
                "main", config={"color_grade": {"enabled": True, "filter": "vintage", "brightness": 0.1}}
            ),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        assert clips[0].config["color_grade"]["filter"] == "vintage"
        assert clips[0].config["color_grade"]["brightness"] == 0.1
        # 第二个素材复用第一个模板配置
        assert clips[1].config["color_grade"]["filter"] == "vintage"

    def test_existing_config_preserved(self):
        """已有 clip.config 内容（如 role）被保留."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(0, config={"role": "b_roll"})]
        clip_configs = [
            self._make_template_clip_config("main", config={"color_grade": {"enabled": True, "filter": "warm"}}),
        ]

        _apply_template_clip_effects(clips, clip_configs, "voice_over")

        assert clips[0].config["role"] == "b_roll"  # 保留原有配置
        assert clips[0].config["color_grade"]["filter"] == "warm"  # 新增滤镜配置

    def test_empty_clip_configs_no_change(self):
        """空模板配置时 clips 保持不变."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(i) for i in range(2)]
        _apply_template_clip_effects(clips, [], "one_take")

        assert clips[0].transition_effect == "cut"
        assert clips[1].transition_effect == "cut"

    def test_cut_transition_not_overwritten(self):
        """模板转场为 cut 时不覆盖（保持默认）."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(0)]
        clips[0].transition_effect = "fade"  # 已有非默认值
        clip_configs = [
            self._make_template_clip_config("main", transition="cut"),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        # 模板是 cut 时，保留原有值（避免无意义覆盖）
        assert clips[0].transition_effect == "fade"

    def test_transition_duration_mapped(self):
        """转场时长（transition_duration）从模板 config 正确映射到 clip."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(i) for i in range(3)]
        clip_configs = [
            self._make_template_clip_config("main", transition="fade", config={"transition_duration": 0.8}),
            self._make_template_clip_config("main", transition="dissolve", config={"transition_duration": 1.2}),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        # 前两个按顺序映射，第三个复用最后一个
        assert clips[0].transition_effect == "fade"
        assert clips[0].transition_duration == 0.8
        assert clips[1].transition_effect == "dissolve"
        assert clips[1].transition_duration == 1.2
        assert clips[2].transition_effect == "dissolve"
        assert clips[2].transition_duration == 1.2

    def test_transition_duration_ignored_for_cut(self):
        """模板转场为 cut 时，transition_duration 不生效（保持默认0）."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(0)]
        clip_configs = [
            self._make_template_clip_config("main", transition="cut", config={"transition_duration": 0.5}),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        # cut 转场不映射，transition_duration 也不应用
        assert clips[0].transition_duration == 0.0

    def test_transition_duration_invalid_value_skipped(self):
        """transition_duration 为无效值时安全跳过."""
        from worker_app.tasks.generation import _apply_template_clip_effects

        clips = [self._make_virtual_clip(0)]
        clip_configs = [
            self._make_template_clip_config("main", transition="fade", config={"transition_duration": "abc"}),
        ]

        _apply_template_clip_effects(clips, clip_configs, "one_take")

        assert clips[0].transition_effect == "fade"
        assert clips[0].transition_duration == 0.0  # 无效值保持默认

    def test_intro_outro_extracted(self):
        """intro/outro 类型 clip_config 正确提取为 plan 级 intro_outro 配置."""
        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs

        clip_configs = [
            self._make_template_clip_config("intro", config={"intro_type": "text", "intro_text_color": "#ffffff"}),
            self._make_template_clip_config("main"),
            self._make_template_clip_config("outro", config={"outro_type": "follow", "outro_follow_text": "关注我们"}),
        ]
        # 设置 intro/outro 的 text_template
        clip_configs[0].text_template = "精彩视频"
        clip_configs[0].default_duration = 2.5

        result = _extract_intro_outro_from_clip_configs(clip_configs)

        assert result["has_intro"] is True
        assert result["intro_type"] == "text"
        assert result["intro_text"] == "精彩视频"
        assert result["intro_duration"] == 2.5
        assert result["intro_text_color"] == "#ffffff"
        assert result["has_outro"] is True
        assert result["outro_type"] == "follow"
        assert result["outro_follow_text"] == "关注我们"

    def test_intro_outro_empty_when_none(self):
        """没有 intro/outro 时返回空 dict."""
        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs

        clip_configs = [
            self._make_template_clip_config("main"),
            self._make_template_clip_config("main"),
        ]

        result = _extract_intro_outro_from_clip_configs(clip_configs)
        assert result == {}


class TestTemplatePlanConfigLoading:
    """验证从模板加载 plan 级配置（BGM、字幕、标题）的逻辑。"""

    def _mock_template(
        self,
        title_config=None,
        subtitle_config=None,
        bgm_config=None,
        is_active=True,
    ):
        template = MagicMock()
        template.id = "tmpl_001"
        template.name = "Test Template"
        template.is_active = is_active
        template.title_config = title_config or {}
        template.subtitle_config = subtitle_config or {}
        template.bgm_config = bgm_config or {}
        return template

    def _mock_session(self, template):
        session = MagicMock()

        # EditTemplateModel 查询返回 None（走旧模板系统 fallback）
        edit_query = MagicMock()
        edit_filter = MagicMock()
        edit_query.filter.return_value = edit_filter
        edit_filter.first.return_value = None

        # TemplateModel 查询返回 template（旧模板系统）
        old_query = MagicMock()
        old_filter = MagicMock()
        old_query.filter.return_value = old_filter
        old_filter.first.return_value = template

        def _query_side_effect(model):
            name = getattr(model, "__name__", "")
            if "EditTemplate" in name:
                return edit_query
            return old_query

        session.query.side_effect = _query_side_effect
        return session

    def test_load_template_config_assembles_three_fields(self):
        """模板的三个独立字段正确组装成 plan.config 格式。"""
        from worker_app.tasks.generation import _load_template_plan_config

        title_cfg = {"enabled": True, "text": "我的标题", "font_size": 36}
        subtitle_cfg = {"enabled": True, "auto_generated": True, "language": "zh"}
        bgm_cfg = {"enabled": True, "preset_id": "bgm-001", "volume": 0.5}

        template = self._mock_template(
            title_config=title_cfg,
            subtitle_config=subtitle_cfg,
            bgm_config=bgm_cfg,
        )
        session = self._mock_session(template)

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            result = _load_template_plan_config("tmpl_001")

        assert result["title"] == title_cfg
        assert result["subtitle"] == subtitle_cfg
        assert result["bgm"] == bgm_cfg

    def test_load_template_config_empty_template_returns_empty(self):
        """模板三个字段都为空时返回空 dict。"""
        from worker_app.tasks.generation import _load_template_plan_config

        template = self._mock_template()
        session = self._mock_session(template)

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            result = _load_template_plan_config("tmpl_001")

        assert result == {}

    def test_load_template_config_only_bgm(self):
        """只有 BGM 配置时只返回 bgm 字段。"""
        from worker_app.tasks.generation import _load_template_plan_config

        bgm_cfg = {"enabled": True, "audio_url": "https://example.com/bgm.mp3"}
        template = self._mock_template(bgm_config=bgm_cfg)
        session = self._mock_session(template)

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            result = _load_template_plan_config("tmpl_001")

        assert "bgm" in result
        assert result["bgm"] == bgm_cfg
        assert "title" not in result
        assert "subtitle" not in result

    def test_load_template_config_empty_template_id(self):
        """空 template_id 直接返回空 dict。"""
        from worker_app.tasks.generation import _load_template_plan_config

        result = _load_template_plan_config("")
        assert result == {}

        result = _load_template_plan_config(None)
        assert result == {}

    def test_load_template_config_not_found_returns_empty(self):
        """模板不存在时返回空 dict，不抛异常。"""
        from worker_app.tasks.generation import _load_template_plan_config

        session = MagicMock()

        def _query_side_effect(model):
            q = MagicMock()
            f = MagicMock()
            q.filter.return_value = f
            f.first.return_value = None
            return q

        session.query.side_effect = _query_side_effect

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            result = _load_template_plan_config("tmpl_nonexist")

        assert result == {}