from django.test import TestCase
from unittest.mock import patch

from rest_framework.test import APIClient

from apps.accounts.models import Team, TeamMember, User
from apps.ai.models import AITask, ModelConfig, ModelProvider
from apps.assets.models import Asset
from apps.billing.models import CreditAccount, CreditLedger
from apps.products.models import Product
from apps.projects.models import (
    Project,
    ProjectStage,
    ScriptSegment,
    ScriptVersion,
    SubtitleTrack,
    Timeline,
    TimelineClip,
    VideoSegment,
    VideoSegmentVersion,
)


class ProjectApiTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username="owner", password="pass")
        self.team = Team.objects.create(name="E2E Team", owner=self.user)
        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
        CreditAccount.objects.create(team=self.team, balance="100.0000")
        self.product = Product.objects.create(team=self.team, created_by=self.user, title="Test Product")
        self.provider = ModelProvider.objects.create(
            name="volcengine",
            display_name="Volcano",
            base_url="https://ark.cn-beijing.volces.com/api/v3",
        )
        self.model = ModelConfig.objects.create(
            provider=self.provider,
            name="doubao-seed-2-0-pro-260215",
            display_name="Doubao",
            capability=ModelConfig.Capability.TEXT,
            endpoint="chat/completions",
            unit_price="2.0000",
        )
        # 数据迁移(0006)会 seed 中转站文本模型,created_at 更早 → get_default_model 会选到它,
        # 从而路由到未被 patch 的 OpenAICompatibleProvider 打真网络。这里只保留自建 volcengine 模型为
        # active,确保命中可 mock 的 VolcanoArkProvider(生产有 bootstrap 豆包在前,默认仍是豆包,不受影响)。
        ModelConfig.objects.filter(capability=ModelConfig.Capability.TEXT).exclude(id=self.model.id).update(status="disabled")
        self.client = APIClient()
        self.client.force_authenticate(self.user)

    def test_create_project_initializes_pipeline(self):
        response = self.client.post(
            "/api/projects/",
            {"name": "Launch Video", "product": str(self.product.id)},
            format="json",
        )

        self.assertEqual(response.status_code, 201)
        project = Project.objects.get(id=response.data["id"])
        self.assertEqual(project.team, self.team)
        self.assertEqual(project.created_by, self.user)
        self.assertEqual(project.stages.count(), 5)
        self.assertEqual(project.video_segments.count(), 4)
        self.assertEqual(
            list(project.stages.values_list("stage", flat=True)),
            [
                ProjectStage.Stage.SCRIPT,
                ProjectStage.Stage.BASE_ASSETS,
                ProjectStage.Stage.STORYBOARD,
                ProjectStage.Stage.VIDEO,
                ProjectStage.Stage.EXPORT,
            ],
        )
        self.assertEqual(
            list(project.video_segments.values_list("target_duration_seconds", flat=True)),
            [15, 15, 15, 15],
        )
        self.assertEqual(
            list(project.video_segments.values_list("sort_order", flat=True)),
            [0, 1, 2, 3],
        )
        self.assertTrue(VideoSegment.objects.filter(project=project).exists())

    @patch("apps.ai.services.VolcanoArkProvider")
    def test_rerun_script_segment_via_agent_rewrites_one_and_charges_once(self, provider_cls):
        """单镜重跑改走 agent:读结构化全脚本,只重写目标镜、保留其余镜,落新版本,计费一次。"""
        import json as _json
        base_draft = {
            "hook": "钩子", "tone": "种草", "aspect_ratio": "9:16", "total_duration": 30, "segment_count": 2,
            "entities": [],
            "segments": [
                {"index": 0, "duration": 15, "role": "钩子", "narration": "旧0", "visual": "画0", "speaker": None, "product_exposure": "", "entity_refs": [], "dialogue": []},
                {"index": 1, "duration": 15, "role": "CTA", "narration": "旧1", "visual": "画1", "speaker": None, "product_exposure": "", "entity_refs": [], "dialogue": []},
            ],
        }
        new_draft = _json.loads(_json.dumps(base_draft))
        new_draft["segments"][1]["narration"] = "全新口播文案"
        new_draft["segments"][1]["visual"] = "全新画面描述"
        provider = provider_cls.return_value
        provider.chat_completion.return_value = {"choices": [{"message": {"content": "x"}}]}
        provider.extract_text.return_value = "```json\n" + _json.dumps(new_draft, ensure_ascii=False) + "\n```"

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
        script = ScriptVersion.objects.create(
            project=project, title="脚本", content=_json.dumps(base_draft, ensure_ascii=False),
            is_adopted=True, metadata={"hook": "钩子", "entities": []},
        )
        ScriptSegment.objects.create(script_version=script, sort_order=0, narration="旧0", visual_prompt="画0")
        seg1 = ScriptSegment.objects.create(script_version=script, sort_order=1, narration="旧1", visual_prompt="画1")

        response = self.client.post(
            f"/api/projects/{project.id}/rerun-script-segment/",
            {"segment_id": str(seg1.id), "instruction": "更俏皮"},
            format="json",
        )
        self.assertEqual(response.status_code, 200)
        # agent 落新 ScriptVersion:目标镜改了,其余镜保留
        new_version = ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
        segs = list(new_version.segments.order_by("sort_order"))
        self.assertEqual(segs[1].narration, "全新口播文案")
        self.assertEqual(segs[1].visual_prompt, "全新画面描述")
        self.assertEqual(segs[0].narration, "旧0")
        self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 1)

    def test_rerun_script_segment_rejects_foreign_segment(self):
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
        other = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="O")
        other_script = ScriptVersion.objects.create(project=other, title="脚本", content="...")
        other_seg = ScriptSegment.objects.create(script_version=other_script, sort_order=0, narration="x")

        response = self.client.post(
            f"/api/projects/{project.id}/rerun-script-segment/",
            {"segment_id": str(other_seg.id)},
            format="json",
        )
        self.assertEqual(response.status_code, 404)

    @patch("apps.ai.services.VolcanoArkProvider")
    def test_extract_entities_streams_and_persists(self, provider_cls):
        """提取走流式通道(豆包 2.0 Pro 思考模型):思考事件丢弃、只收正文 JSON →
        落库 cast/scenes + 每镜 entity_refs + entities_extracted 标记,计费一次。"""
        import json as _json
        extracted = {
            "entities": [
                {"id": "c1", "type": "character", "name": "女主", "visual_prompt": "26岁都市女性,利落短发"},
                {"id": "s1", "type": "scene", "name": "出租屋客厅", "visual_prompt": "ins风小客厅,暖白自然光"},
            ],
            "segments": [
                {"index": 0, "entity_refs": ["c1", "s1"]},
                {"index": 1, "entity_refs": ["c1", "s1"]},
            ],
        }
        provider = provider_cls.return_value
        provider.chat_completion_stream.return_value = iter([
            {"type": "reasoning", "text": "先通读分镜"},  # 思考事件:必须被丢弃,不进正文
            {"type": "delta", "text": "```json\n" + _json.dumps(extracted, ensure_ascii=False)},
            {"type": "delta", "text": "\n```"},
            {"type": "done"},
        ])

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
        script = ScriptVersion.objects.create(
            project=project, title="脚本", content="x", is_adopted=True, metadata={"entities": []},
        )
        ScriptSegment.objects.create(script_version=script, sort_order=0, narration="旧0", visual_prompt="画0")
        ScriptSegment.objects.create(script_version=script, sort_order=1, narration="旧1", visual_prompt="画1")

        response = self.client.post(f"/api/projects/{project.id}/extract-entities/", {}, format="json")
        self.assertEqual(response.status_code, 200)
        # 锁定豆包 2.0 Pro,且走的是流式通道(而非非流式 chat_completion)
        self.assertEqual(provider.chat_completion_stream.call_args.kwargs["model"], "doubao-seed-2-0-pro-260215")
        provider.chat_completion.assert_not_called()
        # 实体落库:cast/scenes + entities_extracted 标记 + 每镜 entity_refs 回填
        project.refresh_from_db()
        self.assertIn("女主", project.metadata.get("cast", []))
        self.assertIn("出租屋客厅", project.metadata.get("scenes", []))
        self.assertTrue(project.metadata.get("entities_extracted"))
        segs = list(script.segments.order_by("sort_order"))
        self.assertEqual(segs[0].entity_refs, ["c1", "s1"])
        self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 1)

    @patch("apps.ai.services.VolcanoArkProvider")
    def test_extract_entities_recovers_when_content_empty_uses_reasoning(self, provider_cls):
        """极端兜底:思考模型整轮只发了 reasoning、正文 content 为空 → 回退用 reasoning 里的 JSON,
        不再像旧非流式那样拿到空串报「未返回有效 JSON」。"""
        import json as _json
        extracted = {
            "entities": [{"id": "c1", "type": "character", "name": "男主", "visual_prompt": "28岁居家男性"}],
            "segments": [{"index": 0, "entity_refs": ["c1"]}],
        }
        provider = provider_cls.return_value
        provider.chat_completion_stream.return_value = iter([
            {"type": "reasoning", "text": _json.dumps(extracted, ensure_ascii=False)},
            {"type": "done"},
        ])

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P2")
        script = ScriptVersion.objects.create(
            project=project, title="脚本", content="x", is_adopted=True, metadata={"entities": []},
        )
        ScriptSegment.objects.create(script_version=script, sort_order=0, narration="旧0", visual_prompt="画0")

        response = self.client.post(f"/api/projects/{project.id}/extract-entities/", {}, format="json")
        self.assertEqual(response.status_code, 200)
        project.refresh_from_db()
        self.assertIn("男主", project.metadata.get("cast", []))

    @patch("apps.ai.services.VolcanoArkProvider")
    def test_extract_entities_inflight_guard_reuses_task(self, provider_cls):
        """已有在途提取任务时再次提交 → 复用同一任务,不新建、不二次预扣、不重跑模型
        (用户刷新后手痒重点 / 并发点击都不会重复扣费)。"""
        from apps.ai.services import submit_extract_entities

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P3")
        script = ScriptVersion.objects.create(
            project=project, title="脚本", content="x", is_adopted=True, metadata={"entities": []},
        )
        ScriptSegment.objects.create(script_version=script, sort_order=0, narration="旧0", visual_prompt="画0")
        inflight = AITask.objects.create(
            team=self.team, created_by=self.user, project=project, model_config=self.model,
            task_type=AITask.Type.ENTITY_EXTRACTION, status=AITask.Status.SUBMITTED,
            idempotency_key="entity_extraction:inflight:1",
        )
        before = AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION).count()

        task = submit_extract_entities(project=project, user=self.user)
        self.assertEqual(str(task.id), str(inflight.id))  # 复用在途任务
        self.assertEqual(  # 没新建任务
            AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION).count(), before
        )
        provider_cls.return_value.chat_completion_stream.assert_not_called()  # 没重跑模型
        # 没二次预扣额度(复用直接 return,根本没走到 create_ai_task)
        self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.RESERVE).count(), 0)

    def test_extract_status_reports_running_then_failure(self):
        """extract-status:在途→running=True;失败→running=False 且把可读错误透给前端(刷新后也能看到失败原因)。"""
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P4")
        t = AITask.objects.create(
            team=self.team, created_by=self.user, project=project, model_config=self.model,
            task_type=AITask.Type.ENTITY_EXTRACTION, status=AITask.Status.SUBMITTED,
            idempotency_key="entity_extraction:status:1",
        )
        res = self.client.get(f"/api/projects/{project.id}/extract-status/")
        self.assertEqual(res.status_code, 200)
        self.assertTrue(res.data["running"])

        t.status = AITask.Status.FAILED
        t.error_message = "提取结果解析失败(模型未返回有效 JSON),请重试"
        t.save(update_fields=["status", "error_message"])
        res = self.client.get(f"/api/projects/{project.id}/extract-status/")
        self.assertFalse(res.data["running"])
        self.assertEqual(res.data["status"], "failed")
        self.assertIn("未返回有效 JSON", res.data["error"])

    @patch("apps.ai.services._store_generated_media")
    @patch("apps.ai.services.get_image_provider")
    def test_generate_base_asset_stores_tag_label(self, get_provider, store_media):
        """流程步骤④:按脚本标签生成基础资产时,label 落进 group.metadata,供前端把结果归回标签卡。"""
        ModelConfig.objects.create(
            provider=self.provider, name="img-model", display_name="Img",
            capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
        )
        asset = Asset.objects.create(
            team=self.team, created_by=self.user, name="女主",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON,
        )
        store_media.return_value = asset
        provider = get_provider.return_value
        provider.image_generation.return_value = {"data": [{"url": "http://x/img.png"}]}
        provider.extract_first_media_url.return_value = "http://x/img.png"
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")

        response = self.client.post(
            f"/api/projects/{project.id}/generate-base-asset/",
            {"kind": "person", "prompt": "26岁都市女性", "label": "女主"},
            format="json",
        )
        # 异步:秒回 202 + 任务;EAGER 下 worker 任务就地执行,组已落库,label 进 metadata
        self.assertEqual(response.status_code, 202)
        self.assertIn("task", response.data)
        from apps.projects.models import BaseAssetGroup
        group = BaseAssetGroup.objects.get(project=project, kind=BaseAssetGroup.Kind.PERSON)
        self.assertEqual(group.metadata.get("label"), "女主")

    @patch("apps.ai.services._store_generated_media")
    @patch("apps.ai.services.get_image_provider")
    def test_product_base_asset_uses_cover_image_as_reference(self, get_provider, store_media):
        """商品三视图:商品有主图时走 image_edit 以主图为参考(锁包装一致),而非纯文生图。"""
        from apps.assets.models import AssetFile

        ModelConfig.objects.create(
            provider=self.provider, name="img-model", display_name="Img",
            capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
        )
        # 商品主图资产(带可访问 preview_url)
        cover = Asset.objects.create(
            team=self.team, created_by=self.user, name="主图",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE,
        )
        AssetFile.objects.create(asset=cover, object_key="k.png", bucket="b", content_type="image/png", preview_url="http://x/cover.png", is_primary=True)
        self.product.cover_asset = cover
        self.product.save(update_fields=["cover_asset"])

        out = Asset.objects.create(
            team=self.team, created_by=self.user, name="商品三视图",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PRODUCT_IMAGE,
        )
        store_media.return_value = out
        provider = get_provider.return_value
        provider.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]}
        provider.extract_first_media_url.return_value = "http://x/tri.png"
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")

        response = self.client.post(
            f"/api/projects/{project.id}/generate-base-asset/",
            {"kind": "product", "prompt": "商品三视图"},
            format="json",
        )
        self.assertEqual(response.status_code, 202)  # 异步:秒回任务,EAGER 下 worker 就地出图
        # 走 image_edit 且把商品主图 URL 作为参考图传入,未走纯文生图
        provider.image_edit.assert_called_once()
        self.assertEqual(provider.image_edit.call_args.kwargs["images"], ["http://x/cover.png"])
        provider.image_generation.assert_not_called()

    @patch("apps.ai.services._store_generated_media")
    @patch("apps.ai.services.get_image_provider")
    def test_product_base_asset_falls_back_without_cover(self, get_provider, store_media):
        """商品无主图时回落纯文生图(image_generation),不报错。"""
        ModelConfig.objects.create(
            provider=self.provider, name="img-model", display_name="Img",
            capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
        )
        out = Asset.objects.create(
            team=self.team, created_by=self.user, name="商品三视图",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PRODUCT_IMAGE,
        )
        store_media.return_value = out
        provider = get_provider.return_value
        provider.image_generation.return_value = {"data": [{"url": "http://x/tri.png"}]}
        provider.extract_first_media_url.return_value = "http://x/tri.png"
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")

        response = self.client.post(
            f"/api/projects/{project.id}/generate-base-asset/",
            {"kind": "product", "prompt": "商品三视图"},
            format="json",
        )
        self.assertEqual(response.status_code, 202)  # 异步:秒回任务,EAGER 下 worker 就地出图
        provider.image_generation.assert_called_once()
        provider.image_edit.assert_not_called()

    @patch("apps.ai.services._store_generated_media")
    @patch("apps.ai.services.get_image_provider")
    def test_triview_async_binds_group_to_portrait(self, get_provider, store_media):
        """三视图异步:提交秒回 202;EAGER 下 worker 就地以立绘为参考 image_edit,生成的组 triview_of 绑定该立绘。"""
        from apps.assets.models import AssetFile
        from apps.projects.models import BaseAssetGroup

        ModelConfig.objects.create(
            provider=self.provider, name="img-model", display_name="Img",
            capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
        )
        portrait = Asset.objects.create(
            team=self.team, created_by=self.user, name="立绘",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON,
        )
        AssetFile.objects.create(asset=portrait, object_key="p.png", bucket="b", content_type="image/png", preview_url="http://x/portrait.png", is_primary=True)
        tri = Asset.objects.create(
            team=self.team, created_by=self.user, name="三视图",
            asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON,
        )
        store_media.return_value = tri
        provider = get_provider.return_value
        provider.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]}
        provider.extract_first_media_url.return_value = "http://x/tri.png"
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")

        response = self.client.post(
            f"/api/projects/{project.id}/generate-triview/",
            {"portrait_asset_id": str(portrait.id)},
            format="json",
        )
        self.assertEqual(response.status_code, 202)
        provider.image_edit.assert_called_once()
        self.assertEqual(provider.image_edit.call_args.kwargs["images"], ["http://x/portrait.png"])
        group = BaseAssetGroup.objects.get(project=project, kind=BaseAssetGroup.Kind.PERSON)
        self.assertEqual(group.metadata.get("triview_of"), str(portrait.id))
        self.assertEqual(group.adopted_asset_id, tri.id)

    def test_extract_cast_and_scenes_parsing_is_robust(self):
        """纯函数:_coerce_tag_entries 去重/容错;无可用模型时 extract 返回空且不抛。"""
        from apps.ai.services import _coerce_tag_entries, extract_cast_and_scenes
        names, prompts = _coerce_tag_entries(
            [{"name": "女主", "prompt": "p1"}, {"name": "女主", "prompt": "dup"}, {"name": "无"}, "同事"]
        )
        self.assertEqual(names, ["女主", "同事"])  # 去重 + 过滤占位「无」+ 容忍纯字符串项
        self.assertEqual(prompts["女主"], "p1")
        ModelConfig.objects.filter(capability=ModelConfig.Capability.TEXT).update(status=ModelConfig.Status.DISABLED)
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
        self.assertEqual(
            extract_cast_and_scenes(project=project, user=self.user, content="脚本"),
            {"cast": [], "scenes": [], "cast_prompts": {}, "scene_prompts": {}},
        )

    def test_adopt_video_version_remaps_timeline_draft(self):
        """切换采用版本后,时间线草稿里引用本场旧版本资产的片段必须跟随换成新资产(剪辑台/导出都读草稿)。"""
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
        make_asset = lambda name: Asset.objects.create(
            team=self.team, created_by=self.user, name=name,
            asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.VIDEO_CLIP,
        )
        old_asset, new_asset = make_asset("v1"), make_asset("v2")
        segment = VideoSegment.objects.create(project=project, sort_order=0)
        old_ver = VideoSegmentVersion.objects.create(video_segment=segment, asset=old_asset, is_adopted=True)
        new_ver = VideoSegmentVersion.objects.create(video_segment=segment, asset=new_asset)
        segment.adopted_version = old_ver
        segment.save(update_fields=["adopted_version"])
        timeline = Timeline.objects.create(project=project)
        clip = TimelineClip.objects.create(
            timeline=timeline, asset=old_asset, sort_order=0,
            duration_ms=15000, trim_start_ms=1000, trim_end_ms=9000,
        )

        response = self.client.post(
            f"/api/projects/{project.id}/adopt-video-version/",
            {"video_segment_id": str(segment.id), "version_id": str(new_ver.id)},
            format="json",
        )

        self.assertEqual(response.status_code, 200)
        clip.refresh_from_db()
        self.assertEqual(clip.asset_id, new_asset.id)
        # 旧素材上的裁剪点对新素材无意义,必须复位
        self.assertEqual(clip.trim_start_ms, 0)
        self.assertIsNone(clip.trim_end_ms)

    def _make_audio_model(self):
        return ModelConfig.objects.create(
            provider=self.provider, name="doubao-voice-bigtts", display_name="豆包语音合成",
            capability="audio", endpoint="api/v1/tts", unit_price="2.0000",
        )

    @patch("apps.ai.services.TosStorage")
    @patch("apps.ai.services.VolcanoTtsProvider")
    def test_generate_voiceover_creates_assets_and_charges_once(self, provider_cls, storage_cls):
        """每镜旁白 TTS:产出音频资产、timeline.metadata 落映射、一次调用只计一次费。"""
        self._make_audio_model()
        provider = provider_cls.return_value
        provider.configured = True
        provider.synthesize.return_value = (b"fake-mp3-bytes", 4200)
        stored = storage_cls.return_value.upload_fileobj.return_value
        stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "k.mp3", "b", "audio/mpeg", 14

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO")
        response = self.client.post(
            f"/api/projects/{project.id}/generate-voiceover/",
            {"items": [{"index": 0, "text": "第一镜旁白"}, {"index": 1, "text": "第二镜旁白"}]},
            format="json",
        )

        self.assertEqual(response.status_code, 201)
        voiceover = response.json()["voiceover"]
        self.assertTrue(voiceover["enabled"])
        self.assertEqual(len(voiceover["items"]), 2)
        self.assertEqual(provider.synthesize.call_count, 2)
        from apps.assets.models import Asset as AssetModel
        self.assertEqual(AssetModel.objects.filter(team=self.team, asset_type="audio").count(), 2)
        self.assertEqual(
            CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 1
        )
        project.refresh_from_db()
        self.assertEqual(len(project.timeline.metadata["voiceover"]["items"]), 2)

    @patch("apps.ai.services.VolcanoTtsProvider")
    def test_generate_voiceover_unconfigured_returns_400_and_no_charge(self, provider_cls):
        """语音凭证未配置:400 + 人话提示,不建任务不扣费。"""
        self._make_audio_model()
        provider_cls.return_value.configured = False
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO2")
        response = self.client.post(
            f"/api/projects/{project.id}/generate-voiceover/",
            {"items": [{"index": 0, "text": "旁白"}]},
            format="json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertIn("语音合成未配置", response.json()["detail"])
        self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 0)

    def test_export_command_mixes_voiceover_above_bgm(self):
        """导出命令:人声按片段起点 adelay 延迟、先与原声 amix、再与 BGM amix。"""
        from apps.projects.services.export import _build_export_command

        specs = [{"ts": 0.0, "te": 15.0, "dur": 15.0}, {"ts": 0.0, "te": 15.0, "dur": 15.0}]
        cmd = _build_export_command(
            n=2, specs=specs, starts=[0.0, 15.0], total=30.0, transition="none",
            sub_overlays=[], bgm_name="bgm.mp3", bgm_volume=0.6, has_audio=[False, False],
            voice_overlays=[("vo0.mp3", 0.0, 15.0), ("vo1.mp3", 15.0, 15.0)],
        )
        self.assertIn("vo0.mp3", cmd)
        self.assertIn("vo1.mp3", cmd)
        graph = cmd[cmd.index("-filter_complex") + 1]
        self.assertIn("adelay=15000|15000", graph)
        self.assertIn("[avoice][vo0][vo1]amix=inputs=3", graph)
        self.assertIn("[anarr][abgm]amix=inputs=2", graph)

    def test_save_timeline_updates_voiceover_offsets(self):
        """拖动字幕块后保存:按 asset 回写句内起点 offset_ms,未拖动的句不受影响。"""
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VOFF")
        Timeline.objects.create(project=project, metadata={"voiceover": {"enabled": True, "items": [
            {"index": 0, "cue": 0, "asset": "a1", "text": "x", "duration_ms": 2000, "offset_ms": 0},
            {"index": 0, "cue": 1, "asset": "a2", "text": "y", "duration_ms": 2000, "offset_ms": 2000},
        ]}})
        response = self.client.post(
            f"/api/projects/{project.id}/save-timeline/",
            {"voiceover": {"items": [{"asset": "a2", "offset_ms": 9000}]}},
            format="json",
        )
        self.assertEqual(response.status_code, 200)
        project.timeline.refresh_from_db()
        items = project.timeline.metadata["voiceover"]["items"]
        self.assertEqual(items[0]["offset_ms"], 0)
        self.assertEqual(items[1]["offset_ms"], 9000)

    def test_subtitle_cues_use_end_ms_and_follow_voiceover(self):
        """字幕烧入时序:cue 自带 end_ms 时按它收尾(不再拖到片段尾)——这是配音对齐的关键;
        无保存字幕的回退路径,有配音时逐句窗口也要压进语音真实时长。"""
        from apps.projects.services.export import _subtitle_cues

        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="SUB")
        timeline = Timeline.objects.create(
            project=project,
            metadata={"voiceover": {"enabled": True, "items": [{"index": 0, "asset": "x", "text": "t", "duration_ms": 4000}]}},
        )
        SubtitleTrack.objects.create(timeline=timeline, enabled=True, content=[
            {"start_ms": 0, "end_ms": 2000, "text": "第一句"},
            {"start_ms": 2000, "end_ms": 4000, "text": "第二句"},
        ])
        specs = [{"ts": 0.0, "te": 15.0, "dur": 15.0}]
        cues = _subtitle_cues(timeline, project, specs, [0.0], 15.0)
        self.assertEqual(len(cues), 2)
        self.assertEqual((cues[0][0], cues[0][1]), (0.0, 2.0))
        self.assertEqual(cues[1][1], 4.0)  # 最后一句在语音念完处收尾,而非拖到 15s

    def test_save_timeline_voiceover_toggle_and_clear(self):
        """保存草稿可关/开配音、可移除映射(不动已生成的音频资产)。"""
        project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO3")
        Timeline.objects.create(
            project=project,
            metadata={"voiceover": {"enabled": True, "voice_type": "v", "items": [{"index": 0, "asset": "a", "text": "t"}]}},
        )
        response = self.client.post(
            f"/api/projects/{project.id}/save-timeline/", {"voiceover": {"enabled": False}}, format="json"
        )
        self.assertEqual(response.status_code, 200)
        project.timeline.refresh_from_db()
        self.assertFalse(project.timeline.metadata["voiceover"]["enabled"])

        response = self.client.post(
            f"/api/projects/{project.id}/save-timeline/", {"voiceover": {"clear": True}}, format="json"
        )
        self.assertEqual(response.status_code, 200)
        project.timeline.refresh_from_db()
        self.assertNotIn("voiceover", project.timeline.metadata)


