1545 lines
60 KiB
Python
1545 lines
60 KiB
Python
"""上传视频提炼 —— 参考视频 → 可人工逐镜编辑的中文分镜稿。
|
|
|
|
优先把**完整视频(含音轨)**内联给 Gemini 3.1 Pro,跟官网「直接上传视频」同一条能力:
|
|
模型能看全片、听口播,不会只拿到稀疏静帧。文件太大塞不进请求时,才退回 ffmpeg 抽帧。
|
|
|
|
拆视频必须会看视频/图。默认语言模型现在可能是纯文本豆包,不能用 get_default_model(TEXT)。
|
|
固定钉 Gemini 3.1 Pro 官转(``gemini-3.1-pro-preview``,展示名带「官转」优先)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import math
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from django.conf import settings
|
|
|
|
# 上传限制:超了直接 400,不进 ffmpeg,也不花模型钱
|
|
ALLOWED_SUFFIXES = (".mp4", ".mov", ".m4v", ".webm")
|
|
MAX_UPLOAD_BYTES = 200 * 1024 * 1024 # 200 MB
|
|
# 30 秒。复刻走 Seedance 2.5,单次出片上限就是 30 秒 —— 参考片再长也用不上,
|
|
# 而且更长的帧采样密度不够,拆出来也是错的。半秒容差同 media_probe 的老规矩。
|
|
MAX_DURATION_SECONDS = 30.5
|
|
|
|
# 官网直接传视频的上限大约是请求 20MB;base64 会胀到 4/3,所以原文件卡在 15MB。
|
|
INLINE_VIDEO_MAX_BYTES = 15 * 1024 * 1024
|
|
_SUFFIX_MIME = {
|
|
".mp4": "video/mp4",
|
|
".m4v": "video/mp4",
|
|
".mov": "video/quicktime",
|
|
".webm": "video/webm",
|
|
}
|
|
|
|
# 抽帧只在整段视频塞不进请求时启用。
|
|
SECONDS_PER_FRAME = 2
|
|
MIN_FRAMES = 8
|
|
MAX_FRAMES = 36
|
|
FRAME_WIDTH = 768
|
|
FRAME_QUALITY = 3
|
|
DIGEST_MAX_TOKENS = 12288
|
|
# 视频复刻内联拆解的同模型重试次数(Gemini 偶尔吐废稿;复刻失败要用户从头重选素材,值得多试一次)
|
|
DIGEST_MAX_ATTEMPTS = 2
|
|
_SHOT_MARK = re.compile(r"【(?:镜头\s*\d+|第\s*\d+\s*镜)】")
|
|
|
|
_FFMPEG_TIMEOUT = 60
|
|
|
|
|
|
class VideoDigestError(ValueError):
|
|
"""用户可见的失败(文件不合格 / ffmpeg 读不动),一律 400。"""
|
|
|
|
|
|
class VideoDigestInProgress(VideoDigestError):
|
|
"""本团队已有提炼在跑。视图转 409,并把在跑的那条带回去让前端接上。"""
|
|
|
|
def __init__(self, job: dict):
|
|
super().__init__("已有一个视频正在提炼中，完成或取消后才能再提交")
|
|
self.job = job
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VideoFrame:
|
|
at_seconds: int
|
|
jpeg: bytes
|
|
|
|
def as_data_url(self) -> str:
|
|
return "data:image/jpeg;base64," + base64.b64encode(self.jpeg).decode("ascii")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DigestVideo:
|
|
mime: str
|
|
data: bytes
|
|
|
|
def as_data_url(self) -> str:
|
|
return f"data:{self.mime};base64," + base64.b64encode(self.data).decode("ascii")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# skill 加载
|
|
# --------------------------------------------------------------------------- #
|
|
def _skill_dir() -> Path:
|
|
"""与 script_agent._skill_dir 同源:优先 BASE_DIR/skills(镜像内),回落仓库根(本地旧布局)。"""
|
|
base = Path(settings.BASE_DIR)
|
|
for cand in (base / "skills", base.parent.parent / "skills"):
|
|
if (cand / "video-shot-digest").is_dir():
|
|
return cand / "video-shot-digest"
|
|
return base / "skills" / "video-shot-digest"
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_digest_skill() -> str:
|
|
main = _skill_dir() / "SKILL.md"
|
|
if main.exists():
|
|
return main.read_text(encoding="utf-8")
|
|
# 兜底:skill 丢了也别整条链路挂掉,退化成一句话提示词(产出会明显变差,交接文档已注明须带 skills 目录)
|
|
return (
|
|
"你是分镜拆解 agent。输入是一条短视频的完整文件,含画面和音轨。"
|
|
"必须覆盖全片,从 00:00 写到片尾,有几镜写几镜,不要概括成几大段。"
|
|
"每镜按【镜头 01】写出时间、时长、景别、机位、运镜、画面、人物动作、人物表情、"
|
|
"人声方式、台词/旁白、音效、背景音乐、字幕、备注。人声方式只能写画内人物对白、画外旁白或无；没有就写无。输出中文纯文本。"
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# ffmpeg:探时长 + 抽帧
|
|
# --------------------------------------------------------------------------- #
|
|
def _binary(name: str) -> str:
|
|
found = shutil.which(name)
|
|
if not found:
|
|
raise VideoDigestError("服务器暂时无法解析视频,请稍后再试")
|
|
return found
|
|
|
|
|
|
def probe_duration(path: str | Path) -> float:
|
|
"""ffprobe 读时长(秒)。读不到 = 不是能解的视频。"""
|
|
try:
|
|
out = subprocess.run(
|
|
[
|
|
_binary("ffprobe"), "-v", "error",
|
|
"-print_format", "json", "-show_format",
|
|
str(path),
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
|
|
).stdout
|
|
duration = float(json.loads(out)["format"]["duration"])
|
|
except VideoDigestError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 — ffprobe 各种失败对用户是同一件事
|
|
raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件") from exc
|
|
if duration <= 0:
|
|
raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件")
|
|
return duration
|
|
|
|
|
|
def probe_video_size(path: str | Path) -> tuple[int, int]:
|
|
"""ffprobe 读画面宽高。读不到返回 0,0,不挡拆解。"""
|
|
try:
|
|
out = subprocess.run(
|
|
[
|
|
_binary("ffprobe"), "-v", "error",
|
|
"-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height",
|
|
"-of", "csv=p=0:s=x",
|
|
str(path),
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
|
|
).stdout.decode("utf-8", errors="replace").strip()
|
|
width_s, height_s = out.split("x", 1)
|
|
return max(0, int(width_s)), max(0, int(height_s))
|
|
except Exception: # noqa: BLE001
|
|
return 0, 0
|
|
|
|
|
|
def extract_cover_jpeg(path: str | Path, duration: float) -> bytes:
|
|
"""抽一帧作历史封面。失败返回空字节,不挡拆解。"""
|
|
try:
|
|
ffmpeg = _binary("ffmpeg")
|
|
except VideoDigestError:
|
|
return b""
|
|
at = max(0.0, min(float(duration) * 0.35, max(0.0, float(duration) - 0.15)))
|
|
try:
|
|
done = subprocess.run(
|
|
[
|
|
ffmpeg, "-v", "error", "-ss", f"{at:.2f}", "-i", str(path),
|
|
"-frames:v", "1", "-vf", "scale=640:-2",
|
|
"-q:v", "4", "-f", "image2", "-",
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
return b""
|
|
return done.stdout or b""
|
|
|
|
|
|
def ratio_label(width: int, height: int) -> str:
|
|
if not width or not height:
|
|
return ""
|
|
ratio = width / height
|
|
if abs(ratio - 9 / 16) < 0.08:
|
|
return "9:16 竖屏"
|
|
if abs(ratio - 16 / 9) < 0.08:
|
|
return "16:9 横屏"
|
|
if abs(ratio - 1) < 0.08:
|
|
return "1:1"
|
|
return "横屏" if width > height else "竖屏"
|
|
|
|
|
|
def title_from_filename(name: str) -> str:
|
|
stem = Path(name or "").stem.strip()
|
|
return (stem or "参考视频")[:80]
|
|
|
|
|
|
def duration_clock(seconds: float) -> str:
|
|
total = max(0, int(round(float(seconds or 0))))
|
|
return f"{total // 60:02d}:{total % 60:02d}"
|
|
|
|
|
|
def shot_count(text: str) -> int:
|
|
return len(_SHOT_MARK.findall(text or ""))
|
|
|
|
|
|
def plan_frame_times(duration: float) -> list[int]:
|
|
"""均匀采样时间点。取每段的**中点**,避开首尾黑场与片尾卡片。"""
|
|
count = max(MIN_FRAMES, min(MAX_FRAMES, math.ceil(duration / SECONDS_PER_FRAME)))
|
|
step = duration / count
|
|
return [int(step * (i + 0.5)) for i in range(count)]
|
|
|
|
|
|
def extract_frames(path: str | Path, times: list[int]) -> list[VideoFrame]:
|
|
"""逐时间点抽一帧。``-ss`` 放在 ``-i`` 前走关键帧快速定位,每帧约几十毫秒。"""
|
|
ffmpeg = _binary("ffmpeg")
|
|
frames: list[VideoFrame] = []
|
|
for at in times:
|
|
try:
|
|
done = subprocess.run(
|
|
[
|
|
ffmpeg, "-v", "error", "-ss", str(at), "-i", str(path),
|
|
"-frames:v", "1", "-vf", f"scale={FRAME_WIDTH}:-2",
|
|
"-q:v", str(FRAME_QUALITY), "-f", "image2", "-",
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
|
|
)
|
|
except Exception: # noqa: BLE001 — 单帧抽失败(定位越界等)跳过,别拖垮整次提炼
|
|
continue
|
|
if done.stdout:
|
|
frames.append(VideoFrame(at_seconds=at, jpeg=done.stdout))
|
|
if not frames:
|
|
raise VideoDigestError("没能从这个视频里取到画面,请换一个文件")
|
|
return frames
|
|
|
|
|
|
def _write_upload(upload) -> tuple[str, str, int]:
|
|
"""校验后缀和体积,把上传落到临时文件。返回 (path, suffix, size)。调用方负责删除。"""
|
|
name = (getattr(upload, "name", "") or "").lower()
|
|
if not name.endswith(ALLOWED_SUFFIXES):
|
|
raise VideoDigestError("只支持 mp4 / mov / m4v / webm 四种视频格式")
|
|
size = getattr(upload, "size", 0) or 0
|
|
if size > MAX_UPLOAD_BYTES:
|
|
raise VideoDigestError(f"视频不能超过 {MAX_UPLOAD_BYTES // 1024 // 1024} MB,请压缩后再传")
|
|
suffix = Path(name).suffix or ".mp4"
|
|
tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
|
|
try:
|
|
for chunk in upload.chunks():
|
|
tmp.write(chunk)
|
|
tmp.flush()
|
|
finally:
|
|
tmp.close()
|
|
return tmp.name, suffix, Path(tmp.name).stat().st_size
|
|
|
|
|
|
def _materialize_upload(upload) -> tuple[str, str, int, float]:
|
|
"""校验 → 落盘 → 探时长。返回 (path, suffix, size, duration),调用方负责删文件。"""
|
|
path, suffix, size = _write_upload(upload)
|
|
try:
|
|
duration = probe_duration(path)
|
|
except Exception:
|
|
Path(path).unlink(missing_ok=True)
|
|
raise
|
|
if duration > MAX_DURATION_SECONDS:
|
|
Path(path).unlink(missing_ok=True)
|
|
raise VideoDigestError(
|
|
f"视频不能超过 {int(MAX_DURATION_SECONDS)} 秒,请剪出要参考的那一段再传"
|
|
)
|
|
return path, suffix, size, duration
|
|
|
|
|
|
def _compress_video(path: str) -> bytes | None:
|
|
"""压到能内联的体积。失败返回 None,由调用方改抽帧。"""
|
|
ffmpeg = _binary("ffmpeg")
|
|
out = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
|
|
out.close()
|
|
try:
|
|
done = subprocess.run(
|
|
[
|
|
ffmpeg, "-v", "error", "-y", "-i", path,
|
|
"-c:v", "libx264", "-preset", "veryfast", "-crf", "28",
|
|
"-vf", "scale='min(1280,iw)':-2",
|
|
"-c:a", "aac", "-b:a", "64k",
|
|
"-movflags", "+faststart", out.name,
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT * 3,
|
|
)
|
|
if done.returncode != 0:
|
|
return None
|
|
data = Path(out.name).read_bytes()
|
|
if not data or len(data) > INLINE_VIDEO_MAX_BYTES:
|
|
return None
|
|
return data
|
|
except Exception: # noqa: BLE001 — 压缩失败就抽帧,别挡住提炼
|
|
return None
|
|
finally:
|
|
Path(out.name).unlink(missing_ok=True)
|
|
|
|
|
|
def _is_browser_playable(path: str) -> bool:
|
|
"""Chrome 播不了微信常见的 HEVC。H.264 + AAC/无音轨才直接存。"""
|
|
try:
|
|
out = subprocess.run(
|
|
[
|
|
_binary("ffprobe"), "-v", "error",
|
|
"-show_entries", "stream=codec_name,codec_type",
|
|
"-of", "json",
|
|
str(path),
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
|
|
)
|
|
streams = json.loads(out.stdout).get("streams") or []
|
|
except Exception: # noqa: BLE001
|
|
return False
|
|
video_ok = False
|
|
audio_ok = True
|
|
for stream in streams:
|
|
kind = stream.get("codec_type")
|
|
codec = str(stream.get("codec_name") or "").lower()
|
|
if kind == "video":
|
|
video_ok = codec in {"h264", "vp8", "vp9", "av1"}
|
|
elif kind == "audio":
|
|
audio_ok = codec in {"aac", "mp3", "opus", "vorbis"}
|
|
return video_ok and audio_ok
|
|
|
|
|
|
def _prepare_browser_video(path: str, suffix: str) -> tuple[str, str, bool]:
|
|
"""转成浏览器能播的 H.264 AAC。失败退回原片。第三项表示调用方要删临时文件。"""
|
|
ext = suffix if str(suffix).startswith(".") else f".{suffix or 'mp4'}"
|
|
if _is_browser_playable(path):
|
|
return path, ext, False
|
|
ffmpeg = _binary("ffmpeg")
|
|
out = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
|
|
out.close()
|
|
try:
|
|
done = subprocess.run(
|
|
[
|
|
ffmpeg, "-v", "error", "-y", "-i", path,
|
|
"-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
|
|
"-c:a", "aac", "-b:a", "128k",
|
|
"-movflags", "+faststart", out.name,
|
|
],
|
|
capture_output=True, timeout=_FFMPEG_TIMEOUT * 3,
|
|
)
|
|
if done.returncode != 0 or not Path(out.name).is_file() or Path(out.name).stat().st_size <= 0:
|
|
Path(out.name).unlink(missing_ok=True)
|
|
return path, ext, False
|
|
return out.name, ".mp4", True
|
|
except Exception: # noqa: BLE001 — 转码失败仍存原片,总比历史卡不能播好
|
|
Path(out.name).unlink(missing_ok=True)
|
|
return path, ext, False
|
|
|
|
|
|
def _native_video(path: str, size: int, suffix: str) -> DigestVideo | None:
|
|
mime = _SUFFIX_MIME.get(suffix.lower(), "video/mp4")
|
|
if size <= INLINE_VIDEO_MAX_BYTES:
|
|
return DigestVideo(mime=mime, data=Path(path).read_bytes())
|
|
compressed = _compress_video(path)
|
|
if compressed:
|
|
return DigestVideo(mime="video/mp4", data=compressed)
|
|
return None
|
|
|
|
|
|
def digest_input_from_upload(
|
|
upload,
|
|
*,
|
|
keep_source: bool = False,
|
|
) -> tuple[DigestVideo | None, list[VideoFrame], float, dict]:
|
|
"""优先整段视频(含音轨);塞不进请求才抽帧。顺带抽出历史卡要用的封面/宽高。
|
|
|
|
keep_source=True 时不删临时文件,调用方上传原片后再删。
|
|
"""
|
|
path = ""
|
|
keep = False
|
|
original_name = Path(getattr(upload, "name", "") or "参考视频.mp4").name or "参考视频.mp4"
|
|
try:
|
|
path, suffix, size, duration = _materialize_upload(upload)
|
|
width, height = probe_video_size(path)
|
|
cover_jpeg = extract_cover_jpeg(path, duration)
|
|
video = _native_video(path, size, suffix)
|
|
frames = [] if video is not None else extract_frames(path, plan_frame_times(duration))
|
|
extras = {
|
|
"file_name": original_name,
|
|
"file_size": size,
|
|
"width": width,
|
|
"height": height,
|
|
"cover_jpeg": cover_jpeg,
|
|
"suffix": suffix,
|
|
"source_path": path if keep_source else "",
|
|
}
|
|
keep = keep_source
|
|
return video, frames, duration, extras
|
|
finally:
|
|
if path and not keep:
|
|
Path(path).unlink(missing_ok=True)
|
|
|
|
|
|
def frames_from_upload(upload) -> tuple[list[VideoFrame], float]:
|
|
"""校验上传文件 → 落临时盘 → 探时长 → 抽帧。单测与抽帧兜底用。"""
|
|
path = ""
|
|
try:
|
|
path, _suffix, _size, duration = _materialize_upload(upload)
|
|
return extract_frames(path, plan_frame_times(duration)), duration
|
|
finally:
|
|
if path:
|
|
Path(path).unlink(missing_ok=True)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 组装多模态消息
|
|
# --------------------------------------------------------------------------- #
|
|
def build_digest_messages(
|
|
frames: list[VideoFrame] | None = None,
|
|
duration: float = 0,
|
|
*,
|
|
product_hint: str = "",
|
|
video: DigestVideo | None = None,
|
|
aspect_ratio: str = "",
|
|
file_title: str = "",
|
|
) -> list[dict]:
|
|
"""system = 拆解 skill;user = 完整视频(优先)或抽帧。"""
|
|
frames = frames or []
|
|
measured = f"实测时长约 {round(duration)} 秒"
|
|
if aspect_ratio:
|
|
measured += f",画面比例 {aspect_ratio}"
|
|
title_hint = f"文件名可参考片名:《{file_title}》。" if file_title else ""
|
|
if video is not None:
|
|
head = [
|
|
f"这是一条{measured}的短视频**完整文件**(含画面和口播音轨)。{title_hint}",
|
|
"请按技能还原全片导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
|
|
"每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、人声方式、台词/旁白、音效、背景音乐、字幕、备注。",
|
|
"人声方式必须区分画内人物对白、画外旁白或无；台词/旁白按听到的口播逐字写，画面中人物开口时必须标画内人物对白；画面上的花字写进字幕。",
|
|
]
|
|
else:
|
|
head = [
|
|
f"这是一条{measured}的短视频,",
|
|
f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。{title_hint}",
|
|
"请按技能还原**全片**导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
|
|
"每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、人声方式、台词/旁白、音效、背景音乐、字幕、备注。",
|
|
]
|
|
if product_hint:
|
|
head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")
|
|
|
|
content: list[dict] = [{"type": "text", "text": "".join(head)}]
|
|
if video is not None:
|
|
data_url = video.as_data_url()
|
|
# 官转(New-API)把 OpenAI image_url 的 data URI 转成 Gemini inline_data,
|
|
# mime 从 data:video/mp4 头读取,模型按整段视频+音轨理解,等同官网直接上传。
|
|
content.append({"type": "image_url", "image_url": {"url": data_url}})
|
|
else:
|
|
for frame in frames:
|
|
content.append({"type": "text", "text": f"[第 {frame.at_seconds} 秒]"})
|
|
content.append({"type": "image_url", "image_url": {"url": frame.as_data_url()}})
|
|
return [
|
|
{"role": "system", "content": load_digest_skill()},
|
|
{"role": "user", "content": content},
|
|
]
|
|
|
|
|
|
def min_digest_shots(duration: float, frame_count: int) -> int:
|
|
"""短片至少 3 镜;20 秒以上约每 6 秒一镜,且不超过抽到的帧数。"""
|
|
if duration < 20 and frame_count < 8:
|
|
return 3
|
|
by_time = max(4, int(duration // 6))
|
|
if frame_count:
|
|
return min(frame_count, by_time)
|
|
return by_time
|
|
|
|
|
|
def validate_digest_text(text: str, *, duration: float = 0, frame_count: int = 0) -> str:
|
|
"""模型偶尔吐空、只写片头、或概括成几大段。废稿不塞给用户。"""
|
|
cleaned = (text or "").strip()
|
|
if len(cleaned) < 80 or "【" not in cleaned:
|
|
raise ValueError("视频拆解结果不完整")
|
|
shots = _SHOT_MARK.findall(cleaned)
|
|
if not shots:
|
|
raise ValueError("视频拆解结果不完整")
|
|
if duration >= 20 or frame_count >= 8:
|
|
needed = min_digest_shots(duration, frame_count)
|
|
if len(shots) < needed:
|
|
raise ValueError("视频拆解镜头过少,请重试")
|
|
return cleaned
|
|
|
|
|
|
# 拆视频必须会看图。火山豆包直连读不了这组帧图,会在 ARK 上挂满 120s。
|
|
# 只认中转站的 Gemini 3.1 Pro;展示名带「官转」的优先。
|
|
DIGEST_VISION_MODEL_NAME = "gemini-3.1-pro-preview"
|
|
|
|
|
|
def _is_digest_vision_model(model) -> bool:
|
|
from apps.ai.services import OFFICIAL_DIRECT_PROVIDERS
|
|
|
|
provider_name = getattr(getattr(model, "provider", None), "name", "") or ""
|
|
if provider_name in OFFICIAL_DIRECT_PROVIDERS:
|
|
return False
|
|
blob = f"{model.name} {model.display_name}".lower()
|
|
return (
|
|
model.name == DIGEST_VISION_MODEL_NAME
|
|
or "gemini-3.1" in blob
|
|
or "gemini 3.1" in blob
|
|
)
|
|
|
|
|
|
def resolve_digest_model_config(preferred_id=None):
|
|
"""视频提炼用的多模态文本模型:Gemini 3.1 Pro 官转。找不到不回落默认语言模型。
|
|
|
|
前端可传 model_config_id(跟生成脚本同一套下拉)。只有会看图的 Gemini 3.1 才认,
|
|
选了豆包等纯文本模型仍钉回官转,避免拆帧直接失败。
|
|
"""
|
|
from apps.ai.models import ModelConfig
|
|
|
|
qs = (
|
|
ModelConfig.objects.select_related("provider")
|
|
.filter(
|
|
capability=ModelConfig.Capability.TEXT,
|
|
status=ModelConfig.Status.ACTIVE,
|
|
provider__status="active",
|
|
)
|
|
)
|
|
if preferred_id:
|
|
chosen = qs.filter(pk=preferred_id).first()
|
|
if chosen is not None and _is_digest_vision_model(chosen):
|
|
return chosen
|
|
|
|
def _blob(model) -> str:
|
|
return " ".join(
|
|
filter(
|
|
None,
|
|
[
|
|
model.name,
|
|
model.display_name,
|
|
getattr(model.provider, "name", ""),
|
|
getattr(model.provider, "display_name", ""),
|
|
],
|
|
)
|
|
)
|
|
|
|
ranked = []
|
|
for model in qs:
|
|
if not _is_digest_vision_model(model):
|
|
continue
|
|
blob = _blob(model)
|
|
# 官转 > 精确模型名 > 其它 Gemini 3.1
|
|
score = 0
|
|
if "官转" in blob:
|
|
score += 100
|
|
if model.name == DIGEST_VISION_MODEL_NAME:
|
|
score += 20
|
|
if getattr(model.provider, "name", "") == "yunqi_gemini":
|
|
score += 5
|
|
ranked.append((score, model.created_at, model))
|
|
if not ranked:
|
|
return None
|
|
ranked.sort(key=lambda item: (-item[0], item[1]))
|
|
return ranked[0][2]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 入口:一次真实的计费调用
|
|
# --------------------------------------------------------------------------- #
|
|
def digest_project_video(*, project, user, upload, model_config_id=None) -> dict:
|
|
"""上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
|
|
product = getattr(project, "product", None)
|
|
return _digest_video(
|
|
team=project.team,
|
|
user=user,
|
|
upload=upload,
|
|
project=project,
|
|
product_hint=" · ".join(
|
|
filter(None, [getattr(product, "title", ""), getattr(product, "category", "")])
|
|
),
|
|
model_config_id=model_config_id,
|
|
)
|
|
|
|
|
|
def digest_team_video(*, team, user, upload, model_config_id=None) -> dict:
|
|
"""视频提炼页:不绑项目,计费挂当前团队。"""
|
|
return _digest_video(
|
|
team=team,
|
|
user=user,
|
|
upload=upload,
|
|
project=None,
|
|
product_hint="",
|
|
model_config_id=model_config_id,
|
|
)
|
|
|
|
|
|
def _store_digest_cover(*, team, jpeg: bytes) -> tuple[str, str]:
|
|
"""封面传到 TOS。失败返回空,历史卡走占位底。"""
|
|
from io import BytesIO
|
|
|
|
from apps.assets.storage import TosStorage
|
|
|
|
if not jpeg:
|
|
return "", ""
|
|
key = f"teams/{team.id}/video-digest/{uuid.uuid4()}.jpg"
|
|
storage = TosStorage()
|
|
stored = storage.upload_fileobj(fileobj=BytesIO(jpeg), object_key=key, content_type="image/jpeg")
|
|
return stored.object_key, storage.public_url(object_key=stored.object_key)
|
|
|
|
|
|
def _store_digest_video(*, team, path: str, suffix: str) -> tuple[str, str]:
|
|
"""原片(优先转成 H.264)传到 TOS,历史封面点击才能播。失败返回空,不挡拆解。"""
|
|
from apps.assets.storage import TosStorage
|
|
|
|
if not path or not Path(path).is_file():
|
|
return "", ""
|
|
upload_path, ext, ephemeral = _prepare_browser_video(path, suffix)
|
|
try:
|
|
if ext.lower() not in ALLOWED_SUFFIXES:
|
|
ext = ".mp4"
|
|
mime = _SUFFIX_MIME.get(ext.lower(), "video/mp4")
|
|
key = f"teams/{team.id}/video-digest/{uuid.uuid4()}{ext.lower()}"
|
|
storage = TosStorage()
|
|
with Path(upload_path).open("rb") as fileobj:
|
|
stored = storage.upload_fileobj(fileobj=fileobj, object_key=key, content_type=mime)
|
|
return stored.object_key, storage.public_url(object_key=stored.object_key)
|
|
finally:
|
|
if ephemeral:
|
|
Path(upload_path).unlink(missing_ok=True)
|
|
|
|
|
|
def _media_url_from_payload(req: dict, url_key: str, object_key_name: str, *, signed: bool = False) -> str:
|
|
stored = str(req.get(url_key) or "").strip()
|
|
key = str(req.get(object_key_name) or "").strip()
|
|
if not key:
|
|
return stored
|
|
try:
|
|
from apps.assets.storage import TosStorage
|
|
|
|
storage = TosStorage()
|
|
if signed:
|
|
return storage.presigned_get_url(object_key=key, expires_in=6 * 3600)
|
|
return storage.public_url(object_key=key)
|
|
except Exception: # noqa: BLE001
|
|
return stored
|
|
|
|
|
|
def serialize_digest_history(task) -> dict:
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
|
|
req = task.request_payload or {}
|
|
resp = task.response_payload or {}
|
|
prompt = str(resp.get("prompt") or resp.get("digest") or "").strip()
|
|
duration = float(req.get("duration_seconds") or 0)
|
|
file_name = str(req.get("file_name") or "") or "参考视频.mp4"
|
|
title = str(req.get("title") or "").strip() or title_from_filename(file_name)
|
|
shots = int(req.get("shot_count") or 0) or shot_count(prompt)
|
|
created = timezone.localtime(task.created_at) if task.created_at else timezone.now()
|
|
return {
|
|
"id": str(task.id),
|
|
"title": title,
|
|
"status": "已完成" if task.status == AITask.Status.SUCCEEDED else "失败",
|
|
"duration": int(round(duration)),
|
|
"duration_label": duration_clock(duration),
|
|
"ratio": str(req.get("ratio") or ""),
|
|
"shots": shots,
|
|
"file_name": file_name,
|
|
"cover_url": _media_url_from_payload(req, "cover_url", "cover_key"),
|
|
"video_url": _media_url_from_payload(req, "video_url", "video_key", signed=True),
|
|
"prompt": prompt,
|
|
"created_at": task.created_at.isoformat() if task.created_at else "",
|
|
"created_date": created.strftime("%Y-%m-%d"),
|
|
}
|
|
|
|
|
|
def list_team_digest_history(*, team, limit: int = 50) -> list[dict]:
|
|
from apps.ai.models import AITask
|
|
|
|
qs = (
|
|
AITask.objects.filter(
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
status=AITask.Status.SUCCEEDED,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
)
|
|
.order_by("-created_at")[: max(1, min(int(limit), 50))]
|
|
)
|
|
return [serialize_digest_history(task) for task in qs]
|
|
|
|
|
|
def save_digest_prompt(*, team, task_id, prompt: str) -> dict | None:
|
|
from apps.ai.models import AITask
|
|
|
|
cleaned = (prompt or "").strip()
|
|
if not cleaned:
|
|
raise VideoDigestError("提示词不能为空")
|
|
task = AITask.objects.filter(
|
|
id=task_id,
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
).first()
|
|
if task is None:
|
|
return None
|
|
payload = dict(task.response_payload or {})
|
|
payload["prompt"] = cleaned[:32000]
|
|
task.response_payload = payload
|
|
task.save(update_fields=["response_payload", "updated_at"])
|
|
return serialize_digest_history(task)
|
|
|
|
|
|
class _FileFromPath:
|
|
def __init__(self, path: str, name: str):
|
|
self.name = name
|
|
self.size = Path(path).stat().st_size
|
|
self._path = path
|
|
|
|
def chunks(self, chunk_size=1024 * 1024):
|
|
with Path(self._path).open("rb") as handle:
|
|
while True:
|
|
data = handle.read(chunk_size)
|
|
if not data:
|
|
break
|
|
yield data
|
|
|
|
|
|
def _store_raw_source(*, team, path: str, suffix: str) -> tuple[str, str]:
|
|
ext = suffix if str(suffix).startswith(".") else f".{suffix or 'mp4'}"
|
|
mime = _SUFFIX_MIME.get(ext.lower(), "video/mp4")
|
|
key = f"teams/{team.id}/video-digest/{uuid.uuid4()}{ext.lower()}"
|
|
from apps.assets.storage import TosStorage
|
|
|
|
storage = TosStorage()
|
|
with Path(path).open("rb") as fileobj:
|
|
stored = storage.upload_fileobj(fileobj=fileobj, object_key=key, content_type=mime)
|
|
return stored.object_key, storage.public_url(object_key=stored.object_key)
|
|
|
|
|
|
def _download_source_to_temp(object_key: str, suffix: str, *, bucket: str = "") -> str:
|
|
from apps.assets.storage import TosStorage
|
|
|
|
ext = suffix if str(suffix).startswith(".") else f".{suffix or 'mp4'}"
|
|
storage = TosStorage()
|
|
body = storage.client.get_object(Bucket=bucket or storage.bucket, Key=object_key)["Body"].read()
|
|
tmp = tempfile.NamedTemporaryFile(suffix=ext.lower(), delete=False)
|
|
try:
|
|
tmp.write(body)
|
|
tmp.flush()
|
|
finally:
|
|
tmp.close()
|
|
return tmp.name
|
|
|
|
|
|
def serialize_digest_job(task) -> dict:
|
|
from apps.ai.generation_errors import public_error_for_task
|
|
from apps.ai.models import AITask
|
|
|
|
req = task.request_payload or {}
|
|
resp = task.response_payload or {}
|
|
prompt = str(resp.get("prompt") or resp.get("digest") or "").strip()
|
|
inflight = task.status in {
|
|
AITask.Status.CREATED,
|
|
AITask.Status.RESERVED,
|
|
AITask.Status.SUBMITTED,
|
|
AITask.Status.POLLING,
|
|
AITask.Status.POSTPROCESSING,
|
|
}
|
|
if inflight:
|
|
job_status = "processing"
|
|
elif task.status == AITask.Status.SUCCEEDED:
|
|
job_status = "succeeded"
|
|
elif task.status == AITask.Status.CANCELLED:
|
|
job_status = "cancelled"
|
|
else:
|
|
job_status = "failed"
|
|
duration = float(req.get("duration_seconds") or 0)
|
|
file_name = str(req.get("file_name") or "") or "参考视频.mp4"
|
|
public_error = public_error_for_task(task, operation="video_digest") if job_status == "failed" else None
|
|
video_url = _media_url_from_payload(req, "video_url", "video_key", signed=True) or _media_url_from_payload(
|
|
req, "source_url", "source_key", signed=True
|
|
)
|
|
return {
|
|
"id": str(task.id),
|
|
"task_id": str(task.id),
|
|
"status": job_status,
|
|
"text": prompt,
|
|
"prompt": prompt,
|
|
"chars": len(prompt),
|
|
"duration": round(duration, 1) if duration else 0,
|
|
"shots": int(req.get("shot_count") or 0) or shot_count(prompt),
|
|
"file_name": file_name,
|
|
"title": str(req.get("title") or "").strip() or title_from_filename(file_name),
|
|
"ratio": str(req.get("ratio") or ""),
|
|
"width": int(req.get("width") or 0),
|
|
"height": int(req.get("height") or 0),
|
|
"cover_url": _media_url_from_payload(req, "cover_url", "cover_key"),
|
|
"video_url": video_url,
|
|
"estimated_cost": str(task.estimated_cost),
|
|
"error_message": (public_error.fallback_message if public_error else "") or str(task.error_message or ""),
|
|
}
|
|
|
|
|
|
def get_team_digest_job(*, team, task_id) -> dict | None:
|
|
from apps.ai.models import AITask
|
|
|
|
task = AITask.objects.filter(
|
|
id=task_id,
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
).first()
|
|
if task is None:
|
|
return None
|
|
return serialize_digest_job(task)
|
|
|
|
|
|
_STALE_DIGEST = timedelta(minutes=15)
|
|
|
|
|
|
def _task_reservation(task):
|
|
try:
|
|
return task.credit_reservation
|
|
except Exception: # noqa: BLE001 — 无预留行
|
|
return None
|
|
|
|
|
|
def cancel_team_digest(*, team, task_id, reason: str = "用户取消") -> dict | None:
|
|
"""取消进行中的提炼:停 UI 轮询、退预留积分。worker 跑到一半会看见 CANCELLED 不再落成功。"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
from apps.billing.services.ledger import release_credit
|
|
|
|
with transaction.atomic():
|
|
task = (
|
|
AITask.objects.select_for_update()
|
|
.filter(
|
|
id=task_id,
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
)
|
|
.first()
|
|
)
|
|
if task is None:
|
|
return None
|
|
if task.status == AITask.Status.SUCCEEDED:
|
|
return serialize_digest_job(task)
|
|
if task.status in {AITask.Status.FAILED, AITask.Status.CANCELLED}:
|
|
return serialize_digest_job(task)
|
|
reservation = _task_reservation(task)
|
|
task.status = AITask.Status.CANCELLED
|
|
task.error_message = (reason or "用户取消")[:2000]
|
|
task.completed_at = timezone.now()
|
|
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
|
|
if reservation is not None:
|
|
try:
|
|
release_credit(reservation=reservation, reason=(reason or "用户取消")[:200])
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("cancel digest release_credit failed for %s", task_id, exc_info=True)
|
|
return serialize_digest_job(task)
|
|
|
|
|
|
def expire_stale_team_digests(*, team) -> None:
|
|
"""卡住超过 15 分钟的提炼自动取消并退费,避免进页永远「正在拆解」。"""
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
|
|
stale = AITask.objects.filter(
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
status__in={
|
|
AITask.Status.CREATED,
|
|
AITask.Status.RESERVED,
|
|
AITask.Status.SUBMITTED,
|
|
AITask.Status.POLLING,
|
|
AITask.Status.POSTPROCESSING,
|
|
},
|
|
created_at__lt=timezone.now() - _STALE_DIGEST,
|
|
)
|
|
for task in stale:
|
|
cancel_team_digest(team=team, task_id=task.id, reason="提炼超时已自动取消")
|
|
|
|
|
|
def get_inflight_team_digest(*, team) -> dict | None:
|
|
from apps.ai.models import AITask
|
|
|
|
task = (
|
|
AITask.objects.filter(
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
status__in={
|
|
AITask.Status.CREATED,
|
|
AITask.Status.RESERVED,
|
|
AITask.Status.SUBMITTED,
|
|
AITask.Status.POLLING,
|
|
AITask.Status.POSTPROCESSING,
|
|
},
|
|
)
|
|
.order_by("-created_at")
|
|
.first()
|
|
)
|
|
if task is None:
|
|
return None
|
|
return serialize_digest_job(task)
|
|
|
|
|
|
def submit_team_digest(*, team, user, upload=None, reuse_task_id=None, model_config_id=None) -> dict:
|
|
"""秒回任务 id。慢活(抽帧 + Gemini)交给 worker,离开页面也不中断。
|
|
|
|
取消后再点生成时,可带 reuse_task_id 复用 TOS 上已有的参考视频,不用重新上传。
|
|
"""
|
|
from django.db import transaction
|
|
|
|
from apps.ai.models import AITask
|
|
from apps.ai.tasks import run_video_digest_task
|
|
from apps.billing.pricing import quote_video_digest
|
|
from apps.billing.services.ledger import reserve_credit
|
|
|
|
# 单飞闸:一个团队同时只允许一条提炼在跑。
|
|
# 刷新页面时前端要异步拉状态,这个空档用户很容易以为「没任务」再传一次 —— 前端加载态
|
|
# 只是治标,真正兜底在这里。先跑过期回收,死任务不会把人永久锁住。
|
|
expire_stale_team_digests(team=team)
|
|
running = get_inflight_team_digest(team=team)
|
|
if running:
|
|
raise VideoDigestInProgress(running)
|
|
|
|
model_config = resolve_digest_model_config(preferred_id=model_config_id)
|
|
if model_config is None:
|
|
raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")
|
|
|
|
if upload is not None:
|
|
path, suffix, size, duration = _materialize_upload(upload)
|
|
file_name = Path(getattr(upload, "name", "") or "参考视频.mp4").name or "参考视频.mp4"
|
|
width = height = 0
|
|
source_key = source_url = cover_key = cover_url = ""
|
|
try:
|
|
width, height = probe_video_size(path)
|
|
cover_key, cover_url = _store_digest_cover(team=team, jpeg=extract_cover_jpeg(path, duration))
|
|
source_key, source_url = _store_raw_source(team=team, path=path, suffix=suffix)
|
|
finally:
|
|
Path(path).unlink(missing_ok=True)
|
|
if not source_key:
|
|
raise VideoDigestError("参考视频上传失败，请重试")
|
|
request_payload = {
|
|
"model": model_config.name,
|
|
"endpoint": model_config.endpoint,
|
|
"feature": "video_remix",
|
|
"duration_seconds": round(duration, 2),
|
|
"file_name": file_name,
|
|
"title": title_from_filename(file_name),
|
|
"file_size": size,
|
|
"width": width,
|
|
"height": height,
|
|
"ratio": ratio_label(width, height),
|
|
"suffix": suffix,
|
|
"source_key": source_key,
|
|
"source_url": source_url,
|
|
"cover_key": cover_key,
|
|
"cover_url": cover_url,
|
|
}
|
|
else:
|
|
request_payload = _payload_from_existing_source(
|
|
team=team,
|
|
reuse_task_id=reuse_task_id,
|
|
model_config=model_config,
|
|
)
|
|
|
|
quote = quote_video_digest(team=team, model_config=model_config)
|
|
if quote.meta.get("rate"):
|
|
request_payload = {**request_payload, "points_per_yuan_snapshot": quote.meta["rate"]}
|
|
|
|
try:
|
|
with transaction.atomic():
|
|
task = AITask.objects.create(
|
|
team=team,
|
|
created_by=user,
|
|
project=None,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
status=AITask.Status.CREATED,
|
|
model_config=model_config,
|
|
idempotency_key=f"video_digest:{team.id}:{uuid.uuid4()}",
|
|
request_payload=request_payload,
|
|
estimated_cost=quote.points,
|
|
base_cost=quote.base_cost_yuan,
|
|
)
|
|
reserve_credit(team=team, user=user, task=task, amount=quote.points)
|
|
task.status = AITask.Status.RESERVED
|
|
task.save(update_fields=["status", "updated_at"])
|
|
except ValueError as exc:
|
|
if "insufficient credit" in str(exc).lower():
|
|
raise VideoDigestError("团队余额不足，请充值后重试") from exc
|
|
raise VideoDigestError(str(exc)) from exc
|
|
|
|
run_video_digest_task.delay(str(task.id))
|
|
return serialize_digest_job(task)
|
|
|
|
|
|
def _payload_from_existing_source(*, team, reuse_task_id, model_config) -> dict:
|
|
from django.core.exceptions import ValidationError
|
|
|
|
from apps.ai.models import AITask
|
|
|
|
task_id = str(reuse_task_id or "").strip()
|
|
if not task_id:
|
|
raise VideoDigestError("请先上传参考视频")
|
|
try:
|
|
previous = AITask.objects.filter(
|
|
id=task_id,
|
|
team=team,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
is_deleted=False,
|
|
purged_at__isnull=True,
|
|
).first()
|
|
except (ValueError, ValidationError, TypeError):
|
|
raise VideoDigestError("请先上传参考视频") from None
|
|
if previous is None:
|
|
raise VideoDigestError("请先上传参考视频")
|
|
req = previous.request_payload or {}
|
|
source_key = str(req.get("source_key") or "").strip()
|
|
if not source_key:
|
|
raise VideoDigestError("参考视频已失效，请重新上传")
|
|
file_name = str(req.get("file_name") or "") or "参考视频.mp4"
|
|
return {
|
|
"model": model_config.name,
|
|
"endpoint": model_config.endpoint,
|
|
"feature": "video_remix",
|
|
"duration_seconds": float(req.get("duration_seconds") or 0),
|
|
"file_name": file_name,
|
|
"title": str(req.get("title") or "").strip() or title_from_filename(file_name),
|
|
"file_size": int(req.get("file_size") or 0),
|
|
"width": int(req.get("width") or 0),
|
|
"height": int(req.get("height") or 0),
|
|
"ratio": str(req.get("ratio") or ""),
|
|
"suffix": str(req.get("suffix") or ".mp4"),
|
|
"source_key": source_key,
|
|
"source_url": str(req.get("source_url") or ""),
|
|
"cover_key": str(req.get("cover_key") or ""),
|
|
"cover_url": str(req.get("cover_url") or ""),
|
|
"video_key": str(req.get("video_key") or ""),
|
|
"video_url": str(req.get("video_url") or ""),
|
|
}
|
|
|
|
|
|
def run_team_digest_task(*, task_id: str) -> None:
|
|
"""Worker:从 TOS 取参考视频,跑抽帧 + Gemini,失败退费。"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
|
|
task = AITask.objects.select_related("team", "created_by", "model_config", "credit_reservation").filter(
|
|
id=task_id,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
project__isnull=True,
|
|
).first()
|
|
if task is None:
|
|
return
|
|
if task.status == AITask.Status.SUCCEEDED:
|
|
return
|
|
if task.status not in {AITask.Status.RESERVED, AITask.Status.SUBMITTED}:
|
|
return
|
|
|
|
with transaction.atomic():
|
|
locked = AITask.objects.select_for_update().get(id=task.id)
|
|
if locked.status == AITask.Status.SUCCEEDED:
|
|
return
|
|
if locked.status not in {AITask.Status.RESERVED, AITask.Status.SUBMITTED}:
|
|
return
|
|
locked.status = AITask.Status.SUBMITTED
|
|
locked.submitted_at = locked.submitted_at or timezone.now()
|
|
locked.save(update_fields=["status", "submitted_at", "updated_at"])
|
|
task = locked
|
|
|
|
req = task.request_payload or {}
|
|
source_key = str(req.get("source_key") or "")
|
|
suffix = str(req.get("suffix") or ".mp4")
|
|
file_name = str(req.get("file_name") or "参考视频.mp4")
|
|
if not source_key:
|
|
reservation = getattr(task, "credit_reservation", None)
|
|
_fail_digest_task(task, reservation, "参考视频丢失，请重新上传")
|
|
return
|
|
|
|
local_path = ""
|
|
try:
|
|
local_path = _download_source_to_temp(source_key, suffix)
|
|
upload = _FileFromPath(local_path, file_name)
|
|
_digest_video(
|
|
team=task.team,
|
|
user=task.created_by,
|
|
upload=upload,
|
|
project=None,
|
|
product_hint="",
|
|
model_config_id=str(task.model_config_id) if task.model_config_id else None,
|
|
existing_task=task,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("async video digest failed for %s", task.id)
|
|
task.refresh_from_db()
|
|
if task.status not in {AITask.Status.SUCCEEDED, AITask.Status.FAILED}:
|
|
reservation = getattr(task, "credit_reservation", None)
|
|
_fail_digest_task(task, reservation, str(exc))
|
|
finally:
|
|
if local_path:
|
|
Path(local_path).unlink(missing_ok=True)
|
|
|
|
|
|
def _digest_video(*, team, user, upload, project=None, product_hint="", model_config_id=None, existing_task=None) -> dict:
|
|
"""上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
from apps.ai.services import create_ai_task, execute_routed_text_request
|
|
from apps.billing.pricing import quote_video_digest
|
|
from apps.billing.services.ledger import charge_reserved_credit, reserve_credit
|
|
|
|
video, frames, duration, extras = digest_input_from_upload(
|
|
upload, keep_source=(project is None)
|
|
)
|
|
extras = extras or {}
|
|
source_path = str(extras.get("source_path") or "")
|
|
file_name = str(extras.get("file_name") or getattr(upload, "name", "") or "参考视频.mp4")
|
|
width = int(extras.get("width") or 0)
|
|
height = int(extras.get("height") or 0)
|
|
file_size = int(extras.get("file_size") or getattr(upload, "size", 0) or 0)
|
|
|
|
model_config = (
|
|
existing_task.model_config
|
|
if existing_task is not None and existing_task.model_config_id
|
|
else resolve_digest_model_config(preferred_id=model_config_id)
|
|
)
|
|
if model_config is None:
|
|
if source_path:
|
|
Path(source_path).unlink(missing_ok=True)
|
|
raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")
|
|
logger.info(
|
|
"video digest using %s:%s (%s) input=%s duration=%.1fs bytes=%s frames=%s",
|
|
model_config.provider.name,
|
|
model_config.name,
|
|
model_config.display_name,
|
|
"native_video" if video is not None else "frames",
|
|
duration,
|
|
len(video.data) if video is not None else 0,
|
|
len(frames),
|
|
)
|
|
|
|
messages = build_digest_messages(
|
|
frames,
|
|
duration,
|
|
product_hint=product_hint,
|
|
video=video,
|
|
aspect_ratio=ratio_label(width, height),
|
|
file_title=title_from_filename(file_name),
|
|
)
|
|
request_payload = {
|
|
"model": model_config.name,
|
|
"endpoint": model_config.endpoint,
|
|
"feature": "video_remix" if project is None else "video_digest",
|
|
"duration_seconds": round(duration, 2),
|
|
"input": "native_video" if video is not None else "frames",
|
|
"frame_count": len(frames),
|
|
"video_bytes": len(video.data) if video is not None else 0,
|
|
"frame_times": [f.at_seconds for f in frames],
|
|
"file_name": file_name,
|
|
"title": title_from_filename(file_name),
|
|
"file_size": file_size,
|
|
"width": width,
|
|
"height": height,
|
|
"ratio": ratio_label(width, height),
|
|
}
|
|
|
|
quote = quote_video_digest(team=team, model_config=model_config)
|
|
if quote.meta.get("rate"):
|
|
request_payload = {**request_payload, "points_per_yuan_snapshot": quote.meta["rate"]}
|
|
|
|
if existing_task is not None:
|
|
task = existing_task
|
|
existing_req = dict(task.request_payload or {})
|
|
request_payload = {**existing_req, **request_payload}
|
|
task.request_payload = request_payload
|
|
task.save(update_fields=["request_payload", "updated_at"])
|
|
elif project is not None:
|
|
task = create_ai_task(
|
|
project=project,
|
|
user=user,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
model_config=model_config,
|
|
# 帧是几百 KB base64,绝不进 request_payload(会把 AITask 表撑爆),只记形状
|
|
request_payload=request_payload,
|
|
quote=quote,
|
|
)
|
|
else:
|
|
try:
|
|
with transaction.atomic():
|
|
task = AITask.objects.create(
|
|
team=team,
|
|
created_by=user,
|
|
project=None,
|
|
task_type=AITask.Type.VIDEO_DIGEST,
|
|
status=AITask.Status.CREATED,
|
|
model_config=model_config,
|
|
idempotency_key=f"video_digest:{team.id}:{uuid.uuid4()}",
|
|
request_payload=request_payload,
|
|
estimated_cost=quote.points,
|
|
base_cost=quote.base_cost_yuan,
|
|
)
|
|
reserve_credit(team=team, user=user, task=task, amount=quote.points)
|
|
task.status = AITask.Status.RESERVED
|
|
task.save(update_fields=["status", "updated_at"])
|
|
except ValueError as exc:
|
|
if source_path:
|
|
Path(source_path).unlink(missing_ok=True)
|
|
if "insufficient credit" in str(exc).lower():
|
|
raise VideoDigestError("团队余额不足，请充值后重试") from exc
|
|
raise VideoDigestError(str(exc)) from exc
|
|
reservation = task.credit_reservation
|
|
try:
|
|
task.refresh_from_db()
|
|
if task.status == AITask.Status.CANCELLED:
|
|
if source_path:
|
|
Path(source_path).unlink(missing_ok=True)
|
|
return serialize_digest_job(task)
|
|
task.status = AITask.Status.SUBMITTED
|
|
task.submitted_at = timezone.now()
|
|
task.save(update_fields=["status", "submitted_at", "updated_at"])
|
|
routed = execute_routed_text_request(
|
|
task=task,
|
|
primary_model=model_config,
|
|
messages=messages,
|
|
streaming=True,
|
|
structured_output=False,
|
|
business_operation="video_digest",
|
|
temperature=0.4,
|
|
validate_text=lambda text: validate_digest_text(
|
|
text, duration=duration, frame_count=len(frames) or (24 if video else 0)
|
|
),
|
|
extra_body={"max_tokens": DIGEST_MAX_TOKENS},
|
|
request_summary={
|
|
"duration_seconds": round(duration, 2),
|
|
"frame_count": len(frames),
|
|
"input": "native_video" if video is not None else "frames",
|
|
},
|
|
allow_retry=False,
|
|
allow_fallback=False,
|
|
)
|
|
_text, _response, digest = routed.value
|
|
except Exception as exc: # noqa: BLE001
|
|
if source_path:
|
|
Path(source_path).unlink(missing_ok=True)
|
|
task.refresh_from_db()
|
|
if task.status == AITask.Status.CANCELLED:
|
|
return serialize_digest_job(task)
|
|
_fail_digest_task(task, reservation, str(exc))
|
|
raise
|
|
|
|
existing_req = dict(task.request_payload or {})
|
|
cover_key = str(existing_req.get("cover_key") or "")
|
|
cover_url = str(existing_req.get("cover_url") or "")
|
|
video_key = str(existing_req.get("video_key") or "")
|
|
video_url = str(existing_req.get("video_url") or "")
|
|
if project is None:
|
|
if not cover_key:
|
|
try:
|
|
cover_key, cover_url = _store_digest_cover(team=team, jpeg=extras.get("cover_jpeg") or b"")
|
|
except Exception: # noqa: BLE001 — 封面失败不挡拆解结果
|
|
logger.warning("video digest cover upload failed", exc_info=True)
|
|
try:
|
|
stored_key, stored_url = _store_digest_video(
|
|
team=team,
|
|
path=source_path,
|
|
suffix=str(extras.get("suffix") or ".mp4"),
|
|
)
|
|
if stored_key:
|
|
video_key, video_url = stored_key, stored_url
|
|
except Exception: # noqa: BLE001 — 原片失败仍可看提示词,封面不能播
|
|
logger.warning("video digest source upload failed", exc_info=True)
|
|
if source_path:
|
|
Path(source_path).unlink(missing_ok=True)
|
|
source_path = ""
|
|
|
|
shots = shot_count(digest)
|
|
request_payload = {
|
|
**existing_req,
|
|
**request_payload,
|
|
"shot_count": shots,
|
|
"cover_key": cover_key or existing_req.get("cover_key") or "",
|
|
"cover_url": cover_url or existing_req.get("cover_url") or "",
|
|
"video_key": video_key or existing_req.get("video_key") or "",
|
|
"video_url": video_url or existing_req.get("video_url") or "",
|
|
}
|
|
|
|
with transaction.atomic():
|
|
locked = AITask.objects.select_for_update().get(id=task.id)
|
|
if locked.status == AITask.Status.CANCELLED:
|
|
return serialize_digest_job(locked)
|
|
task = locked
|
|
task.status = AITask.Status.SUCCEEDED
|
|
task.request_payload = request_payload
|
|
task.response_payload = {"digest": digest[:32000], "prompt": digest[:32000]}
|
|
task.actual_cost = task.estimated_cost
|
|
task.completed_at = timezone.now()
|
|
task.save(
|
|
update_fields=[
|
|
"status",
|
|
"request_payload",
|
|
"response_payload",
|
|
"actual_cost",
|
|
"completed_at",
|
|
"updated_at",
|
|
]
|
|
)
|
|
charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
|
|
|
|
return {
|
|
"text": digest,
|
|
"chars": len(digest),
|
|
"frames": len(frames),
|
|
"input": "native_video" if video is not None else "frames",
|
|
"duration": round(duration, 1),
|
|
"task_id": str(task.id),
|
|
"estimated_cost": str(task.estimated_cost),
|
|
"title": request_payload.get("title") or title_from_filename(file_name),
|
|
"file_name": file_name,
|
|
"ratio": request_payload.get("ratio") or "",
|
|
"shots": shots,
|
|
"cover_url": cover_url,
|
|
"video_url": video_url,
|
|
"width": width,
|
|
"height": height,
|
|
}
|
|
|
|
|
|
def _fail_digest_task(task, reservation, message: str) -> None:
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AITask
|
|
from apps.billing.services.ledger import release_credit
|
|
|
|
try:
|
|
task.status = AITask.Status.FAILED
|
|
task.error_message = message[:2000]
|
|
task.completed_at = timezone.now()
|
|
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
|
|
finally:
|
|
try:
|
|
release_credit(reservation=reservation, reason=message[:200])
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 复用入口:视频复刻·商品 拿分镜稿
|
|
# --------------------------------------------------------------------------- #
|
|
def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]:
|
|
"""把一条已入库的参考视频拆成中文分镜稿,供「视频复刻·商品」当提示词骨架。
|
|
|
|
与「提炼提示词」页共用同一份 SKILL.md、同一个 Gemini 3.1 Pro 和同一套质检,
|
|
保证两处产出一致;区别只在于这里不另建 VIDEO_DIGEST 任务、也不单独扣积分——
|
|
复刻本来就是一次收费,拆解是它的内部工序。模型调用的审计仍记在传入的复刻
|
|
task 上(AIModelAttempt),出问题查得到。
|
|
"""
|
|
import time
|
|
|
|
from apps.ai.models import AIModelAttempt
|
|
from apps.ai.services import _collect_extract_text, get_text_provider
|
|
|
|
primary = asset.files.filter(is_primary=True).first() or asset.files.first()
|
|
if primary is None or not primary.object_key:
|
|
raise VideoDigestError("参考视频没有可用文件，请重新上传")
|
|
suffix = Path(primary.object_key).suffix.lower()
|
|
if suffix not in ALLOWED_SUFFIXES:
|
|
suffix = ".mp4"
|
|
|
|
model_config = resolve_digest_model_config(preferred_id=model_config_id)
|
|
if model_config is None:
|
|
raise VideoDigestError("视频复刻需要 Gemini 3.1 Pro(会看图)，当前没有启用，请联系管理员")
|
|
|
|
local_path = ""
|
|
try:
|
|
local_path = _download_source_to_temp(primary.object_key, suffix, bucket=primary.bucket or "")
|
|
base_name = (asset.name or "参考视频").rsplit(".", 1)[0]
|
|
upload = _FileFromPath(local_path, f"{base_name}{suffix}")
|
|
video, frames, duration, extras = digest_input_from_upload(upload)
|
|
extras = extras or {}
|
|
logger.info(
|
|
"replace digest using %s:%s input=%s duration=%.1fs frames=%s",
|
|
model_config.provider.name,
|
|
model_config.name,
|
|
"native_video" if video is not None else "frames",
|
|
duration,
|
|
len(frames),
|
|
)
|
|
# ⚠️ 这里必须和「提炼提示词」页(_digest_video)构造出完全相同的 messages:
|
|
# 同一份 SKILL.md 作 system、同一段 user 引导语、同样的 temperature / max_tokens。
|
|
# 绝对不要在这里追加商品信息(product_hint 保持默认空)——两处提炼稿一旦不同,
|
|
# 用户在提炼页看到的分镜和复刻实际用的分镜就对不上,排查会乱。
|
|
# test_video_replace.test_replace_digest_prompt_matches_standalone 会守住这一点。
|
|
messages = build_digest_messages(
|
|
frames,
|
|
duration,
|
|
video=video,
|
|
aspect_ratio=ratio_label(int(extras.get("width") or 0), int(extras.get("height") or 0)),
|
|
file_title=title_from_filename(str(extras.get("file_name") or "")),
|
|
)
|
|
# 这里不能走 execute_routed_text_request:它硬性要求 task.model_config 就是本次主模型,
|
|
# 而复刻任务的 model_config 是 Seedance(视频模型),传进去必抛
|
|
# 「AITask.model_config 必须保持为用户选择或系统默认的主模型」。
|
|
# 改为直连脚本/提炼同一条流式通道,并自己做「同模型重试」——绝不 fallback 到别的
|
|
# 文本模型,它们看不了视频,换过去必然废稿。
|
|
provider = get_text_provider(model_config)
|
|
expected_frames = len(frames) or (24 if video else 0)
|
|
digest = ""
|
|
last_error: Exception | None = None
|
|
for attempt_no in range(1, DIGEST_MAX_ATTEMPTS + 1):
|
|
# 这一步不走 execute_model_call(它要求 task.model_config 就是本次主模型,
|
|
# 而复刻任务的主模型是 Seedance),所以尝试记录得自己落 —— 不落的话
|
|
# 运营后台的任务详情看不到这次 Gemini 调用,出问题无从查起。
|
|
record = _open_digest_attempt(task, model_config, attempt_no, duration, expected_frames, video)
|
|
started = time.monotonic()
|
|
try:
|
|
text, _payload = _collect_extract_text(
|
|
provider,
|
|
model_config,
|
|
messages,
|
|
temperature=0.4,
|
|
extra_body={"max_tokens": DIGEST_MAX_TOKENS},
|
|
)
|
|
digest = validate_digest_text(text, duration=duration, frame_count=expected_frames)
|
|
_close_digest_attempt(
|
|
record, AIModelAttempt.Status.SUCCEEDED, started,
|
|
summary={"chars": len(digest), "shots": shot_count(digest)},
|
|
)
|
|
break
|
|
except Exception as exc: # noqa: BLE001 — 空文/废稿/网络抖动都值得再来一次
|
|
last_error = exc
|
|
_close_digest_attempt(
|
|
record, AIModelAttempt.Status.FAILED, started, error=str(exc),
|
|
)
|
|
logger.warning(
|
|
"replace digest attempt %s/%s failed for task %s: %s",
|
|
attempt_no, DIGEST_MAX_ATTEMPTS, task.id, exc,
|
|
)
|
|
if not digest:
|
|
raise VideoDigestError(f"参考视频拆解失败：{last_error}")
|
|
meta = {
|
|
"digest_model": model_config.name,
|
|
"digest_input": "native_video" if video is not None else "frames",
|
|
"digest_frames": len(frames),
|
|
"digest_duration": round(duration, 2),
|
|
"digest_shots": shot_count(digest),
|
|
"digest_ratio": ratio_label(int(extras.get("width") or 0), int(extras.get("height") or 0)),
|
|
}
|
|
return digest, meta
|
|
finally:
|
|
if local_path:
|
|
Path(local_path).unlink(missing_ok=True)
|
|
|
|
|
|
def _open_digest_attempt(task, model_config, attempt_no: int, duration: float, frames: int, video):
|
|
"""给复刻任务补一条「提炼」尝试记录,让运营后台的尝试链能看到这次 Gemini 调用。
|
|
|
|
审计失败绝不能拖垮拆解本身,所以整段吞异常、返回 None。
|
|
"""
|
|
from django.db import transaction
|
|
from django.db.models import Max
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.models import AIModelAttempt, AITask
|
|
|
|
try:
|
|
with transaction.atomic():
|
|
AITask.objects.select_for_update().only("id").get(pk=task.pk)
|
|
sequence = (
|
|
AIModelAttempt.objects.filter(task_id=task.pk).aggregate(m=Max("sequence"))["m"] or 0
|
|
) + 1
|
|
provider = model_config.provider
|
|
return AIModelAttempt.objects.create(
|
|
task_id=task.pk,
|
|
sequence=sequence,
|
|
provider=provider,
|
|
model_config=model_config,
|
|
provider_name=provider.name,
|
|
provider_display_name=provider.display_name,
|
|
model_name=model_config.name,
|
|
model_display_name=model_config.display_name,
|
|
public_model_name=model_config.display_name or model_config.name,
|
|
capability="text",
|
|
operation="video_digest",
|
|
status=AIModelAttempt.Status.STARTED,
|
|
is_retry=attempt_no > 1,
|
|
started_at=timezone.now(),
|
|
request_summary={
|
|
"for": "video_replace",
|
|
"duration_seconds": round(float(duration or 0), 2),
|
|
"frame_count": frames,
|
|
"input": "native_video" if video is not None else "frames",
|
|
},
|
|
)
|
|
except Exception: # noqa: BLE001 — 审计写失败不影响出片
|
|
logger.warning("video replace digest attempt record failed", exc_info=True)
|
|
return None
|
|
|
|
|
|
def _close_digest_attempt(record, status, started_at: float, *, summary: dict | None = None, error: str = ""):
|
|
import time
|
|
|
|
from django.utils import timezone
|
|
|
|
if record is None:
|
|
return
|
|
try:
|
|
record.status = status
|
|
record.finished_at = timezone.now()
|
|
record.duration_ms = max(0, round((time.monotonic() - started_at) * 1000))
|
|
if summary:
|
|
record.response_summary = summary
|
|
if error:
|
|
record.error_type = "processing_failed"
|
|
record.raw_error = error[:2000]
|
|
record.safe_error_summary = "参考视频拆解未通过校验"
|
|
record.save(update_fields=[
|
|
"status", "finished_at", "duration_ms", "response_summary",
|
|
"error_type", "raw_error", "safe_error_summary", "updated_at",
|
|
])
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("video replace digest attempt close failed", exc_info=True)
|