azmat1127/yingqing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

极速成片页面和提炼提示词

Browse Source
This commit is contained in:
Azmat@qq.com
2026-08-24 15:52:05 +08:00
parent 2f5b7304b0
commit 00fc454db7
29 changed files with 1424 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/video_digest.py
+194 -53
View File
@@ -1,16 +1,10 @@
"""上传视频提炼 —— 参考视频 → 可人工逐镜编辑的中文分镜稿。
链路:ffmpeg 抽帧(均匀采样) → 帧内联进多模态 messages → 走现有文本模型路由 → 纯文本分镜稿。
优先把**完整视频(含音轨)**内联给 Gemini 3.1 Pro,跟官网「直接上传视频」同一条能力:
模型能看全片、听口播,不会只拿到稀疏静帧。文件太大塞不进请求时,才退回 ffmpeg 抽帧。
两个「本来以为要新建、其实已经有」的前提:
1. **读图能力**:拆视频是多模态(帧图 + 文本)。默认语言模型现在可能是纯文本豆包,
**不能再用 get_default_model(TEXT)**。固定钉 Gemini 3.1 Pro 官转
(``gemini-3.1-pro-preview``,展示名带「官转」优先)。
2. **ffmpeg**:第 5 阶段导出早就依赖它,已装进后端镜像(见 Dockerfile)。
**没有音轨**:帧里读不到口播,只能读画面上的字幕。这是刻意取舍——接语音转写要另开火山 ASR 服务、
另加一套凭证与计价,而带货参考片绝大多数带硬字幕,且口播词下游本来就要按用户自己的商品重写。
skill 里已要求模型「无字幕就如实写缺失,不许编口播词」。
拆视频必须会看视频/图。默认语言模型现在可能是纯文本豆包,不能用 get_default_model(TEXT)。
固定钉 Gemini 3.1 Pro 官转(``gemini-3.1-pro-preview``,展示名带「官转」优先)。
"""
from __future__ import annotations
@@ -19,6 +13,7 @@ import base64
import json
import logging
import math
import re
import shutil
import subprocess
import tempfile
@@ -36,12 +31,23 @@ ALLOWED_SUFFIXES = (".mp4", ".mov", ".m4v", ".webm")
MAX_UPLOAD_BYTES = 200 * 1024 * 1024 # 200 MB
MAX_DURATION_SECONDS = 180 # 3 分钟。带货参考片远短于此;更长的帧采样密度不够,拆出来也是错的
# 抽帧:每 5 秒一帧,夹在 4~12 帧之间。12 帧 × 512px JPEG ≈ 0.5 MB base64,单次请求扛得住
SECONDS_PER_FRAME = 5
MIN_FRAMES = 4
MAX_FRAMES = 12
FRAME_WIDTH = 512 # 帧宽上限;分镜拆解看的是构图与景别,不需要原分辨率
FRAME_QUALITY = 5 # ffmpeg -q:v,2(最好)~31(最差)
# 官网直接传视频的上限大约是请求 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 = 8192
_SHOT_MARK = re.compile(r"【第\s*\d+\s*镜】")
_FFMPEG_TIMEOUT = 60
@@ -59,6 +65,15 @@ class VideoFrame:
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 加载
# --------------------------------------------------------------------------- #
@@ -78,9 +93,9 @@ def load_digest_skill() -> str:
return main.read_text(encoding="utf-8")
# 兜底:skill 丢了也别整条链路挂掉,退化成一句话提示词(产出会明显变差,交接文档已注明须带 skills 目录)
return (
"你是分镜拆解 agent。输入是一条电商短视频按时间均匀抽出的截帧。"
"逐镜还原分镜,每镜写清主体/动作/场景/景别/运镜/光线氛围/商品露出七要素,"
"台词只抄画面上的字幕,看不见的不要编。输出中文纯文本。"
"你是分镜拆解 agent。输入是一条电商短视频的完整文件,含画面和口播。"
"必须覆盖全片,从 0 秒写到片尾,有几镜写几镜,不要概括成几大段。"
"每镜写画面和解说词,解说词按听到的口播逐字写。输出中文纯文本。"
)
@@ -145,61 +160,169 @@ def extract_frames(path: str | Path, times: list[int]) -> list[VideoFrame]:
return frames
def frames_from_upload(upload) -> tuple[list[VideoFrame], float]:
"""校验上传文件 → 落临时盘 → 探时长 → 抽帧。临时文件退出即删。"""
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"
with tempfile.NamedTemporaryFile(suffix=suffix) as tmp:
tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
try:
for chunk in upload.chunks():
tmp.write(chunk)
tmp.flush()
duration = probe_duration(tmp.name)
if duration > MAX_DURATION_SECONDS:
raise VideoDigestError(
f"视频不能超过 {MAX_DURATION_SECONDS // 60} 分钟,请剪出要参考的那一段再传"
)
return extract_frames(tmp.name, plan_frame_times(duration)), duration
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"视频不能超过 {MAX_DURATION_SECONDS // 60} 分钟,请剪出要参考的那一段再传"
)
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 _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) -> tuple[DigestVideo | None, list[VideoFrame], float]:
"""优先整段视频(含音轨);塞不进请求才抽帧。"""
path = ""
try:
path, suffix, size, duration = _materialize_upload(upload)
video = _native_video(path, size, suffix)
if video is not None:
return video, [], duration
return None, extract_frames(path, plan_frame_times(duration)), duration
finally:
if path:
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],
duration: float,
frames: list[VideoFrame] | None = None,
duration: float = 0,
*,
product_hint: str = "",
video: DigestVideo | None = None,
) -> list[dict]:
"""system = 拆解 skill;user = 时间戳 + 帧图交替,让模型知道每张图在原片的第几秒。"""
head = [
f"这是一条时长约 {round(duration)} 秒的电商带货短视频,",
f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。",
]
"""system = 拆解 skill;user = 完整视频(优先)或抽帧。"""
frames = frames or []
if video is not None:
head = [
f"这是一条时长约 {round(duration)} 秒的电商带货短视频的**完整文件**(含画面和口播音轨)。",
"请按技能还原全片分镜:从 0 秒写到片尾,有几镜写几镜。",
"解说词按听到的口播逐字写;画面上的花字一并写进画面。",
]
else:
head = [
f"这是一条时长约 {round(duration)} 秒的电商带货短视频,",
f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。",
"请按技能还原**全片**分镜:从 0 秒写到片尾,有几镜写几镜。",
]
if product_hint:
head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")
head.append("请按技能里的输出格式还原它的分镜。")
content: list[dict] = [{"type": "text", "text": "".join(head)}]
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()}})
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 validate_digest_text(text: str) -> str:
"""模型偶尔吐空 / 吐一句道歉。判空后交给路由层重试或切模型,别把废稿塞给用户。"""
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
@@ -295,7 +418,7 @@ def digest_project_video(*, project, user, upload, model_config_id=None) -> dict
def digest_team_video(*, team, user, upload, model_config_id=None) -> dict:
"""视频复刻页:不绑项目,计费挂当前团队。"""
"""视频提炼页:不绑项目,计费挂当前团队。"""
return _digest_video(
team=team,
user=user,
@@ -313,31 +436,43 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
from apps.ai.models import AITask
from apps.ai.services import create_ai_task, execute_routed_text_request
from apps.billing.pricing import quote_flat
from apps.billing.pricing import quote_video_digest
from apps.billing.services.ledger import charge_reserved_credit, reserve_credit
frames, duration = frames_from_upload(upload)
video, frames, duration = digest_input_from_upload(upload)
model_config = resolve_digest_model_config(preferred_id=model_config_id)
if model_config is None:
raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")
logger.info(
"video digest using %s:%s (%s)",
"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)
messages = build_digest_messages(
frames, duration, product_hint=product_hint, video=video
)
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],
}
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 project is not None:
task = create_ai_task(
project=project,
@@ -346,11 +481,9 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
model_config=model_config,
# 帧是几百 KB base64,绝不进 request_payload(会把 AITask 表撑爆),只记形状
request_payload=request_payload,
quote=quote,
)
else:
quote = quote_flat(model_config, team=team)
if quote.meta.get("rate"):
request_payload = {**request_payload, "points_per_yuan_snapshot": quote.meta["rate"]}
try:
with transaction.atomic():
task = AITask.objects.create(
@@ -385,8 +518,15 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
structured_output=False,
business_operation="video_digest",
temperature=0.4,
validate_text=validate_digest_text,
request_summary={"duration_seconds": round(duration, 2), "frame_count": len(frames)},
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,
)
@@ -397,7 +537,7 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
with transaction.atomic():
task.status = AITask.Status.SUCCEEDED
task.response_payload = {"digest": digest[:8000]}
task.response_payload = {"digest": digest[:32000]}
task.actual_cost = task.estimated_cost
task.completed_at = timezone.now()
task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
@@ -407,6 +547,7 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
"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),