class WorkerGateTests(TestCase):
    """无 Celery worker 时,图片/视频生成入口必须 503 拒绝(防结果无人取回=数据丢失)。"""

    def setUp(self):
        self.user = User.objects.create_user(username="gate", password="pass")
        self.team = Team.objects.create(name="Gate Team", owner=self.user)
        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
        CreditAccount.objects.create(team=self.team, balance="100.0000")
        self.product = Product.objects.create(team=self.team, created_by=self.user, title="Gate Product")
        self.project = Project.objects.create(team=self.team, created_by=self.user, name="Gate", product=self.product)
        self.client = APIClient()
        self.client.force_authenticate(self.user)

    def _blocked(self):
        return patch("apps.common.celery_health.celery_worker_available", return_value=False)

    def test_generation_endpoints_blocked_without_worker(self):
        cases = [
            (f"/api/projects/{self.project.id}/generate-base-asset/", {"kind": "person"}),
            (f"/api/projects/{self.project.id}/generate-triview/", {"portrait_asset_id": "x"}),
            (f"/api/projects/{self.project.id}/generate-storyboard/", {}),
            (f"/api/projects/{self.project.id}/submit-video-segment/", {"video_segment_id": "x"}),
            ("/api/ai/generate-image/", {"prompt": "测试"}),
        ]
        with self._blocked():
            for url, payload in cases:
                response = self.client.post(url, payload, format="json")
                self.assertEqual(response.status_code, 503, msg=f"{url} 应被 worker 闸拦截")
                self.assertIn("worker", response.json()["detail"].lower() if isinstance(response.json().get("detail"), str) else "")

    def test_eager_mode_bypasses_gate(self):
        from apps.common.celery_health import celery_worker_available

        # 测试配置 CELERY_TASK_ALWAYS_EAGER=True:任务就地执行,无需 worker,闸放行
        self.assertTrue(celery_worker_available())

    def test_ping_failure_means_unavailable(self):
        from apps.common import celery_health

        celery_health._cache["expires"] = 0.0
        with self.settings(CELERY_TASK_ALWAYS_EAGER=False):
            with patch.object(celery_health, "_ping_workers", side_effect=ConnectionError("broker down")):
                self.assertFalse(celery_health.celery_worker_available())
        celery_health._cache["expires"] = 0.0


