"""为模型路由浏览器验收创建可删除的 QA 任务。 只允许在 DEBUG 环境且显式设置 MODEL_ROUTING_BROWSER_QA=1 时运行。脚本不会调用真实 Provider 或对象存储，也不会使用现有团队积分；它创建不可登录的临时用户和团队，真实走 AITask、预留积分、统一路由执行器、尝试日志及最终结算/释放流程。 用法（从 core/backend 目录运行）： MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-reference MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-tryon MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-platform MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-base MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-model-triview MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-project-triview MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-storyboard MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-entity MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-script MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-voice MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-video MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py setup-free-video MODEL_ROUTING_BROWSER_QA=1 python ../../core/qa/model-routing-browser-qa.py cleanup """ from __future__ import annotations import argparse from copy import deepcopy from decimal import Decimal from io import BytesIO import json import os from pathlib import Path import sys from unittest.mock import Mock, patch BACKEND_DIR = Path(__file__).resolve().parents[1] / "backend" sys.path.insert(0, str(BACKEND_DIR)) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "airshelf.settings.development") import django # noqa: E402 django.setup() import requests # noqa: E402 from django.conf import settings # noqa: E402 from django.test import override_settings # noqa: E402 from apps.accounts.models import Team, TeamMember, User # noqa: E402 from apps.ai.models import AITask, ModelConfig, ModelProvider # noqa: E402 from apps.ai.script_agent import regenerate_segment_via_agent, stream_script_agent # noqa: E402 from apps.ai.services import ( # noqa: E402 _storyboard_shot_worker, enqueue_standalone_images, generate_base_asset, generate_model_triview, generate_person_triview, run_extract_entities_task, run_base_asset_task, run_model_triview_task, run_standalone_image_task, run_triview_task, poll_storyboard, poll_video_segment, submit_video_segment, submit_extract_entities, synthesize_project_voiceover, ) from apps.assets.models import Asset, AssetFile, Model # noqa: E402 from apps.assets.storage import StoredObject # noqa: E402 from apps.billing.models import CreditAccount, CreditLedger # noqa: E402 from apps.products.models import Product # noqa: E402 from apps.projects.models import ( # noqa: E402 BaseAssetGroup, Project, ScriptSegment, ScriptVersion, StoryboardShot, VideoSegment, ) QA_USERNAME = "__routing_browser_qa__" QA_TEAM_NAME = "__Routing Browser QA__" QA_PROVIDER_NAMES = ( "qa-routing-primary", "qa-routing-fallback-1", "qa-routing-fallback-2", ) def _assert_safe() -> None: if not settings.DEBUG or os.environ.get("MODEL_ROUTING_BROWSER_QA") != "1": raise SystemExit( "拒绝运行：仅允许在 DEBUG 环境，并需显式设置 MODEL_ROUTING_BROWSER_QA=1。" ) def cleanup() -> dict[str, int]: tasks = AITask.objects.filter(team__name=QA_TEAM_NAME).count() teams = Team.objects.filter(name=QA_TEAM_NAME).count() users = User.objects.filter(username=QA_USERNAME).count() providers = ModelProvider.objects.filter(name__in=QA_PROVIDER_NAMES).count() # 项目通过 PROTECT 引用商品；先删精确 QA 团队下的项目，再让团队级联清理商品与其余数据。 Project.objects.filter(team__name=QA_TEAM_NAME).delete() Team.objects.filter(name=QA_TEAM_NAME).delete() ModelProvider.objects.filter(name__in=QA_PROVIDER_NAMES).delete() User.objects.filter(username=QA_USERNAME).delete() return {"tasks": tasks, "teams": teams, "users": users, "providers": providers} def _image_metadata(*, priority: int) -> dict: return { "routing": { "fallback_on_failure": True, "fallback_candidate": True, }, "capabilities": { "operations": ["image_generate", "image_edit"], "features": [], "reference_modes": ["none", "single", "multiple"], "max_reference_images": 9, "aspect_ratios": ["1:1", "3:4", "4:5", "9:16", "16:9"], }, "pricing": {"base_cost_yuan": "0.10"}, "qa": {"browser_routing_priority": priority}, } def _create_model(provider_name: str, priority: int, model_name: str) -> ModelConfig: provider = ModelProvider.objects.create( name=provider_name, display_name=f"QA Provider {priority}", status=ModelProvider.Status.ACTIVE, base_url="http://127.0.0.1.invalid/v1", api_key="", metadata={"routing": {"fallback_priority": priority}, "qa": {"browser_routing": True}}, ) return ModelConfig.objects.create( provider=provider, name=model_name, display_name=f"QA Image {priority}", capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price=Decimal("20"), status=ModelConfig.Status.ACTIVE, metadata=_image_metadata(priority=priority), ) def _create_text_model(provider_name: str, priority: int, model_name: str) -> ModelConfig: provider = ModelProvider.objects.create( name=provider_name, display_name=f"QA Provider {priority}", status=ModelProvider.Status.ACTIVE, base_url="http://127.0.0.1.invalid/v1", api_key="", metadata={"routing": {"fallback_priority": priority}, "qa": {"browser_routing": True}}, ) return ModelConfig.objects.create( provider=provider, name=model_name, display_name=f"QA Script {priority}", capability=ModelConfig.Capability.TEXT, endpoint="chat/completions", unit_price=Decimal("10"), status=ModelConfig.Status.ACTIVE, metadata={ "routing": {"fallback_on_failure": True, "fallback_candidate": True}, "capabilities": { "operations": ["chat"], "features": ["streaming", "structured_output"], }, "pricing": {"base_cost_yuan": "0.10"}, "qa": {"browser_routing_priority": priority}, }, ) def _create_audio_model(provider_name: str, priority: int, model_name: str) -> ModelConfig: provider = ModelProvider.objects.create( name=provider_name, display_name=f"QA Provider {priority}", status=ModelProvider.Status.ACTIVE, base_url="http://127.0.0.1.invalid/v1", api_key="", metadata={"routing": {"fallback_priority": priority}, "qa": {"browser_routing": True}}, ) return ModelConfig.objects.create( provider=provider, name=model_name, display_name=f"QA Voice {priority}", capability=ModelConfig.Capability.AUDIO, endpoint="audio/speech", unit_price=Decimal("10"), status=ModelConfig.Status.ACTIVE, metadata={ "routing": {"fallback_on_failure": True, "fallback_candidate": True}, "capabilities": { "operations": ["tts"], "features": [], "languages": ["zh-CN"], "voice_map": {"BV700_streaming": "alloy"}, "max_chars": 10000, "speed_range": [0.5, 2.0], "output_formats": ["mp3"], }, "pricing": { "chars_per_unit": 500, "points_per_unit": 10, "min_units": 1, "base_cost_yuan_per_unit": "0.10", }, "qa": {"browser_routing_priority": priority}, }, ) def _create_video_model(provider_name: str, priority: int, model_name: str) -> ModelConfig: provider = ModelProvider.objects.create( name=provider_name, display_name=f"QA Provider {priority}", status=ModelProvider.Status.ACTIVE, base_url="http://127.0.0.1.invalid/v1", api_key="", metadata={"routing": {"fallback_priority": priority}, "qa": {"browser_routing": True}}, ) return ModelConfig.objects.create( provider=provider, name=model_name, display_name=f"QA Video {priority}", capability=ModelConfig.Capability.VIDEO, endpoint="videos", unit_price=Decimal("1"), status=ModelConfig.Status.ACTIVE, metadata={ "routing": {"fallback_on_failure": True, "fallback_candidate": True}, "capabilities": { "operations": ["video_generate"], "features": ["generate_audio"], "max_reference_images": 9, "max_reference_videos": 3, "max_reference_audios": 3, "aspect_ratios": ["9:16", "16:9"], "resolutions": ["480p", "720p"], "durations": [4, 15], }, "pricing": { "unit": "cny_per_million_tokens", "default": {"no_ref_video": 23, "with_ref_video": 14}, }, "qa": {"browser_routing_priority": priority}, }, ) def _ledger_counts(task: AITask) -> dict[str, int]: return { kind: CreditLedger.objects.filter(task=task, ledger_type=kind).count() for kind in ( CreditLedger.Type.RESERVE, CreditLedger.Type.CHARGE, CreditLedger.Type.RELEASE, ) } def setup(*, with_references: bool = False, tryon: bool = False, platform: bool = False) -> dict: cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-image") fallback_1 = _create_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-image-1") fallback_2 = _create_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-image-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} scenario = {"name": "first_success"} provider_mocks: dict[object, Mock] = {} def provider_for(model: ModelConfig) -> Mock: provider = provider_mocks.get(model.id) if provider is None: provider = Mock() provider.extract_first_media_url.side_effect = lambda response: response["data"][0]["url"] provider_mocks[model.id] = provider current = scenario["name"] route_name = current.removeprefix("reference_").removeprefix("tryon_").removeprefix("platform_") should_fail = route_name == "all_failed" or ( route_name == "fallback_success" and model.id == primary.id ) if model.id not in qa_model_ids: should_fail = True provider.image_generation.side_effect = ( requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None ) provider.image_generation.return_value = { "data": [{"url": "http://qa.invalid/generated.png"}], "usage": {"images": 1}, } provider.image_edit.side_effect = provider.image_generation.side_effect provider.image_edit.return_value = provider.image_generation.return_value return provider policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["image"]["retry_delays"] = [0] policy["jitter_ratio"] = 0 stored = StoredObject( bucket="qa-browser", object_key="qa/generated.png", content_type="image/png", size_bytes=8, ) created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_image_provider", side_effect=provider_for), patch( "apps.ai.services.VolcanoArkProvider.media_to_bytes", return_value=(BytesIO(b"qa-image"), "image/png"), ), patch("apps.ai.services.TosStorage") as storage, ): storage.return_value.upload_fileobj.return_value = stored reference_ids: list[str] = [] tryon_product = None tryon_model = None if with_references or tryon or platform: for index in range(2): reference = Asset.objects.create( team=team, created_by=user, name=f"QA 参考图 {index + 1}", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, ) AssetFile.objects.create( asset=reference, object_key=f"qa/reference-{index + 1}.png", bucket="qa-browser", content_type="image/png", preview_url=f"http://qa.invalid/reference-{index + 1}.png", is_primary=True, ) reference_ids.append(str(reference.id)) if tryon: product_asset = Asset.objects.get(id=reference_ids[0]) product_asset.category = Asset.Category.PRODUCT_IMAGE product_asset.save(update_fields=["category"]) tryon_product = Product.objects.create( team=team, created_by=user, title="QA 模特上身图商品", category="服饰内衣", cover_asset=product_asset, ) tryon_model = Asset.objects.get(id=reference_ids[1]) tryon_model.source = Asset.Source.AI_GENERATED tryon_model.category = Asset.Category.PERSON tryon_model.save(update_fields=["source", "category"]) platform_product = None if platform: product_asset = Asset.objects.get(id=reference_ids[0]) product_asset.category = Asset.Category.PRODUCT_IMAGE product_asset.save(update_fields=["category"]) platform_product = Product.objects.create( team=team, created_by=user, title="QA 平台套图商品", category="服饰内衣", cover_asset=product_asset, ) prefix = "platform_" if platform else ("tryon_" if tryon else ("reference_" if with_references else "")) label = "平台套图" if platform else ("模特上身图" if tryon else ("多参考图" if with_references else "无参考图")) for short_name, description in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): name = f"{prefix}{short_name}" prompt = f"QA 浏览器验收（{label}）：{description}" scenario["name"] = name provider_mocks.clear() task = enqueue_standalone_images( team=team, user=user, prompt=prompt, mode="cover" if platform else ("model" if tryon else "image"), count=1, ratio="1:1", image_model=f"{primary.provider.name}:{primary.name}", reference_image_ids=reference_ids if with_references and not tryon else [], product_id=( str(platform_product.id) if platform_product is not None else (str(tryon_product.id) if tryon_product is not None else None) ), model_id=str(tryon_model.id) if tryon_model is not None else None, platform_id="taobao" if platform else None, dispatch=False, )[0] run_standalone_image_task(task_id=str(task.id)) task.refresh_from_db() created.append( { "scenario": name, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "reference_images": list( task.model_attempts.values_list("request_summary__reference_images", flat=True) ), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_base() -> dict: """创建商品、人物、场景三类基础资产的受控成功/Fallback/全失败任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-image") fallback_1 = _create_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-image-1") fallback_2 = _create_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-image-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} cover = Asset.objects.create( team=team, created_by=user, name="QA 商品主图", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE, ) AssetFile.objects.create( asset=cover, object_key="qa/base-product.png", bucket="qa-browser", content_type="image/png", preview_url="http://qa.invalid/base-product.png", is_primary=True, ) product = Product.objects.create( team=team, created_by=user, title="QA 基础资产商品", category="服饰内衣", cover_asset=cover, ) project = Project.objects.create( team=team, created_by=user, product=product, name="QA 基础资产项目", ) scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model: ModelConfig) -> Mock: provider = provider_mocks.get(model.id) if provider is None: provider = Mock() provider.extract_first_media_url.side_effect = lambda response: response["data"][0]["url"] provider_mocks[model.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or ( current.endswith("fallback_success") and model.id == primary.id ) if model.id not in qa_model_ids: should_fail = True failure = requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None response = {"data": [{"url": "http://qa.invalid/generated.png"}], "usage": {"images": 1}} provider.image_generation.side_effect = failure provider.image_generation.return_value = response provider.image_edit.side_effect = failure provider.image_edit.return_value = response return provider def store_media(**kwargs): return Asset.objects.create( team=kwargs["team"], created_by=kwargs["user"], name=kwargs["name"], asset_type=kwargs["asset_type"], source=Asset.Source.AI_GENERATED, category=kwargs["category"], origin_task=kwargs["task"], ) policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["image"]["retry_delays"] = [0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_image_provider", side_effect=provider_for), patch("apps.ai.services._store_generated_media", side_effect=store_media), patch("apps.ai.tasks.generate_base_asset_task.delay"), patch("apps.assets.review.submit_asset_for_review"), ): for kind in ( BaseAssetGroup.Kind.PRODUCT, BaseAssetGroup.Kind.PERSON, BaseAssetGroup.Kind.SCENE, ): for short_name, description in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): name = f"base_{kind}_{short_name}" scenario["name"] = name provider_mocks.clear() task = generate_base_asset( project=project, user=user, kind=kind, prompt=f"QA 浏览器验收（{kind} 基础资产）：{description}", label=f"QA-{kind}-{short_name}", ) run_base_asset_task(task_id=str(task.id)) task.refresh_from_db() created.append( { "scenario": name, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "reference_images": list( task.model_attempts.values_list( "request_summary__reference_images", flat=True ) ), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_model_triview(*, project_scope: bool = False) -> dict: """创建团队模特或项目角色三视图的受控成功/Fallback/全失败任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-image") fallback_1 = _create_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-image-1") fallback_2 = _create_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-image-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} portrait = Asset.objects.create( team=team, created_by=user, name="QA 模特形象", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.MODEL_PORTRAIT, in_library=False, ) AssetFile.objects.create( asset=portrait, object_key="qa/model-portrait.png", bucket="qa-browser", content_type="image/png", preview_url="http://qa.invalid/model-portrait.png", is_primary=True, ) model = None project = None if project_scope: product = Product.objects.create( team=team, created_by=user, title="QA 项目角色三视图商品", ) project = Project.objects.create( team=team, created_by=user, product=product, name="QA 项目角色三视图", ) else: model = Model.objects.create( team=team, created_by=user, name="QA 团队模特", portrait_asset=portrait, ) scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.get(model_config.id) if provider is None: provider = Mock() provider.extract_first_media_url.side_effect = lambda response: response["data"][0]["url"] provider_mocks[model_config.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or ( current.endswith("fallback_success") and model_config.id == primary.id ) if model_config.id not in qa_model_ids: should_fail = True provider.image_edit.side_effect = ( requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None ) provider.image_edit.return_value = { "data": [{"url": "http://qa.invalid/model-triview.png"}], "usage": {"images": 1}, } return provider policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["image"]["retry_delays"] = [0] policy["jitter_ratio"] = 0 stored = StoredObject( bucket="qa-browser", object_key="qa/model-triview.png", content_type="image/png", size_bytes=8, ) def store_media(**kwargs): return Asset.objects.create( team=kwargs["team"], created_by=kwargs["user"], name=kwargs["name"], asset_type=kwargs["asset_type"], source=Asset.Source.AI_GENERATED, category=kwargs["category"], origin_task=kwargs["task"], ) created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_image_provider", side_effect=provider_for), patch("apps.ai.tasks.generate_model_triview_task.delay"), patch("apps.ai.tasks.generate_triview_task.delay"), patch("apps.ai.services._store_generated_media", side_effect=store_media), patch( "apps.ai.services.VolcanoArkProvider.media_to_bytes", return_value=(BytesIO(b"qa-image"), "image/png"), ), patch("apps.ai.services.TosStorage") as storage, patch("apps.assets.review.submit_asset_for_review"), ): storage.return_value.upload_fileobj.return_value = stored for short_name, description in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): prefix = "project_triview" if project_scope else "model_triview" name = f"{prefix}_{short_name}" scenario["name"] = name provider_mocks.clear() if project_scope: task = generate_person_triview( project=project, user=user, portrait_asset=portrait, ) run_triview_task(task_id=str(task.id)) else: task, task_created = generate_model_triview(model=model, user=user) if not task_created: raise RuntimeError("QA 模特三视图任务被错误复用") run_model_triview_task(task_id=str(task.id)) task.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "reference_images": list( task.model_attempts.values_list( "request_summary__reference_images", flat=True ) ), "aspect_ratios": list( task.model_attempts.values_list( "request_summary__aspect_ratio", flat=True ) ), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_storyboard() -> dict: """创建单个故事板分镜的受控成功/Fallback/全失败任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-image") fallback_1 = _create_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-image-1") fallback_2 = _create_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-image-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} product = Product.objects.create( team=team, created_by=user, title="QA 故事板商品", ) project = Project.objects.create( team=team, created_by=user, product=product, name="QA 故事板项目", ) script = ScriptVersion.objects.create( project=project, title="QA 故事板脚本", content="结构化脚本", is_adopted=True, ) segment = ScriptSegment.objects.create( script_version=script, sort_order=0, visual_prompt="QA 人物展示商品", ) shot = StoryboardShot.objects.create( project=project, script_segment=segment, sort_order=0, status=StoryboardShot.Status.QUEUED, ) refs = [ {"url": "http://qa.invalid/storyboard-person.png", "label": "女主", "type": "person"}, {"url": "http://qa.invalid/storyboard-product.png", "label": "商品", "type": "product"}, ] scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.get(model_config.id) if provider is None: provider = Mock() provider.extract_first_media_url.side_effect = lambda response: response["data"][0]["url"] provider_mocks[model_config.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or ( current.endswith("fallback_success") and model_config.id == primary.id ) if model_config.id not in qa_model_ids: should_fail = True provider.image_edit.side_effect = ( requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None ) provider.image_edit.return_value = { "data": [{"url": "http://qa.invalid/storyboard.png"}], "usage": {"images": 1}, } return provider def store_media(**kwargs): return Asset.objects.create( team=kwargs["team"], created_by=kwargs["user"], name=kwargs["name"], asset_type=kwargs["asset_type"], source=Asset.Source.AI_GENERATED, category=kwargs["category"], origin_task=kwargs["task"], ) policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["image"]["retry_delays"] = [0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy, STORYBOARD_MAX_PARALLEL=1), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_image_provider", side_effect=provider_for), patch("apps.ai.services._storyboard_reference_images", return_value=refs), patch( "apps.ai.services.build_storyboard_frame_prompt_refs", return_value="QA 带编号参考图故事板提示词", ), patch("apps.ai.services._store_generated_media", side_effect=store_media), patch("apps.assets.review.submit_asset_for_review"), patch("apps.ai.services.notify_generation_failure"), patch("threading.Thread"), ): for short_name, description in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): name = f"storyboard_{short_name}" scenario["name"] = name provider_mocks.clear() StoryboardShot.objects.filter(id=shot.id).update( status=StoryboardShot.Status.QUEUED, error_message="", ) poll_storyboard(project=project, user=user) task = AITask.objects.filter( project=project, task_type=AITask.Type.STORYBOARD, request_payload__storyboard_shot=str(shot.id), ).latest("created_at") _storyboard_shot_worker(str(task.id), str(shot.id), str(user.id)) task.refresh_from_db() shot.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "reference_images": list( task.model_attempts.values_list( "request_summary__reference_images", flat=True ) ), "aspect_ratios": list( task.model_attempts.values_list( "request_summary__aspect_ratio", flat=True ) ), "shot_versions": shot.versions.count(), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_entity() -> dict: """创建实体提取的受控成功/Fallback/全失败任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_text_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-script") fallback_1 = _create_text_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-script-1") fallback_2 = _create_text_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-script-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} product = Product.objects.create(team=team, created_by=user, title="QA 实体提取商品") project = Project.objects.create( team=team, created_by=user, product=product, name="QA 实体提取项目", ) script = ScriptVersion.objects.create( project=project, title="QA 实体提取脚本", content="结构化脚本", is_adopted=True, ) ScriptSegment.objects.create( script_version=script, sort_order=0, narration="女主在客厅展示商品", visual_prompt="女主站在客厅", ) valid_events = [ { "type": "delta", "text": ( '{"entities":[' '{"id":"c1","type":"character","name":"女主","visual_prompt":"都市女主"},' '{"id":"s1","type":"scene","name":"客厅","visual_prompt":"现代客厅"}' '],"segments":[{"index":0,"entity_refs":["c1","s1"]}]}' ), }, {"type": "done"}, ] scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.get(model_config.id) if provider is None: provider = Mock() provider_mocks[model_config.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or ( current.endswith("fallback_success") and model_config.id == primary.id ) if model_config.id not in qa_model_ids: should_fail = True provider.chat_completion_stream.side_effect = ( requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None ) provider.chat_completion_stream.return_value = valid_events return provider policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["text"]["retry_delays"] = [0, 0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services._resolve_extract_model_config", return_value=primary), patch("apps.ai.services.get_text_provider", side_effect=provider_for), patch("apps.ai.tasks.extract_entities_task.delay"), ): for short_name, description in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): name = f"entity_{short_name}" scenario["name"] = name provider_mocks.clear() task = submit_extract_entities(project=project, user=user) run_extract_entities_task(task_id=str(task.id)) task.refresh_from_db() project.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "streaming": list( task.model_attempts.values_list("request_summary__streaming", flat=True) ), "structured_output": list( task.model_attempts.values_list( "request_summary__structured_output", flat=True ) ), "entities_extracted": bool((project.metadata or {}).get("entities_extracted")), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_script() -> dict: """创建单镜同步优化与整稿流式生成各自的成功/Fallback/全失败任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_text_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-script") fallback_1 = _create_text_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-script-1") fallback_2 = _create_text_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-script-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} product = Product.objects.create(team=team, created_by=user, title="QA 脚本路由商品") project = Project.objects.create( team=team, created_by=user, product=product, name="QA 脚本路由项目", ) base_draft = { "hook": "QA 钩子", "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": [], }, ], } base = ScriptVersion.objects.create( project=project, title="QA 基准稿", content=json.dumps(base_draft, ensure_ascii=False), metadata={"hook": "QA 钩子", "entities": [], "total_duration": 30}, ) ScriptSegment.objects.create( script_version=base, sort_order=0, duration_seconds=15, narration="旧口播0", visual_prompt="旧画面0", ) target = ScriptSegment.objects.create( script_version=base, sort_order=1, duration_seconds=15, narration="旧口播1", visual_prompt="旧画面1", ) revised = deepcopy(base_draft) revised["segments"][1]["narration"] = "QA 优化口播" revised["segments"][1]["visual"] = "QA 优化画面" raw = json.dumps(revised, ensure_ascii=False) events = [ {"type": "reasoning", "text": "QA 分析"}, {"type": "delta", "text": raw}, {"type": "done"}, ] scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.get(model_config.id) if provider is None: provider = Mock() provider_mocks[model_config.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or ( current.endswith("fallback_success") and model_config.id == primary.id ) if model_config.id not in qa_model_ids: should_fail = True error = requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None provider.chat_completion.side_effect = error provider.chat_completion.return_value = {"usage": {"total_tokens": 42}} provider.extract_text.return_value = raw provider.chat_completion_stream.side_effect = error provider.chat_completion_stream.return_value = events return provider policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["text"]["retry_delays"] = [0, 0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_text_provider", side_effect=provider_for), ): for entry, description in (("segment", "单镜同步优化"), ("stream", "整稿流式生成")): for short_name, outcome in ( ("first_success", "主模型首次成功"), ("fallback_success", "主模型重试后 Fallback 成功"), ("all_failed", "全部候选失败并释放预留"), ): name = f"script_{entry}_{short_name}" scenario["name"] = name provider_mocks.clear() before_versions = ScriptVersion.objects.filter(project=project).count() try: if entry == "segment": regenerate_segment_via_agent( project=project, user=user, model_config=primary, segment=target, instruction="更有吸引力", ) else: list( stream_script_agent( project=project, user=user, model_config=primary, mode="auto", user_prompt="生成一版 QA 脚本", aspect_ratio="9:16", total_duration=30, ) ) except Exception: pass task_type = ( AITask.Type.SCRIPT_OPTIMIZATION if entry == "segment" else AITask.Type.SCRIPT_GENERATION ) task = AITask.objects.filter(project=project, task_type=task_type).latest("created_at") task.refresh_from_db() created.append( { "scenario": name, "description": f"{description}：{outcome}", "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "operations": list(task.model_attempts.values_list("operation", flat=True)), "streaming": list( task.model_attempts.values_list("request_summary__streaming", flat=True) ), "structured_output": list( task.model_attempts.values_list( "request_summary__structured_output", flat=True ) ), "new_versions": ( ScriptVersion.objects.filter(project=project).count() - before_versions ), "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_voice() -> dict: """创建当前直连锁定与未来 OpenAI 兼容候选的配音受控任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("1000"), reserved_balance=Decimal("0")) primary = _create_audio_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-voice") fallback_1 = _create_audio_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-voice-1") fallback_2 = _create_audio_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-voice-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} product = Product.objects.create(team=team, created_by=user, title="QA 配音商品") project = Project.objects.create(team=team, created_by=user, product=product, name="QA 配音项目") scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.get(model_config.id) if provider is None: provider = Mock() provider.configured = True provider_mocks[model_config.id] = provider current = scenario["name"] should_fail = current.endswith("all_failed") or current.endswith("locked_failed") or ( current.endswith("fallback_success") and model_config.id == primary.id ) if model_config.id not in qa_model_ids: should_fail = True provider.synthesize.side_effect = ( requests.ConnectionError(f"QA controlled failure: {current}") if should_fail else None ) provider.synthesize.return_value = (b"qa-mp3", 1200) return provider policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["audio"]["retry_delays"] = [0] policy["jitter_ratio"] = 0 stored = Mock(object_key="qa-voice.mp3", bucket="qa", content_type="audio/mpeg", size_bytes=6) created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_audio_provider", side_effect=provider_for), patch("apps.ai.services.TosStorage.upload_fileobj", return_value=stored), ): for name, description, outbound in ( ("voice_locked_first_success", "当前直连首次成功", False), ("voice_locked_failed", "当前直连失败只重试不外切", False), ("voice_fallback_success", "未来打开开关后动态 Fallback 成功", True), ("voice_all_failed", "未来候选全部失败并释放预留", True), ): routing = dict((primary.metadata or {}).get("routing") or {}) routing["fallback_on_failure"] = outbound primary.metadata = {**(primary.metadata or {}), "routing": routing} primary.save(update_fields=["metadata", "updated_at"]) scenario["name"] = name provider_mocks.clear() before_assets = Asset.objects.filter(team=team, asset_type=Asset.Type.AUDIO).count() try: synthesize_project_voiceover( project=project, user=user, items=[{"index": 0, "text": "QA 配音旁白"}], voice_type="BV700_streaming", speed_ratio=1.0, ) except Exception: pass task = AITask.objects.filter(project=project, task_type=AITask.Type.VOICEOVER).latest("created_at") task.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "models": list(task.model_attempts.values_list("model_name", flat=True)), "operations": list(task.model_attempts.values_list("operation", flat=True)), "new_assets": Asset.objects.filter(team=team, asset_type=Asset.Type.AUDIO).count() - before_assets, "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": { "balance": str(account.balance), "reserved_balance": str(account.reserved_balance), }, } def setup_video() -> dict: """创建视频片段提交、Fallback、状态未知与全部失败的受控任务。""" cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("10000"), reserved_balance=Decimal("0")) primary = _create_video_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-video") fallback_1 = _create_video_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-video-1") fallback_2 = _create_video_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-video-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} product = Product.objects.create(team=team, created_by=user, title="QA 视频商品") project = Project.objects.create(team=team, created_by=user, product=product, name="QA 视频片段项目") scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.setdefault(model_config.id, Mock()) current = scenario["name"] provider.create_video_task.side_effect = None provider.create_video_task.return_value = { "id": f"qa-{current}-{model_config.id}", "status": "queued", } if current == "video_state_unknown" and model_config.id == primary.id: provider.create_video_task.side_effect = requests.ReadTimeout("QA response lost") elif current == "video_all_failed" or ( current == "video_fallback_success" and model_config.id == primary.id ) or model_config.id not in qa_model_ids: provider.create_video_task.side_effect = requests.ConnectionError( f"QA controlled failure: {current}" ) provider.poll_video_task.return_value = { "status": "succeeded", "usage": {"total_tokens": 300000}, "content": {"video_url": "https://qa.invalid/video.mp4"}, } provider.extract_first_media_url.return_value = "https://qa.invalid/video.mp4" return provider def store_media(**kwargs): return Asset.objects.create( team=kwargs["team"], created_by=kwargs["user"], name=kwargs["name"], asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.VIDEO_CLIP, origin_task=kwargs["task"], ) policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["video"]["submit_retry_delays"] = [0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.services.get_default_model", return_value=primary), patch("apps.ai.services.get_video_provider", side_effect=provider_for), patch("apps.ai.services._video_reference_images", return_value=[]), patch("apps.ai.services.build_video_segment_prompt", return_value="QA 视频提示词"), patch("apps.ai.services._store_generated_media", side_effect=store_media), patch("apps.ai.services.notify_generation_failure"), ): for index, (name, description) in enumerate( ( ("video_first_success", "视频片段首次提交成功并成片"), ("video_fallback_success", "视频片段原模型重试后动态 Fallback 并成片"), ("video_state_unknown", "视频片段提交状态未知且禁止重提"), ("video_all_failed", "视频片段所有候选明确失败并释放预留"), ) ): scenario["name"] = name provider_mocks.clear() segment = VideoSegment.objects.create( project=project, sort_order=index, target_duration_seconds=15, ) before_assets = Asset.objects.filter(team=team, category=Asset.Category.VIDEO_CLIP).count() try: submit_video_segment(video_segment=segment, user=user, prompt="QA") poll_video_segment(video_segment=segment, user=user) except Exception: pass task = AITask.objects.filter( project=project, task_type=AITask.Type.VIDEO_SEGMENT, request_payload__video_segment_id=str(segment.id), ).latest("created_at") task.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "models": list(task.model_attempts.values_list("model_name", flat=True)), "provider_task_id": task.provider_task_id, "new_assets": Asset.objects.filter( team=team, category=Asset.Category.VIDEO_CLIP ).count() - before_assets, "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": {"balance": str(account.balance), "reserved_balance": str(account.reserved_balance)}, } def setup_free_video() -> dict: """创建自由创作视频多素材提交与动态 Fallback 的受控任务。""" from apps.ai.free_video import finalize_free_video, submit_free_video cleanup() user = User.objects.create(username=QA_USERNAME, status=User.Status.ACTIVE) user.set_unusable_password() user.save(update_fields=["password"]) team = Team.objects.create(name=QA_TEAM_NAME, owner=user) TeamMember.objects.create( team=team, user=user, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE, ) CreditAccount.objects.create(team=team, balance=Decimal("10000"), reserved_balance=Decimal("0")) primary = _create_video_model(QA_PROVIDER_NAMES[0], -300, "qa-primary-free-video") fallback_1 = _create_video_model(QA_PROVIDER_NAMES[1], -200, "qa-fallback-free-video-1") fallback_2 = _create_video_model(QA_PROVIDER_NAMES[2], -100, "qa-fallback-free-video-2") qa_model_ids = {primary.id, fallback_1.id, fallback_2.id} scenario = {"name": ""} provider_mocks: dict[object, Mock] = {} def provider_for(model_config: ModelConfig) -> Mock: provider = provider_mocks.setdefault(model_config.id, Mock()) current = scenario["name"] provider.create_video_task.side_effect = None provider.create_video_task.return_value = { "id": f"qa-{current}-{model_config.id}", "status": "queued", } if current == "free_video_state_unknown" and model_config.id == primary.id: provider.create_video_task.side_effect = requests.ReadTimeout("QA response lost") elif current == "free_video_all_failed" or ( current == "free_video_fallback_success" and model_config.id == primary.id ) or model_config.id not in qa_model_ids: provider.create_video_task.side_effect = requests.ConnectionError( f"QA controlled failure: {current}" ) provider.poll_video_task.return_value = { "status": "succeeded", "usage": {"total_tokens": 30000}, "content": {"video_url": "https://qa.invalid/free-video.mp4"}, } provider.extract_first_media_url.return_value = "https://qa.invalid/free-video.mp4" return provider def store_media(*, task, media): del media return Asset.objects.create( team=task.team, created_by=task.created_by, name="QA 自由创作视频", asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=task, ) references = [ {"url": "https://qa.invalid/ref.png", "type": "image", "label": "图片"}, {"url": "https://qa.invalid/ref.mp4", "type": "video", "label": "视频", "duration": 1}, {"url": "https://qa.invalid/ref.mp3", "type": "audio", "label": "音频"}, ] policy = deepcopy(settings.MODEL_ROUTING_POLICY) policy["video"]["submit_retry_delays"] = [0] policy["jitter_ratio"] = 0 created: list[dict] = [] with ( override_settings(MODEL_ROUTING_POLICY=policy), patch("apps.ai.free_video.FREE_VIDEO_MODELS", {primary.name}), patch("apps.ai.services.get_video_provider", side_effect=provider_for), patch("apps.ai.tasks.poll_free_video_task.apply_async"), patch("apps.ai.free_video._store_free_video_media", side_effect=store_media), patch("apps.ai.free_video._notify_failure"), ): for name, description in ( ("free_video_first_success", "自由创作多素材首次提交成功并成片"), ("free_video_fallback_success", "自由创作多素材原模型重试后 Fallback 并成片"), ("free_video_state_unknown", "自由创作提交状态未知且禁止重提"), ("free_video_all_failed", "自由创作所有候选明确失败并释放预留"), ): scenario["name"] = name provider_mocks.clear() before_assets = Asset.objects.filter(team=team, category=Asset.Category.FREE_CREATE).count() task = submit_free_video( team=team, user=user, params={ "prompt": "QA 多素材自由创作", "mode": "universal", "model": primary.name, "aspect_ratio": "16:9", "resolution": "480p", "duration": 4, "generate_audio": True, "references": references, }, ) if task.status in (AITask.Status.SUBMITTED, AITask.Status.POLLING): task = finalize_free_video(task=task) task.refresh_from_db() created.append( { "scenario": name, "description": description, "task_id": str(task.id), "status": task.status, "attempts": task.model_attempts.count(), "models": list(task.model_attempts.values_list("model_name", flat=True)), "provider_task_id": task.provider_task_id, "new_assets": Asset.objects.filter( team=team, category=Asset.Category.FREE_CREATE ).count() - before_assets, "ledger": _ledger_counts(task), } ) account = CreditAccount.objects.get(team=team) return { "team": QA_TEAM_NAME, "tasks": created, "account": {"balance": str(account.balance), "reserved_balance": str(account.reserved_balance)}, } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument( "action", choices=( "setup", "setup-reference", "setup-tryon", "setup-platform", "setup-base", "setup-model-triview", "setup-project-triview", "setup-storyboard", "setup-entity", "setup-script", "setup-voice", "setup-video", "setup-free-video", "cleanup", ), ) args = parser.parse_args() _assert_safe() if args.action == "setup": result = setup() elif args.action == "setup-reference": result = setup(with_references=True) elif args.action == "setup-tryon": result = setup(tryon=True) elif args.action == "setup-platform": result = setup(platform=True) elif args.action == "setup-base": result = setup_base() elif args.action == "setup-model-triview": result = setup_model_triview() elif args.action == "setup-project-triview": result = setup_model_triview(project_scope=True) elif args.action == "setup-storyboard": result = setup_storyboard() elif args.action == "setup-entity": result = setup_entity() elif args.action == "setup-script": result = setup_script() elif args.action == "setup-voice": result = setup_voice() elif args.action == "setup-video": result = setup_video() elif args.action == "setup-free-video": result = setup_free_video() else: result = {"cleaned": cleanup()} print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()