"""自由创作·独立视频生成(不绑 project)。 移植自 jimeng-clone(apps/generation/views.py 的 video_generate_view / tasks.py 轮询), 嫁接 AirShelf 底座:AITask 状态机 + CreditAccount 三段式计费(reserve/charge/release)+ TOS 存储 + VolcanoArkProvider。 链路: submit_free_video → 校验/估价/并发闸 → AITask(RESERVED) → 调火山(SUBMITTED) → 派兜底轮询 finalize_free_video → 查火山;终态幂等化(POSTPROCESSING 认领防双 poll 双扣) 成功: 下载→TOS→Asset(FREE_CREATE)+首帧封面 → 按真实 tokens 结算 失败: 错误码映射中文 + 退费 web poll 端点与 worker 兜底任务共用 finalize;本地无 worker 也能全程收尾。 """ import logging import re import uuid from datetime import timedelta from decimal import Decimal from io import BytesIO from django.conf import settings from django.core.exceptions import ObjectDoesNotExist from django.db import transaction from django.utils import timezone from apps.assets.models import Asset, AssetFile, FreeAsset, FreeAssetGroup from apps.assets.storage import TosStorage from apps.billing.pricing import settle_video_from_payload, quote_video_estimate, video_quote_payload, video_reserve_amount from apps.billing.services.ledger import charge_reserved_credit, release_credit, reserve_credit from .models import AITask, ModelConfig from .providers.volcano import VolcanoArkProvider from .generation_errors import classify_generation_error, public_error_for_task from .media_probe import REF_DURATION_MAX from .video_errors import parse_provider_error from .video_pricing import get_resolution logger = logging.getLogger(__name__) FREE_VIDEO_MODELS = { "doubao-seedance-2-5-260628", "doubao-seedance-2-0-260128", "doubao-seedance-2-0-fast-260128", "doubao-seedance-2-0-mini-260615", } HIGH_RES_MODEL = "doubao-seedance-2-0-260128" # 1080p/4k 仅标准档(火山限制) # 视频复刻固定 Seedance 2.5(单次最长 30 秒)。原片只用来提炼分镜,不传给火山。 REPLACE_MODEL = "doubao-seedance-2-5-260628" # 出片时长的兜底上限。真实上限按 ModelConfig.metadata.durations 取(见 model_duration_range), # 元数据缺失时回落这里 —— 别写死 15,否则新模型上来就被老常量卡住。 DEFAULT_MAX_DURATION = 15 MIN_DURATION = 4 def model_duration_range(model_config) -> tuple[int, int]: """模型支持的出片时长区间。取 metadata.durations(catalog / 后台可配),缺失回落 4–15。""" meta = (model_config.metadata or {}) if model_config is not None else {} durations = (meta.get("capabilities") or {}).get("durations") or meta.get("durations") or [] values = [int(v) for v in durations if str(v).isdigit()] if not values: return MIN_DURATION, DEFAULT_MAX_DURATION return min(values), max(values) def model_resolutions(model_config) -> set[str]: """模型支持的分辨率档。缺失回落全集,交给火山自己拒 —— 总比把能用的档误拒好。""" meta = (model_config.metadata or {}) if model_config is not None else {} listed = (meta.get("capabilities") or {}).get("resolutions") or meta.get("resolutions") or [] values = {str(v) for v in listed if str(v) in RESOLUTIONS} return values or set(RESOLUTIONS) RATIOS = {"21:9", "16:9", "4:3", "1:1", "3:4", "9:16"} RESOLUTIONS = {"480p", "720p", "1080p", "4k"} MODES = {"universal", "keyframe"} IN_FLIGHT_STATUSES = ( AITask.Status.CREATED, # 视频复刻审核中也占并发,避免连点刷出一堆待审任务 AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING, AITask.Status.POSTPROCESSING, ) _ORPHAN_MATERIAL_MENTION_RE = re.compile(r"@(?:图片|视频|音频|素材)[^\s，。！？、；：,.!?;:)）]*") def find_orphan_material_mention(prompt: str, references: list) -> str | None: """prompt 里 @了素材类占位但 references 为空 → 返回该 mention(供 400 提示)。""" if not prompt or references: return None match = _ORPHAN_MATERIAL_MENTION_RE.search(prompt) return match.group(0) if match else None def _format_prompt_for_ark(prompt: str, label_placeholders: list[tuple[str, str]]) -> str: """@label → 「图片N/视频N/音频N」。火山只认「类型+序号」指代(官方 FAQ Q3), 文件名/asset id 会按位置概率对齐 → 人物颠倒。调用方保证按 label 长度降序, 防「碧」先于「碧碧」被替换的子串吞噬;用 str.replace 防 label 含正则元字符崩溃。 用户原文保留在 request_payload 供「再次生成」回填 mention chip。""" result = prompt for label, placeholder in label_placeholders: if not label: continue result = result.replace(f"@{label}", placeholder) return result def _normalize_remote_asset_id(remote_id: str) -> str: # 火山返回 "Asset-xxx",引用时须小写前缀 "asset-xxx" if remote_id.startswith("Asset-"): return "asset-" + remote_id[6:] return remote_id def _refresh_processing_free_asset(free_asset: FreeAsset) -> bool: """本地 processing 的素材实时查火山刷新。返回是否 active。best-effort。""" try: from apps.assets import assets_client result = assets_client.get_asset(free_asset.remote_asset_id) if result and result.get("Status") == "Active": free_asset.status = FreeAsset.Status.ACTIVE free_asset.url = result.get("Url", free_asset.url) or free_asset.url free_asset.save(update_fields=["status", "url", "updated_at"]) return True except Exception: # noqa: BLE001 — 刷新失败按未就绪处理 logger.warning("free asset %s refresh failed", free_asset.id, exc_info=True) return False def _guard_asset_reference(asset: Asset, label: str) -> None: """三库引用前的审核闸(模块4 · 4.2)。 平台生成的资产免审直接过;用户上传的必须真过一遍审核才能当生成参考。 判据是 Asset.source,**不是**「在不在资产库里」—— 按库免审等于把审核架空: 用户传一张图进库、再从自由创作引用出去,就绕过了整套人像审核。 """ from apps.assets.review import poll_asset_review, reference_review_state, submit_asset_for_review state = reference_review_state(asset) if state == "processing": poll_asset_review(asset) # 实时刷一次,别让用户干等下一轮轮询 state = reference_review_state(asset) if state == "allowed": return name = label or asset.name or "未命名" if state == "processing": raise ValueError(f"素材「{name}」正在审核中，请稍后再引用") if state == "failed": raise ValueError(f"素材「{name}」未通过审核，不能用作生成参考") # 从没送过审(资产库上传不自动送审):这里补送一次,用户等审核结果即可,不必回去手动点 submit_asset_for_review(asset, force=True) raise ValueError(f"素材「{name}」是上传素材，已提交审核，通过后即可引用") def build_content_items( *, team, prompt: str, mode: str, references: list, max_ref_seconds: float = REF_DURATION_MAX ) -> dict: """references → 火山 content_items + api_prompt(@label 已替换)。 移植 jimeng views.py:369-567:URL 去重、blob: 拦截、素材库引用(FreeAsset → asset://)、 素材组展开、图≤9/视频≤3/音频≤3 校验、@label→「图片N」映射(不变量:任意时刻 image_n/video_n/audio_n == content_items 里该类型已 push 的个数)。 校验失败抛 ValueError(用户友好中文),由 view 转 400。 """ content_items: list[dict] = [] snapshots: list[dict] = [] resolved_library_assets: list[dict[str, str]] = [] seen_urls: set[str] = set() group_cache: dict[str, list[FreeAsset]] = {} label_to_placeholder: dict[str, str] = {} image_n = video_n = audio_n = 0 video_duration_total = 0.0 # 输入参考视频总时长(token 公式的输入项 + ≤15s 校验) def _placeholder_for(asset_type: str) -> str: if asset_type == "Video": return f"视频{video_n}" if asset_type == "Audio": return f"音频{audio_n}" return f"图片{image_n}" def _push(kind: str, url: str, role: str, duration: float = 0.0) -> str: """push 一个 content_item 并递增 counter,返回素材类型(Image/Video/Audio)。""" nonlocal image_n, video_n, audio_n, video_duration_total if kind == "video": video_n += 1 video_duration_total += duration or 0.0 item: dict = {"type": "video_url", "video_url": {"url": url}} if role: item["role"] = role content_items.append(item) return "Video" if kind == "audio": audio_n += 1 item = {"type": "audio_url", "audio_url": {"url": url}} if role: item["role"] = role content_items.append(item) return "Audio" image_n += 1 item = {"type": "image_url", "image_url": {"url": url}} if role: item["role"] = role content_items.append(item) return "Image" def _resolve_group_assets(group: FreeAssetGroup) -> list[FreeAsset]: resolved: list[FreeAsset] = [] for fa in group.assets.exclude(remote_asset_id="").order_by("created_at"): if fa.status == FreeAsset.Status.PROCESSING and not _refresh_processing_free_asset(fa): continue # 未就绪的跳过 if fa.status != FreeAsset.Status.ACTIVE: continue resolved.append(fa) return resolved def _remember_library_asset(fa: FreeAsset) -> None: if any(item["local_asset_id"] == str(fa.id) for item in resolved_library_assets): return resolved_library_assets.append( { "local_asset_id": str(fa.id), "remote_asset_id": fa.remote_asset_id, "submitted_remote_asset_id": _normalize_remote_asset_id(fa.remote_asset_id), "asset_name": fa.name, } ) for ref in references or []: url = str(ref.get("url") or "") ref_type = str(ref.get("type") or "image") role = str(ref.get("role") or "") label = str(ref.get("label") or "") source = str(ref.get("source") or "upload") duration = float(ref.get("duration") or 0) dedupe_base = url or f"{source}:{ref.get('asset_id') or ref.get('group_id')}" dedupe_key = f"{role}:{dedupe_base}" if mode == "keyframe" else dedupe_base if dedupe_key in seen_urls: continue seen_urls.add(dedupe_key) if url.startswith("blob:"): raise ValueError(f"素材「{label or '未命名'}」上传失败，请删除后重新添加") snap = {"url": url, "type": ref_type, "role": role, "label": label, "source": source} if ref.get("thumb_url"): snap["thumb_url"] = ref["thumb_url"] if duration: snap["duration"] = duration if ref.get("asset_id"): snap["asset_id"] = str(ref["asset_id"]) if ref.get("group_id"): snap["group_id"] = str(ref["group_id"]) snapshots.append(snap) # 素材库单素材:FreeAsset → asset://{remote_id} if source == "library" and ref.get("asset_id"): fa = FreeAsset.objects.filter( id=ref["asset_id"], group__team=team, group__is_deleted=False ).first() if fa is None: raise ValueError(f"素材「{label or '未命名'}」不存在或已被删除") if fa.status == FreeAsset.Status.PROCESSING and not _refresh_processing_free_asset(fa): raise ValueError(f"素材「{label or fa.name}」尚在处理中，请稍后重试") if fa.status != FreeAsset.Status.ACTIVE or not fa.remote_asset_id: raise ValueError(f"素材「{label or fa.name}」尚未就绪，请稍后重试") resolved_url = f"asset://{_normalize_remote_asset_id(fa.remote_asset_id)}" _remember_library_asset(fa) kind = {"Video": "video", "Audio": "audio"}.get(fa.asset_type, "image") if mode == "keyframe": if kind != "image": raise ValueError("首尾帧模式仅支持图片素材") effective_role = role if role in {"first_frame", "last_frame"} else "first_frame" else: effective_role = "reference_video" if kind == "video" else ("reference_audio" if kind == "audio" else "reference_image") asset_type = _push(kind, resolved_url, effective_role, fa.duration or 0.0) if label and label not in label_to_placeholder: label_to_placeholder[label] = _placeholder_for(asset_type) continue # 素材组引用:展开组内全部 active 素材(一个 label 对应 N 素材,语义变化 → 不登记 label) if source == "library_group" and ref.get("group_id"): gid = str(ref["group_id"]) if gid not in group_cache: group = FreeAssetGroup.objects.filter(id=gid, team=team, is_deleted=False).first() group_cache[gid] = _resolve_group_assets(group) if group else [] asset_list = group_cache[gid] if not asset_list: raise ValueError(f"素材「{label or '未命名'}」尚未就绪，请在素材库中确认状态为「可用」后重试") for fa in asset_list: kind = {"Video": "video", "Audio": "audio"}.get(fa.asset_type, "image") asset_url = f"asset://{_normalize_remote_asset_id(fa.remote_asset_id)}" _push( kind, asset_url, "reference_video" if kind == "video" else ("reference_audio" if kind == "audio" else "reference_image"), fa.duration or 0.0, ) _remember_library_asset(fa) continue # 三库引用(模块4 · 4.1):资产库 / 模特库 / 商品库 挑出来的东西最终都是一行 Asset, # 所以后端只认一种 source=asset,三个库的差别全在前端的 picker 上。 if source == "asset" and ref.get("asset_id"): asset = Asset.objects.filter(id=ref["asset_id"], team=team, is_deleted=False).first() if asset is None: raise ValueError(f"素材「{label or '未命名'}」不存在或已被删除") _guard_asset_reference(asset, label) from .services import _asset_preview_url, _seedance_ref_url raw_url = _asset_preview_url(asset) if not raw_url: raise ValueError(f"素材「{label or asset.name}」没有可用文件，无法引用") # 已登记火山素材库的走 asset://(写实人脸走直链会被 InputImageSensitiveContentDetected 拒) resolved_url = _seedance_ref_url(raw_url, asset.review_status, asset.review_remote_id) kind = asset.asset_type if asset.asset_type in {"image", "video", "audio"} else "image" if mode == "keyframe": if kind != "image": raise ValueError("首尾帧模式仅支持图片素材") effective_role = role if role in {"first_frame", "last_frame"} else "first_frame" else: effective_role = "reference_video" if kind == "video" else ("reference_audio" if kind == "audio" else "reference_image") asset_type = _push(kind, resolved_url, effective_role, duration) if label and label not in label_to_placeholder: label_to_placeholder[label] = _placeholder_for(asset_type) continue # 直传素材(已上传 TOS 的直链)。resolved_url 可覆盖为 asset://(官方模特跨团队等) push_url = str(ref.get("resolved_url") or url) if ref_type == "image": # 参考图模式下所有图 role 必须 reference_image;keyframe 用 first_frame/last_frame effective_role = "reference_image" if mode == "universal" else (role or "first_frame") asset_type = _push("image", push_url, effective_role) elif ref_type == "video": asset_type = _push("video", push_url, role or "reference_video", duration) elif ref_type == "audio": asset_type = _push("audio", push_url, role or "reference_audio", duration) else: logger.warning("unknown ref_type=%s url=%s label=%s, skipped", ref_type, url, label) continue if label and label not in label_to_placeholder: label_to_placeholder[label] = _placeholder_for(asset_type) if image_n > 9: raise ValueError(f"参考图片最多 9 张（含素材库引用，同一素材按 1 张计算），当前 {image_n} 张，请减少后重试") if video_n > 3: raise ValueError(f"参考视频最多 3 条，当前 {video_n} 条，请减少后重试") if audio_n > 3: raise ValueError(f"参考音频最多 3 条，当前 {audio_n} 条，请减少后重试") if audio_n > 0 and image_n + video_n == 0: raise ValueError("音频不能单独作为参考素材，请同时提供参考图片或视频") if video_duration_total > max_ref_seconds: # 上限跟着模型走:2.0 是 15 秒,2.5 是 30 秒。 raise ValueError(f"参考视频总时长不能超过 {int(max_ref_seconds)} 秒，请缩短后重试") # @label 替换:按 label 长度降序,防子串吞噬 ordered = sorted(label_to_placeholder.items(), key=lambda kv: len(kv[0]), reverse=True) api_prompt = _format_prompt_for_ark(prompt, ordered) return { "content_items": content_items, "api_prompt": api_prompt, "snapshots": snapshots, "resolved_library_assets": resolved_library_assets, "image_n": image_n, "video_n": video_n, "audio_n": audio_n, "video_duration_total": video_duration_total, } def _unavailable_library_asset_target(resolved_assets: list[dict], raw_message: str) -> dict | None: """从本次已解析引用中精确定位 Provider 指出的失效素材；不按陌生字符串模糊查库。""" message = str(raw_message or "").casefold() matches = [] for item in resolved_assets: submitted_id = str(item.get("submitted_remote_asset_id") or "").casefold() original_id = str(item.get("remote_asset_id") or "").casefold() if (submitted_id and submitted_id in message) or (original_id and original_id in message): matches.append(item) if len(matches) == 1: return matches[0] if not matches and len(resolved_assets) == 1: return resolved_assets[0] return None def _reap_stale_free_video_tasks(*, team) -> None: """僵尸回收(趁每次新提交顺手做,无需定时任务): · RESERVED 超过视频提交阶段总时限:没提交到火山就死(worker 崩溃/进程重启)→ 标失败退费; · SUBMITTED/POLLING 已有远端任务 ID，只能按供应商终态收尾，不按本地等待时间回收; · POSTPROCESSING 超 30 分钟:转存/结算中途崩溃 → 标失败退费(火山可能已出片,平台承担该笔成本)。""" now = timezone.now() from .routing_policy import load_model_routing_policy video_policy = load_model_routing_policy().video buckets = [ ( [AITask.Status.CREATED], {"updated_at__lt": now - timedelta(minutes=16)}, "素材审核超时(自动回收)", ), ( [AITask.Status.RESERVED], {"updated_at__lt": now - timedelta(seconds=video_policy.submit_total_timeout)}, "任务未在配置的提交总时限内完成(自动回收)", ), ( [AITask.Status.POSTPROCESSING], {"updated_at__lt": now - timedelta(minutes=30)}, "视频结果处理超时(自动回收)", ), ] for statuses, stale_filter, reason in buckets: stale = AITask.objects.filter( team=team, project__isnull=True, task_type=AITask.Type.FREE_VIDEO, status__in=statuses, **stale_filter, ) for task in stale: try: with transaction.atomic(): locked = AITask.objects.select_for_update().get(id=task.id) if locked.status not in statuses: continue locked.status = AITask.Status.FAILED locked.error_message = reason locked.completed_at = timezone.now() locked.save(update_fields=["status", "error_message", "completed_at", "updated_at"]) try: reservation = locked.credit_reservation except ObjectDoesNotExist: reservation = None if reservation is not None: release_credit(reservation=reservation, reason=reason) except Exception: # noqa: BLE001 — 单个回收失败不应阻断新提交 logger.warning("reap stale free video task %s failed", task.id, exc_info=True) continue def submit_free_video(*, team, user, params: dict) -> AITask: """提交一条自由创作视频。校验/估价/预留在前(失败不留半套),火山调用在事务外。 创建失败不抛:返回 FAILED 任务(带友好中文错误),前端渲染失败卡。校验类错误抛 ValueError → 400。""" prompt = str(params.get("prompt") or "").strip() feature = str(params.get("feature") or "free_video").strip() or "free_video" mode = str(params.get("mode") or "universal") model_name = str(params.get("model") or HIGH_RES_MODEL) aspect_ratio = str(params.get("aspect_ratio") or "16:9") resolution = str(params.get("resolution") or "720p") generate_audio = bool(params.get("generate_audio", True)) search_mode = str(params.get("search_mode") or "off") references = params.get("references") or [] try: duration = int(params.get("duration") or 5) except (TypeError, ValueError): raise ValueError("时长参数无效") try: seed = int(params.get("seed") if params.get("seed") is not None else -1) except (TypeError, ValueError): seed = -1 if not prompt: raise ValueError("提示词不能为空") if mode not in MODES: raise ValueError("生成模式无效") if model_name not in FREE_VIDEO_MODELS: raise ValueError("模型无效") if aspect_ratio not in RATIOS: raise ValueError("画面比例无效") if resolution not in RESOLUTIONS: raise ValueError("分辨率无效") get_resolution(aspect_ratio, resolution) # 组合合法性 fail loud orphan = find_orphan_material_mention(prompt, references) if orphan: raise ValueError(f"「{orphan}」对应的内容为空，请补充素材或删除该引用") if mode == "keyframe": roles = [str(r.get("role") or "") for r in references] if any(str(r.get("type") or "image") != "image" for r in references): raise ValueError("首尾帧模式仅支持图片素材") if "first_frame" not in roles: raise ValueError("首尾帧模式需要提供首帧图片") if len(references) > 2: raise ValueError("首尾帧模式最多提供首帧和尾帧各一张图片") model_config = ( ModelConfig.objects.select_related("provider") .filter(name=model_name, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE) .first() ) if model_config is None: raise ValueError("视频模型未配置，请联系管理员") # 分辨率与时长都按所选模型自己的能力判,不再写死「1080p/4k 仅 2.0 标准档」—— # Seedance 2.5 同样支持 1080p/4k,写死会把它误拒。能力表见 ModelConfig.metadata。 allowed_resolutions = model_resolutions(model_config) if resolution not in allowed_resolutions: raise ValueError( f"{model_config.display_name or model_name} 不支持 {resolution}，请切换模型或降低分辨率" ) min_seconds, max_seconds = model_duration_range(model_config) if not min_seconds <= duration <= max_seconds: raise ValueError(f"视频时长需在 {min_seconds}-{max_seconds} 秒之间") _reap_stale_free_video_tasks(team=team) # 团队并发闸(移植 jimeng Layer2.6):视频是长时高价任务,必须限并发 max_concurrent = int(getattr(settings, "FREE_VIDEO_MAX_CONCURRENT", 3)) in_flight = AITask.objects.filter( team=team, task_type=AITask.Type.FREE_VIDEO, status__in=IN_FLIGHT_STATUSES ).count() if in_flight >= max_concurrent: raise ValueError(f"当前有 {in_flight} 个视频任务进行中（上限 {max_concurrent}），请等待完成后再提交") built = build_content_items( team=team, prompt=prompt, mode=mode, references=references, max_ref_seconds=max_seconds, ) # 统一计价引擎:¥成本 × 毛利系数 → 积分;预留 = 积分 × buffer(均为 BillingConfig 可配) tokens, quote = quote_video_estimate( model_config, aspect_ratio=aspect_ratio, resolution=resolution, duration=duration, references=built["snapshots"], team=team, ) reserve_amount = video_reserve_amount(quote.points, rule=quote.meta.get("rule")) request_payload = { "feature": feature, "mode": mode, "model": model_name, "endpoint": model_config.endpoint, "prompt": prompt, "api_prompt": built["api_prompt"], "aspect_ratio": aspect_ratio, "resolution": resolution, "duration": duration, "seed": seed, "generate_audio": generate_audio, "search_mode": search_mode, "estimated_tokens": tokens, # 计价快照:挂牌秒价/团队系数下单时钉死,结算只认这些 **video_quote_payload(quote), "references": built["snapshots"], "model_routing_v1": True, } extra = params.get("extra_payload") if isinstance(extra, dict): protected = set(request_payload) for key, value in extra.items(): if key in protected or value in (None, ""): continue request_payload[key] = value # 建任务 + 预留同一事务:余额不足/限额拦截时回滚任务行,不留半套 with transaction.atomic(): task = AITask.objects.create( team=team, created_by=user, project=None, task_type=AITask.Type.FREE_VIDEO, status=AITask.Status.CREATED, model_config=model_config, idempotency_key=f"free_video:{team.id}:{uuid.uuid4()}", request_payload=request_payload, estimated_cost=quote.points, base_cost=Decimal("0"), ) try: reserve_credit(team=team, user=user, task=task, amount=reserve_amount) except ValueError as exc: if "insufficient credit" in str(exc): raise ValueError("团队余额不足，请充值后重试") from exc raise task.status = AITask.Status.RESERVED task.save(update_fields=["status", "updated_at"]) return _dispatch_free_video_provider( task=task, built=built, model_config=model_config, aspect_ratio=aspect_ratio, duration=duration, resolution=resolution, generate_audio=generate_audio, seed=seed, search_mode=search_mode, feature=feature, mode=mode, poll_countdown=30, ) def start_pending_free_video(task: AITask) -> AITask: """CREATED 任务审核已过后:预留积分 + 调火山。并发 poll 用行锁认领,失败不扣费。""" if task.status != AITask.Status.CREATED: return task payload = task.request_payload or {} prompt = str(payload.get("prompt") or "").strip() mode = str(payload.get("mode") or "universal") aspect_ratio = str(payload.get("aspect_ratio") or "16:9") resolution = str(payload.get("resolution") or "720p") generate_audio = bool(payload.get("generate_audio", True)) search_mode = str(payload.get("search_mode") or "off") feature = str(payload.get("feature") or "free_video") try: duration = int(payload.get("duration") or 5) except (TypeError, ValueError): duration = 5 try: seed = int(payload.get("seed") if payload.get("seed") is not None else -1) except (TypeError, ValueError): seed = -1 references = list(payload.get("references") or []) if feature == "video_replace" and payload.get("replace_mode") == "product": # 商品复刻:原片只用来提炼,绝不能进 Seedance,否则真人素材直接被火山拒。 # 角色复刻走另一条已入火山素材库的 video-reference 链路，必须保留视频。 references = [item for item in references if (item or {}).get("type") != "video"] try: built = build_content_items( team=task.team, prompt=prompt, mode=mode, references=references, max_ref_seconds=model_duration_range(task.model_config)[1], ) except ValueError as exc: message = str(exc) if "正在审核" in message or "已提交审核" in message or "尚未完成合规审核" in message: return task return _fail_pending_free_video(task, message) # 角色视频编辑向火山传 duration=-1 / ratio=adaptive；账务仍按参考视频的真实时长与比例预估。 billing_duration = int(payload.get("billing_duration") or duration) billing_ratio = str(payload.get("billing_aspect_ratio") or aspect_ratio) tokens, quote = quote_video_estimate( task.model_config, aspect_ratio=billing_ratio, resolution=resolution, duration=billing_duration, references=built["snapshots"], team=task.team, ) reserve_amount = video_reserve_amount(quote.points, rule=quote.meta.get("rule")) with transaction.atomic(): locked = ( AITask.objects.select_for_update() .select_related("model_config", "model_config__provider", "team", "created_by") .get(id=task.id) ) if locked.status != AITask.Status.CREATED: return locked try: reserve_credit(team=locked.team, user=locked.created_by, task=locked, amount=reserve_amount) except ValueError as exc: message = "团队余额不足，请充值后重试" if "insufficient credit" in str(exc) else str(exc) locked.status = AITask.Status.FAILED locked.error_code = "user_credit_insufficient" if "余额不足" in message else "invalid_input" locked.error_message = message[:2000] locked.completed_at = timezone.now() locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"]) return locked next_payload = dict(locked.request_payload or {}) next_payload["api_prompt"] = built["api_prompt"] next_payload["references"] = built["snapshots"] next_payload["estimated_tokens"] = tokens next_payload["review_pending"] = False next_payload.update(video_quote_payload(quote)) locked.request_payload = next_payload locked.estimated_cost = quote.points locked.status = AITask.Status.RESERVED locked.save(update_fields=["request_payload", "estimated_cost", "status", "updated_at"]) return _dispatch_free_video_provider( task=locked, built=built, model_config=locked.model_config, aspect_ratio=aspect_ratio, duration=duration, resolution=resolution, generate_audio=generate_audio, seed=seed, search_mode=search_mode, feature=feature, mode=mode, poll_countdown=30, ) def _fail_pending_free_video(task: AITask, message: str) -> AITask: if task.status not in (AITask.Status.CREATED, AITask.Status.RESERVED): return task task.status = AITask.Status.FAILED task.error_code = "content_rejected" task.error_message = message[:2000] task.completed_at = timezone.now() task.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"]) _notify_failure(task, raw=message, hint=message) return task def _dispatch_free_video_provider( *, task, built, model_config, aspect_ratio, duration, resolution, generate_audio, seed, search_mode, feature, mode, poll_countdown=30, ): """RESERVED 任务调火山创建。失败退费。""" try: from .services import execute_routed_video_submit routed = execute_routed_video_submit( task=task, primary_model=model_config, prompt=built["api_prompt"], ratio=aspect_ratio, duration=duration, resolution=resolution, reference_images=[], content_items=built["content_items"], pricing_references=built["snapshots"], generate_audio=generate_audio, seed=seed if seed != -1 else None, search_mode=search_mode, request_summary={"feature": feature, "mode": mode}, ) response, provider_task_id = routed.value task.refresh_from_db(fields=["base_cost"]) task.provider_task_id = provider_task_id task.response_payload = response payload = dict(task.request_payload or {}) payload["actual_model_config_id"] = str(routed.actual_model.id) task.request_payload = payload task.status = AITask.Status.SUBMITTED task.submitted_at = timezone.now() task.save( update_fields=[ "provider_task_id", "response_payload", "request_payload", "status", "submitted_at", "updated_at", ] ) except Exception as exc: # noqa: BLE001 — 创建失败:标失败退费,返回失败卡(不向上抛) code, raw_message = parse_provider_error(exc) public_error = classify_generation_error( exc, operation="video_generate", provider_code=code, reference_id=str(task.id), ) if public_error.code == "asset_unavailable": target = _unavailable_library_asset_target( built["resolved_library_assets"], raw_message, ) if target is not None: try: from apps.assets.free_asset_state import mark_remote_asset_unavailable mark_remote_asset_unavailable( team=task.team, local_asset_id=target["local_asset_id"], remote_asset_id=target["remote_asset_id"], ) except Exception: # noqa: BLE001 - 素材状态同步是附带自愈，不能遮蔽任务失败与退费。 logger.exception("failed to mark unavailable free asset for task %s", task.id) task.status = AITask.Status.FAILED task.error_code = (code or "CreateTaskError")[:64] task.error_message = raw_message[:2000] task.completed_at = timezone.now() task.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"]) release_credit(reservation=task.credit_reservation, reason=raw_message[:200]) _notify_failure( task, raw=f"[{code}] {raw_message}" if code else raw_message, hint=public_error.fallback_message, ) logger.warning("free video create failed: %s", exc) return task try: from .tasks import poll_free_video_task poll_free_video_task.apply_async(args=[str(task.id), 0], countdown=poll_countdown) except Exception: # noqa: BLE001 logger.error("poll_free_video_task enqueue failed; relying on client polling", exc_info=True) return task def _notify_failure(task: AITask, *, raw: str, hint: str) -> None: from .services import notify_generation_failure notify_generation_failure( task=task, project=None, recipient=task.created_by, stage_label="视频复刻" if (task.request_payload or {}).get("feature") == "video_replace" else "自由创作视频", raw=raw, hint=hint, ) def _store_free_video_media(*, task: AITask, media: str = "", video_bytes: bytes | None = None) -> Asset: """下载火山结果(或已拼好的字节) → 转存 TOS → 建 Asset + ffmpeg 抽首帧封面。""" content_type = "video/mp4" if video_bytes is None: fileobj, content_type = VolcanoArkProvider.media_to_bytes(media) if "video" not in content_type: content_type = "video/mp4" # 先取字节再上传:boto3 upload_fileobj 完成后会 close 掉 BytesIO,之后 getvalue() 直接抛 # "I/O operation on closed file",封面抽帧就永远做不了(实测踩坑)。 video_bytes = fileobj.getvalue() if isinstance(fileobj, BytesIO) else b"" fileobj = BytesIO(video_bytes) asset_id = uuid.uuid4() object_key = f"teams/{task.team_id}/free-create/{asset_id}.mp4" stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type) payload = task.request_payload or {} prompt = payload.get("prompt") or "" feature = payload.get("feature") or "free_video" subject = str(payload.get("subject_name") or "").strip() if feature == "video_replace": mode_label = "角色复刻" if payload.get("replace_mode") == "character" else "商品复刻" asset_name = f"{subject}{mode_label}" if subject else mode_label elif feature == "omni_create": asset_name = prompt[:255] or "全能创作视频" else: asset_name = prompt[:255] or "自由创作视频" asset = Asset.objects.create( id=asset_id, team=task.team, created_by=task.created_by, # Asset.name 是通用资产字段，受 255 字符上限约束；完整显示标题由资产接口从 # 关联任务的 request_payload.prompt 派生，避免再出现 50 字符业务截断。 name=asset_name[:255], asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=task, # 全能创作成品只挂在会话结果卡上,不进资产库/专业创作用的视频列表。 in_library=feature != "omni_create", metadata={"feature": feature}, ) AssetFile.objects.create( asset=asset, object_key=stored.object_key, bucket=stored.bucket, content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True, ) # 首帧封面(best-effort):任务流/资产库缩略图 try: if video_bytes: import subprocess import tempfile from pathlib import Path with tempfile.TemporaryDirectory(prefix="airshelf-fc-poster-") as tmp: tmp_dir = Path(tmp) video_path = tmp_dir / "in.mp4" poster_path = tmp_dir / "poster.jpg" video_path.write_bytes(video_bytes) proc = subprocess.run( ["ffmpeg", "-y", "-ss", "0", "-i", str(video_path), "-frames:v", "1", "-q:v", "3", str(poster_path)], capture_output=True, timeout=60, ) if proc.returncode == 0 and poster_path.exists() and poster_path.stat().st_size: poster_key = f"teams/{task.team_id}/free-create/{asset_id}-poster.jpg" poster_stored = TosStorage().upload_fileobj( fileobj=BytesIO(poster_path.read_bytes()), object_key=poster_key, content_type="image/jpeg" ) AssetFile.objects.create( asset=asset, object_key=poster_stored.object_key, bucket=poster_stored.bucket, content_type=poster_stored.content_type, size_bytes=poster_stored.size_bytes, is_primary=False, ) except Exception: # noqa: BLE001 — 封面仅用于展示,失败不阻断 logger.warning("free video poster extract failed for task %s", task.id, exc_info=True) return asset def finalize_free_video(*, task: AITask) -> AITask: """单次轮询 + 幂等终态化。web poll 端点与 worker 兜底共用。 防双 poll 双扣:succeeded 时先持锁把任务从 SUBMITTED/POLLING「认领」成 POSTPROCESSING, 并发的另一路看到 POSTPROCESSING 直接返回;认领者独占 下载→TOS→建资产→结算 全程。 认领后崩溃由 _reap_stale_free_video_tasks(30 分钟)兜底退费。""" if task.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING): return task if not task.provider_task_id: return task from .routing_policy import load_model_routing_policy from .services import get_video_provider video_policy = load_model_routing_policy().video submit_attempt = ( task.model_attempts.filter(status="succeeded", operation="video_generate") .select_related("model_config__provider") .order_by("-sequence") .first() ) actual_model = submit_attempt.model_config if submit_attempt and submit_attempt.model_config else task.model_config provider = get_video_provider(actual_model) response = provider.poll_video_task( endpoint=actual_model.endpoint, provider_task_id=task.provider_task_id, timeout=video_policy.poll_request_timeout, ) remote_status = str(response.get("status") or "") if remote_status in {"queued", "running", "processing", "submitted"}: # 仍在生成:只在首次进入 POLLING 时落一次库(不逐次回写完整 response,省写带宽) if task.status != AITask.Status.POLLING: task.status = AITask.Status.POLLING task.save(update_fields=["status", "updated_at"]) return task if remote_status in {"failed", "expired", "cancelled"}: err = response.get("error") or {} code = str(err.get("code") or "") raw_message = str(err.get("message") or "video generation failed") public_error = classify_generation_error( RuntimeError(raw_message), operation="video_generate", provider_code=code, reference_id=str(task.id), ) with transaction.atomic(): locked = AITask.objects.select_for_update().get(id=task.id) if locked.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING): return locked locked.status = AITask.Status.FAILED locked.error_code = code[:64] locked.error_message = raw_message[:2000] locked.response_payload = response locked.completed_at = timezone.now() locked.save( update_fields=["status", "error_code", "error_message", "response_payload", "completed_at", "updated_at"] ) release_credit(reservation=locked.credit_reservation, reason=raw_message[:200]) _notify_failure(locked, raw=f"[{code}] {raw_message}", hint=public_error.fallback_message) from apps.ai.creation import sync_generating_for_task sync_generating_for_task(locked) return locked # succeeded —— 认领 POSTPROCESSING(并发 finalize 只有一路进入慢活) with transaction.atomic(): locked = AITask.objects.select_for_update().get(id=task.id) if locked.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING): return locked locked.status = AITask.Status.POSTPROCESSING locked.save(update_fields=["status", "updated_at"]) payload = dict(locked.request_payload or {}) try: media = provider.extract_first_media_url(response) try: _store_free_video_media(task=locked, media=media) except Exception: # noqa: BLE001 — TOS 转存失败:兜底记火山临时 URL(约 24h 内可看),不吞成功 logger.exception("free video TOS store failed for task %s, falling back to raw url", locked.id) payload["fallback_video_url"] = media payload["fallback_note"] = "结果转存失败，当前链接约 24 小时后失效" usage = response.get("usage") or {} try: total_tokens = int(usage.get("total_tokens") or 0) except (TypeError, ValueError): total_tokens = 0 settle = settle_video_from_payload(actual_model, payload=payload, tokens=total_tokens) if settle.meta.get("rule") == "missing_usage": actual, base_cost = locked.estimated_cost, locked.base_cost else: actual, base_cost = settle.points, settle.base_cost_yuan if total_tokens > 0: payload["actual_tokens"] = total_tokens if settle.meta.get("rate"): payload["points_per_yuan_snapshot"] = settle.meta["rate"] seed_out = response.get("seed") if seed_out is not None: payload["seed_used"] = seed_out with transaction.atomic(): locked = AITask.objects.select_for_update().get(id=locked.id) if locked.status != AITask.Status.POSTPROCESSING: return locked reservation = locked.credit_reservation if actual > reservation.amount: # ledger 禁超预留扣费 → clamp 到预留额,差额平台承担并告警(长期观测调 buffer) logger.warning( "free video task %s actual cost %s exceeds reserved %s, clamped", locked.id, actual, reservation.amount, ) actual = reservation.amount locked.status = AITask.Status.SUCCEEDED locked.actual_cost = actual locked.base_cost = base_cost locked.request_payload = payload locked.response_payload = response locked.completed_at = timezone.now() locked.save( update_fields=["status", "actual_cost", "base_cost", "request_payload", "response_payload", "completed_at", "updated_at"] ) charge_reserved_credit(reservation=reservation, actual_amount=actual) from apps.ai.creation import sync_generating_for_task sync_generating_for_task(locked) return locked except Exception as exc: # noqa: BLE001 — 后处理失败:标失败退费(release 幂等,已扣则不动) logger.exception("free video finalize failed for task %s", locked.id) with transaction.atomic(): locked = AITask.objects.select_for_update().get(id=locked.id) if locked.status != AITask.Status.POSTPROCESSING: return locked locked.status = AITask.Status.FAILED locked.error_code = "PostprocessError" public_error = classify_generation_error( exc, operation="video_generate", internal_kind="processing_failed", reference_id=str(locked.id), ) locked.error_message = str(exc)[:2000] locked.completed_at = timezone.now() locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"]) release_credit(reservation=locked.credit_reservation, reason=str(exc)[:200]) _notify_failure(locked, raw=str(exc), hint=public_error.fallback_message) from apps.ai.creation import sync_generating_for_task sync_generating_for_task(locked) return locked def rehydrate_ref_urls(references) -> list: """参考素材快照里的 URL 是提交当时签的 TOS 预签名链(1 小时过期),历史记录隔天再点开就播不了。 凡是带 asset_id 的,按当前资产重新取长期直链(顺带补上视频封面),老任务也一起救回来。 素材库引用(source=library)的 asset_id 指向 FreeAsset 不是 Asset,查不到就原样保留快照。""" refs = [item for item in (references or [])] ids = set() for item in refs: if not isinstance(item, dict) or item.get("source") == "library": continue if item.get("asset_id"): try: ids.add(uuid.UUID(str(item["asset_id"]))) except (ValueError, TypeError, AttributeError): continue if not ids: return refs from .services import asset_stable_url assets = {str(a.id): a for a in Asset.objects.filter(id__in=ids).prefetch_related("files")} fresh = [] for item in refs: asset = assets.get(str(item.get("asset_id") or "")) if isinstance(item, dict) else None url, thumb = asset_stable_url(asset) if asset is not None else ("", "") if not url: fresh.append(item) continue fresh.append({**item, "url": url, **({"thumb_url": thumb} if thumb else {})}) return fresh def serialize_free_video_task(task: AITask, *, include_deleted_assets: bool = False) -> dict: """任务 → 前端契约。视频/封面直链从 generated_assets 取(TOS 公读),转存失败回落火山临时 URL。""" payload = task.request_payload or {} video_url = "" thumbnail_url = "" try: storage = TosStorage() if settings.TOS.get("endpoint") else None except Exception: # noqa: BLE001 storage = None def _file_url(f) -> str: if f.preview_url: return f.preview_url if storage and f.object_key: try: return storage.public_url(object_key=f.object_key, bucket=f.bucket or None) except Exception: # noqa: BLE001 return "" return "" generated_assets = list(task.generated_assets.all()) for asset in generated_assets: if asset.purged_at is not None or (asset.is_deleted and not include_deleted_assets): continue for f in asset.files.all(): url = _file_url(f) if not url: continue if f.is_primary and not video_url: video_url = url elif not f.is_primary and "image" in (f.content_type or "") and not thumbnail_url: thumbnail_url = url if not video_url and not generated_assets: video_url = payload.get("fallback_video_url") or "" public_error = public_error_for_task(task, operation="video_generate") return { "id": str(task.id), "status": task.status, "mode": payload.get("mode") or "universal", "model": payload.get("model") or "", "feature": payload.get("feature") or "free_video", "prompt": payload.get("prompt") or "", "aspect_ratio": payload.get("aspect_ratio") or "16:9", "resolution": payload.get("resolution") or "720p", "duration": payload.get("duration") or 5, "seed": payload.get("seed", -1), "seed_used": payload.get("seed_used"), "generate_audio": payload.get("generate_audio", True), "references": rehydrate_ref_urls(payload.get("references")), "estimated_tokens": payload.get("estimated_tokens") or 0, "actual_tokens": payload.get("actual_tokens") or 0, "estimated_cost": str(task.estimated_cost), "actual_cost": str(task.actual_cost), "error": public_error.as_dict() if public_error else None, "error_message": public_error.fallback_message if public_error else "", "fallback_note": payload.get("fallback_note") or "", "is_favorited": task.is_favorited, "video_url": video_url, "thumbnail_url": thumbnail_url, "created_at": task.created_at.isoformat() if task.created_at else None, "completed_at": task.completed_at.isoformat() if task.completed_at else None, }