seaislee1209andClaude Opus 4.8
71af0ae166
问题:进资产阶段点"提取人物/场景"间歇报「提取结果解析失败(模型未返回有效 JSON)」。根因——提取走非流式只读 content,而默认文本模型是会思考的推理模型 (豆包2.0Pro think:True),思考期常把内容留在 reasoning_content、content 返空 → 正则抽不到 JSON。脚本生成不犯此病是因其走流式、早已分离思考/正文。 改动: - 锁模型:提取固定用 doubao-seed-2-0-pro-260215(_resolve_extract_model_config), 取不到再回落默认,不再受各环境 DB 创建序漂移影响路由。 - 根治:提取复用与脚本生成同一条已验证稳定的流式通道(_collect_extract_text)—— 丢弃 reasoning 事件、只收正文 delta;正文为空时兜底回退 reasoning 里的 JSON。 response_payload 改存精简流式存档(streamed/model/content/had_reasoning)。 - 测试:提取步原零测试,新增 2 个——流式正文JSON正确落库+计费一次、content空→ 回退reasoning。其余链路(脚本生成/单镜重跑)未动。 验收:本机 Python3.9 跑不了 Django5 全套,核心逻辑(事件过滤/回退/JSON抽取)已 独立验证通过;py_compile 绿;完整套件由 CI 跑。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
638 lines
34 KiB
Python
638 lines
34 KiB
Python
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 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._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())
|