770 lines
35 KiB
Python
770 lines
35 KiB
Python
"""自由创作·独立视频生成(不绑 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 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 quote_video_actual, quote_video_estimate, 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 .video_errors import parse_provider_error
|
||
from .video_pricing import get_resolution
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
FREE_VIDEO_MODELS = {
|
||
"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 仅标准档(火山限制)
|
||
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.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 build_content_items(*, team, prompt: str, mode: str, references: list) -> 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] = []
|
||
seen_urls: set[str] = set()
|
||
group_cache: dict[str, list[tuple[str, str, float]]] = {}
|
||
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[tuple[str, str, float]]:
|
||
resolved: list[tuple[str, str, float]] = []
|
||
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(
|
||
(f"asset://{_normalize_remote_asset_id(fa.remote_asset_id)}", fa.asset_type, fa.duration or 0.0)
|
||
)
|
||
return resolved
|
||
|
||
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)}"
|
||
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 asset_url, asset_type, dur in asset_list:
|
||
kind = {"Video": "video", "Audio": "audio"}.get(asset_type, "image")
|
||
_push(kind, asset_url, "reference_video" if kind == "video" else ("reference_audio" if kind == "audio" else "reference_image"), dur)
|
||
continue
|
||
|
||
# 直传素材(已上传 TOS 的直链)
|
||
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", url, effective_role)
|
||
elif ref_type == "video":
|
||
asset_type = _push("video", url, role or "reference_video", duration)
|
||
elif ref_type == "audio":
|
||
asset_type = _push("audio", 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 > 15:
|
||
raise ValueError("参考视频总时长不能超过 15 秒，请缩短后重试")
|
||
|
||
# @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,
|
||
"image_n": image_n,
|
||
"video_n": video_n,
|
||
"audio_n": audio_n,
|
||
"video_duration_total": video_duration_total,
|
||
}
|
||
|
||
|
||
def _reap_stale_free_video_tasks(*, team) -> None:
|
||
"""僵尸回收(趁每次新提交顺手做,无需定时任务):
|
||
· RESERVED 超 10 分钟:没提交到火山就死(worker 崩溃/进程重启)→ 标失败退费;
|
||
· SUBMITTED/POLLING 超 2 小时:轮询链早已断且无人认领(正常出片 5-10 分钟)→ 标失败退费;
|
||
· POSTPROCESSING 超 30 分钟:转存/结算中途崩溃 → 标失败退费(火山可能已出片,平台承担该笔成本)。"""
|
||
now = timezone.now()
|
||
buckets = [
|
||
([AITask.Status.RESERVED], now - timedelta(minutes=10), "任务未在预期时间内提交(自动回收)"),
|
||
([AITask.Status.SUBMITTED, AITask.Status.POLLING], now - timedelta(hours=2), "生成超时(自动回收)"),
|
||
([AITask.Status.POSTPROCESSING], now - timedelta(minutes=30), "视频结果处理超时(自动回收)"),
|
||
]
|
||
for statuses, cutoff, reason in buckets:
|
||
stale = AITask.objects.filter(
|
||
team=team,
|
||
project__isnull=True,
|
||
task_type=AITask.Type.FREE_VIDEO,
|
||
status__in=statuses,
|
||
updated_at__lt=cutoff,
|
||
)
|
||
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()
|
||
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("分辨率无效")
|
||
if not 4 <= duration <= 15:
|
||
raise ValueError("视频时长需在 4-15 秒之间")
|
||
if resolution in ("1080p", "4k") and model_name != HIGH_RES_MODEL:
|
||
raise ValueError(f"{resolution} 仅标准档模型支持，请切换模型或降低分辨率")
|
||
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("视频模型未配置，请联系管理员")
|
||
|
||
_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)
|
||
|
||
# 统一计价引擎:¥成本 × 毛利系数 → 积分;预留 = 积分 × 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)
|
||
|
||
request_payload = {
|
||
"feature": "free_video",
|
||
"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,
|
||
# 团队价格系数快照:按实结算用它,中途改价不影响在途任务
|
||
"price_multiplier": quote.meta.get("price_multiplier", "1"),
|
||
"points_per_yuan_snapshot": quote.meta.get("rate", ""),
|
||
"references": built["snapshots"],
|
||
}
|
||
|
||
# 建任务 + 预留同一事务:余额不足/限额拦截时回滚任务行,不留半套
|
||
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=quote.base_cost_yuan,
|
||
)
|
||
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"])
|
||
|
||
# 火山调用在事务外(不持锁调外网)
|
||
try:
|
||
from .services import build_provider
|
||
|
||
provider = build_provider(model_config)
|
||
response = provider.create_video_task(
|
||
model=model_config.name,
|
||
endpoint=model_config.endpoint,
|
||
prompt=built["api_prompt"],
|
||
ratio=aspect_ratio,
|
||
duration=duration,
|
||
resolution=resolution,
|
||
generate_audio=generate_audio,
|
||
content_items=built["content_items"],
|
||
seed=seed if seed != -1 else None,
|
||
search_mode=search_mode,
|
||
)
|
||
task.provider_task_id = str(response.get("id") or response.get("task_id") or "")
|
||
task.response_payload = response
|
||
task.status = AITask.Status.SUBMITTED
|
||
task.submitted_at = timezone.now()
|
||
task.save(update_fields=["provider_task_id", "response_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),
|
||
)
|
||
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
|
||
|
||
# worker 兜底轮询(自重排);派发失败仅 log,前端主动 poll 仍能收尾
|
||
try:
|
||
from .tasks import poll_free_video_task
|
||
|
||
poll_free_video_task.apply_async(args=[str(task.id), 0], countdown=30)
|
||
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="自由创作视频",
|
||
raw=raw,
|
||
hint=hint,
|
||
)
|
||
|
||
|
||
def _store_free_video_media(*, task: AITask, media: str) -> Asset:
|
||
"""下载火山结果 → 转存 TOS(火山原始 URL 仅 7 天有效)→ 建 Asset(FREE_CREATE,自动入库)
|
||
+ ffmpeg 抽首帧封面挂同 Asset 非主文件。"""
|
||
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""
|
||
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)
|
||
prompt = (task.request_payload or {}).get("prompt") or ""
|
||
asset = Asset.objects.create(
|
||
id=asset_id,
|
||
team=task.team,
|
||
created_by=task.created_by,
|
||
# Asset.name 是通用资产字段，受 255 字符上限约束；完整显示标题由资产接口从
|
||
# 关联任务的 request_payload.prompt 派生，避免再出现 50 字符业务截断。
|
||
name=(prompt[:255] or "自由创作视频"),
|
||
asset_type=Asset.Type.VIDEO,
|
||
source=Asset.Source.AI_GENERATED,
|
||
category=Asset.Category.FREE_CREATE,
|
||
origin_task=task,
|
||
metadata={"feature": "free_video"},
|
||
)
|
||
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 .services import build_provider
|
||
|
||
provider = build_provider(task.model_config)
|
||
response = provider.poll_video_task(
|
||
endpoint=task.model_config.endpoint, provider_task_id=task.provider_task_id
|
||
)
|
||
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)
|
||
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
|
||
with_video_ref = any((r or {}).get("type") == "video" for r in payload.get("references") or [])
|
||
resolution = payload.get("resolution") or "720p"
|
||
if total_tokens > 0:
|
||
from decimal import Decimal
|
||
|
||
settle = quote_video_actual(
|
||
locked.model_config, tokens=total_tokens, with_video_ref=with_video_ref, resolution=resolution,
|
||
multiplier=Decimal(str(payload.get("price_multiplier") or "1")),
|
||
)
|
||
actual, base_cost = settle.points, settle.base_cost_yuan
|
||
payload["actual_tokens"] = total_tokens
|
||
if settle.meta.get("rate"):
|
||
payload["points_per_yuan_snapshot"] = settle.meta["rate"]
|
||
else:
|
||
actual, base_cost = locked.estimated_cost, locked.base_cost
|
||
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)
|
||
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)
|
||
return locked
|
||
|
||
|
||
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 "",
|
||
"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": payload.get("references") or [],
|
||
"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,
|
||
}
|