class ExportClipsTests(TestCase):
    """导出全部:把项目已采用视频片段打成一个 zip 下载。"""

    def setUp(self):
        self.user = User.objects.create_user(username="exp", password="p")
        self.team = Team.objects.create(name="ExpT", owner=self.user)
        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
        self.product = Product.objects.create(team=self.team, created_by=self.user, title="P")
        self.project = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user)
        self.client = APIClient()
        self.client.force_authenticate(self.user)

    def _add_clip(self, order):
        from apps.assets.models import AssetFile

        asset = Asset.objects.create(team=self.team, name=f"clip{order}", asset_type="video", source="ai_generated", category="video_clip")
        AssetFile.objects.create(asset=asset, object_key=f"k{order}.mp4", bucket="b", content_type="video/mp4", is_primary=True)
        seg = VideoSegment.objects.create(project=self.project, sort_order=order)
        ver = VideoSegmentVersion.objects.create(video_segment=seg, asset=asset, is_adopted=True)
        seg.adopted_version = ver
        seg.save(update_fields=["adopted_version"])

    def test_export_zip_packs_all_clips(self):
        self._add_clip(0)
        self._add_clip(1)
        with patch("apps.projects.views.TosStorage") as Tos, patch("requests.get") as rget:
            Tos.return_value.presigned_get_url.return_value = "http://x/c.mp4"
            rget.return_value.content = b"MP4DATA"
            res = self.client.get(f"/api/projects/{self.project.id}/export-clips/")
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res["Content-Type"], "application/zip")
        import io
        import zipfile

        zf = zipfile.ZipFile(io.BytesIO(res.content))
        self.assertEqual(len(zf.namelist()), 2)

    def test_export_empty_returns_400(self):
        res = self.client.get(f"/api/projects/{self.project.id}/export-clips/")
        self.assertEqual(res.status_code, 400)


class BaseAssetAdoptDeleteTests(TestCase):
    """显性采用/未采用 + 删除角色组。"""

    def setUp(self):
        from apps.projects.models import BaseAssetGroup

        self.user = User.objects.create_user(username="bad", password="x")
        self.team = Team.objects.create(name="BAD", owner=self.user)
        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
        self.product = Product.objects.create(team=self.team, created_by=self.user, title="P")
        self.proj = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user)
        self.group = BaseAssetGroup.objects.create(project=self.proj, kind=BaseAssetGroup.Kind.PERSON, metadata={"label": "小明"})
        self.client = APIClient()
        self.client.force_authenticate(self.user)

    def test_set_adopt_state(self):
        res = self.client.post(f"/api/projects/{self.proj.id}/set-base-asset-adopt/", {"group_id": str(self.group.id), "state": "unadopted"}, format="json")
        self.assertEqual(res.status_code, 200)
        self.group.refresh_from_db()
        self.assertEqual(self.group.metadata.get("adopt"), "unadopted")

    def test_set_adopt_invalid_state(self):
        res = self.client.post(f"/api/projects/{self.proj.id}/set-base-asset-adopt/", {"group_id": str(self.group.id), "state": "x"}, format="json")
        self.assertEqual(res.status_code, 400)

    def test_delete_base_asset(self):
        from apps.projects.models import BaseAssetGroup

        res = self.client.post(f"/api/projects/{self.proj.id}/delete-base-asset/", {"group_id": str(self.group.id)}, format="json")
        self.assertEqual(res.status_code, 200)
        self.assertFalse(BaseAssetGroup.objects.filter(id=self.group.id).exists())