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-25 11:13:07 +08:00
parent 00fc454db7
commit e2ec2d14af
46 changed files with 4734 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/script_agent.py
+63 -23
View File
@@ -52,6 +52,8 @@ VIDEO_STRUCTURES: dict[str, str] = {
"contrast": "前后对比",
"review": "测评验证",
"scene": "场景种草",
"promo": "促销抢购",
"knowledge": "知识分享",
}
DEFAULT_PRESENTATION_FORMAT = "oral"
DEFAULT_VIDEO_STRUCTURE = "pain"
@@ -62,7 +64,9 @@ FORBIDDEN_COMBOS: set[tuple[str, str]] = {("drama", "review")}
# 表现形式推荐的默认总时长:口播短平快,短剧要装下三幕,Vlog 要铺氛围。
FORMAT_DEFAULT_DURATION: dict[str, int] = {"oral": 30, "drama": 45, "vlog": 30}
# 各结构能压到的最短总时长(须落在 15 秒步进上),见 playbooks/combo-matrix.md。
STRUCTURE_MIN_DURATION: dict[str, int] = {"pain": 15, "contrast": 15, "review": 30, "scene": 30}
STRUCTURE_MIN_DURATION: dict[str, int] = {
"pain": 15, "contrast": 15, "review": 30, "scene": 30, "promo": 15, "knowledge": 30,
}
# 设定卡人物 key → 中文(与前端 WIZ_PERSONA_LABEL / 模板 coerce_persona 对齐)。
PERSONA_LABELS: dict[str, str] = {
@@ -83,11 +87,11 @@ PERSONA_BRIEFS: dict[str, str] = {
}
_PERSONA_KEY_BY_LABEL = {label: key for key, label in PERSONA_LABELS.items()}
# 成片有同期画面与自然停顿，3.5 字/秒会让 15 秒口播在约 10 秒时说完。
# 电商真人口播以 3.6–4.5 字/秒为可用区间：既留出换气，也不会让画面空转。
NARRATION_CHARS_PER_SECOND = 4.5
NARRATION_CHARS_PER_SECOND_MIN = 3.6
NARRATION_CHARS_HARD_CAP = 68
# 电商口播需要同时承载观点、证据和转折；15 秒以 4.3–5.2 字/秒为可用区间。
# 这比日常闲聊快，但仍给句间停顿和镜头切换留出空间。
NARRATION_CHARS_PER_SECOND = 5.2
NARRATION_CHARS_PER_SECOND_MIN = 4.3
NARRATION_CHARS_HARD_CAP = 78
VISUAL_CHARS_MIN = 72
SHOT_BEATS_MIN = 3
BEAT_SPAN_RE = re.compile(
@@ -161,7 +165,7 @@ def coerce_combo(fmt: str | None, structure: str | None) -> tuple[str, str]:
def narration_limit(duration: int) -> int:
"""这一镜旁白的字数上限:秒数 × 4.5,且不超过硬上限 68。"""
"""这一镜旁白的字数上限:秒数 × 5.2,且不超过硬上限 78。"""
return max(1, min(NARRATION_CHARS_HARD_CAP, int(duration * NARRATION_CHARS_PER_SECOND)))
@@ -342,14 +346,15 @@ _CREATIVE_DIRECTION = """
每个 segment 的 `visual` 采用下面的层级；这不是给用户看的散文，而是会原样交给故事板和视频模型的执行指令：
```
【本镜任务】这一段要让观众看懂的变化或悬念。
【声音】只写可发生的同期声 / 人声状态；没有配乐就明确写「无配乐，仅同期声」。
【声音】台词/旁白状态；音效；背景音乐；字幕。没有就写「无」；没有配乐写「无配乐，仅同期声」。
【画面内容】
0-3s：景别；构图；运镜；人物/商品动作；情绪或信息变化。
3-8s：景别；构图；运镜；手与商品的空间关系；可见细节。
8-12s：景别；构图；运镜；动作结果或卖点证据。
12-15s：景别；构图；运镜；反应/悬念/自然收束。
0-3s：景别；机位；运镜；人物/商品动作；情绪或信息变化。
3-8s：景别；机位；运镜；手与商品的空间关系；可见细节。
8-12s：景别；机位；运镜；动作结果或卖点证据。
12-15s：景别；机位；运镜；反应/悬念/自然收束。
```
每条秒级分镜都要明确写出 **景别、构图、运镜、动作、信息变化** 五项中的至少四项。
每条秒级分镜都要明确写出 **景别、机位、运镜、动作、信息变化** 五项中的至少四项。
输入如果是 `【镜头 01】` 导演分镜稿，把原稿的景别、机位、运镜、动作、表情、音效、背景音乐、字幕、备注折进对应栏，不要压成一句画面摘要。
同一角色、商品、场景的外观一律引用 entities 里的既定设定，不在每一镜随意换发型、服装、包装、光线或地点。
"""
@@ -492,7 +497,7 @@ def build_agent_messages(
beats_line = (
f"【秒级分镜】每个 {SEGMENT_DURATION_MAX} 秒场必须拆成 3–5 个分镜,visual 必须按「导演说明书」写成多行:"
"先写【本镜任务】、【声音】、【画面内容】,再写秒级分镜;"
f"每条格式`0-3s：景别；构图；运镜；谁在做什么；信息变化`,最后一条接到 {SEGMENT_DURATION_MAX}s。"
f"每条格式`0-3s：景别；机位；运镜；谁在做什么；信息变化`,最后一条接到 {SEGMENT_DURATION_MAX}s。"
"每条写清手/商品/容器的空间关系和真实用法"
"(茶=热水+蒸汽+茶汤变色,禁止茶包丢进冷白开;手从真实方向入画,禁止悬浮肢体)。"
"禁止一句空画面撑满 15 秒。允许另给 beats 数组,后端会折进 visual。\n"
@@ -519,6 +524,18 @@ def build_agent_messages(
"【场景种草骨架】按镜走完:场景建立(商品先不出现)→商品被使用着入场→场景因它更好/软CTA。"
"每镜写出具体时刻+地点+光/声音/物件。口播时每镜仍要说满字数,留白靠停顿不是靠少写字。\n"
)
elif structure == "promo":
structure_line = (
"【促销抢购合规】只可使用商品资料中明确给出的价格、优惠、库存、赠品和活动时间；"
"没有明确数据时，绝不写原价、最低价、限时、限量、领券或赠品。"
"结尾只给一个清楚的购买入口和操作动作，不制造虚假的紧迫感。\n"
)
elif structure == "knowledge":
structure_line = (
"【知识分享骨架】先提出一个与商品相关的选购/使用误区或判断方法，再用实物与步骤证明，"
"最后把商品作为符合该判断的具体实例。只能讲商品资料可支持的事实，"
"禁止编造专业身份、数据、标准或功效承诺。\n"
)
combo_line = (
f"严格按已加载的「{PRESENTATION_FORMATS[fmt]} × {VIDEO_STRUCTURES[structure]}」套路写,"
f"不要串成别的结构的套话。\n"
@@ -562,15 +579,30 @@ def build_agent_messages(
)
else:
extra_block = f"\n\n【补充要求】{extra}\n" if extra else "\n"
user = (
"【任务】全自动(模式①):仅凭上面的商品事实与前置条件,按指定的表现形式与视频结构套路"
"自动定镜/选 tone/造 entity/填结构骨架。"
"不要另起一个空主题;钩子、痛点、卖点必须能对上这份商品,而不是品类套话。"
"每镜旁白要能撑满 15 秒;画面必须按秒拆分镜,够导演在 15 秒里切 3–5 刀。\n"
f"{head}"
f"{extra_block}\n"
"请按技能流程一次性产出 ScriptDraft。"
)
if extra and _looks_like_shot_digest(extra):
user = (
"【任务】参考视频改写:用户给了一份逐镜拆解稿。"
"照搬它的镜头顺序、每镜时长比例、景别、机位、运镜、人物动作和声音层次,"
"把人物、商品、品牌、台词全部换成当前商品。"
"写 visual 时把拆解稿里的景别/机位/运镜/人物动作/表情/音效/背景音乐/字幕/备注"
"折进【声音】和【画面内容】的秒级分镜,不要压成一句画面摘要。"
"原稿写「无 / 不可见 / 听不清」的栏不要编造。"
"钩子、痛点、卖点必须能对上这份商品。"
"每镜旁白要能撑满指定时长;画面必须按秒拆分镜,够导演在 15 秒里切 3–5 刀。\n"
f"{head}"
f"{extra_block}\n"
"请按技能流程一次性产出 ScriptDraft。"
)
else:
user = (
"【任务】全自动(模式①):仅凭上面的商品事实与前置条件,按指定的表现形式与视频结构套路"
"自动定镜/选 tone/造 entity/填结构骨架。"
"不要另起一个空主题;钩子、痛点、卖点必须能对上这份商品,而不是品类套话。"
"每镜旁白要能撑满 15 秒;画面必须按秒拆分镜,够导演在 15 秒里切 3–5 刀。\n"
f"{head}"
f"{extra_block}\n"
"请按技能流程一次性产出 ScriptDraft。"
)
return [{"role": "system", "content": system}, {"role": "user", "content": user}]
@@ -607,6 +639,14 @@ def _balanced_object(text: str) -> str | None:
return None
def _looks_like_shot_digest(text: str) -> bool:
"""用户贴进来的是视频提炼分镜稿,不是一句补充要求。"""
blob = text or ""
has_shot = "【镜头" in blob or ("【第" in blob and "镜】" in blob)
has_picture = "画面：" in blob or "画面:" in blob
return has_shot and has_picture
def _looks_like_draft(blob: str) -> bool:
try:
d = json.loads(blob)
core/backend/apps/ai/services.py
+32 -9
View File
@@ -3225,8 +3225,31 @@ def collect_video_review_blockers(project, only_segment: "VideoSegment | None" =
return blockers
def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> VideoSegmentVersion | None:
model_config = get_default_model(ModelConfig.Capability.VIDEO)
def submit_video_segment(
*,
video_segment: VideoSegment,
user,
prompt: str,
model_config_id=None,
aspect_ratio: str = "9:16",
resolution: str = "720p",
) -> VideoSegmentVersion | None:
model_config = None
if model_config_id:
model_config = (
ModelConfig.objects.select_related("provider")
.filter(
id=model_config_id,
capability=ModelConfig.Capability.VIDEO,
status=ModelConfig.Status.ACTIVE,
provider__status="active",
)
.first()
)
if model_config is None:
raise ValueError("selected video model is unavailable")
else:
model_config = get_default_model(ModelConfig.Capability.VIDEO)
if model_config is None:
raise ValueError("no active video model configured")
project = video_segment.project
@@ -3244,15 +3267,15 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
reference_images = [r["url"] for r in refs]
final_prompt = build_video_segment_prompt(project, video_segment, scene, refs, prompt)
# 视频段 token 计量计价(与自由创作同一成本表+同一毛利):按 9:16/720p/目标时长预估,
# 视频段 token 计量计价(与自由创作同一成本表+同一毛利):按用户选定的比例/清晰度/目标时长预估,
# 预留=积分×buffer,终态按火山真实 usage.total_tokens 结算(poll_video_segment true-up)。
# 这里终结了「视频 ¥1/段、成本 ¥15」的倒贴定价。
from apps.billing.pricing import quote_video_estimate, video_reserve_amount
est_tokens, quote = quote_video_estimate(
model_config,
aspect_ratio="9:16",
resolution="720p",
aspect_ratio=aspect_ratio,
resolution=resolution,
duration=video_segment.target_duration_seconds,
references=[],
team=project.team,
@@ -3269,8 +3292,8 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
"endpoint": model_config.endpoint,
"prompt": final_prompt,
"duration": video_segment.target_duration_seconds,
"ratio": "9:16",
"resolution": "720p",
"ratio": aspect_ratio,
"resolution": resolution,
"estimated_tokens": est_tokens,
# 团队价格系数快照:按实结算用它,中途改价不影响在途任务(jimeng 同款纪律)
"price_multiplier": quote.meta.get("price_multiplier", "1"),
@@ -3288,8 +3311,8 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
primary_model=model_config,
prompt=final_prompt,
duration=video_segment.target_duration_seconds,
ratio="9:16",
resolution="720p",
ratio=aspect_ratio,
resolution=resolution,
reference_images=reference_images,
request_summary={"video_segment_id": str(video_segment.id)},
)
core/backend/apps/ai/test_script_duration_combo.py
+30 -7
View File
@@ -249,6 +249,29 @@ class PromptAssemblyTests(SimpleTestCase):
self.assertIn("禁止一句 20 字收工", user)
self.assertIn("秒级分镜", user)
def test_video_digest_prompt_maps_camera_and_sound(self):
digest = (
"片名：《雨夜归家》\n视频类型：剧情短片\n"
"【镜头 01】\n时间：00:00-00:04\n景别：城市远景\n运镜：缓慢向前推进\n"
"画面：深夜街道被雨水覆盖。\n台词/旁白：无\n音效：雨声\n"
)
messages = build_agent_messages(
project=self._project(),
mode="auto",
user_prompt=digest,
selling_point_ids=["1"],
base_draft=None,
aspect_ratio="9:16",
total_duration=15,
persona="urban",
)
user = messages[1]["content"]
self.assertIn("参考视频改写", user)
self.assertNotIn("全自动", user)
self.assertIn("景别/机位/运镜", user)
self.assertIn("音效/背景音乐", user)
self.assertIn("【镜头 01】", user)
def test_scene_oral_prompt_locks_skeleton_and_density(self):
messages = build_agent_messages(
project=self._project(),
@@ -284,15 +307,15 @@ class PromptAssemblyTests(SimpleTestCase):
class NarrationLimitTests(SimpleTestCase):
def test_limit_scales_with_shot_length(self):
self.assertEqual(narration_limit(4), 18)
self.assertEqual(narration_limit(8), 36)
self.assertEqual(narration_limit(15), 67)
self.assertEqual(narration_limit(4), 20)
self.assertEqual(narration_limit(8), 41)
self.assertEqual(narration_limit(15), 78)
def test_never_exceeds_hard_cap(self):
self.assertLessEqual(narration_limit(60), 68)
self.assertLessEqual(narration_limit(60), 78)
def test_fifteen_second_floor_is_about_fifty_four(self):
self.assertEqual(narration_floor(15), 54)
def test_fifteen_second_floor_is_about_sixty_four(self):
self.assertEqual(narration_floor(15), 64)
self.assertLess(narration_floor(15), narration_limit(15))
@@ -408,7 +431,7 @@ class ComboTests(SimpleTestCase):
def test_drama_cannot_pick_review(self):
# 演出来的实测没有可信度 —— 这是唯一的禁用组合
self.assertNotIn("review", allowed_structures("drama"))
self.assertEqual(len(allowed_structures("drama")), 3)
self.assertEqual(len(allowed_structures("drama")), 5)
def test_other_formats_allow_everything(self):
for fmt in ("oral", "vlog"):
core/backend/apps/ai/test_video_digest.py
+168 -7
View File
@@ -10,7 +10,7 @@ import subprocess
from unittest.mock import patch
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import SimpleTestCase, TestCase
from django.test import SimpleTestCase, TestCase, override_settings
from rest_framework.test import APIClient
from apps.accounts.models import Team, TeamMember, User
@@ -36,6 +36,14 @@ from apps.ai.video_digest import (
)
def _shot_digest(shots=3, *, heading="lens"):
lines = ["片名：《示例》\n视频类型：口播带货\n时长：15秒\n整体风格：写实\n"]
for index in range(1, shots + 1):
mark = f"【镜头 {index:02d}】" if heading == "lens" else f"【第 {index} 镜】"
lines.append(f"{mark}\n时间：00:00-00:04\n画面：一位女生拿着商品\n台词/旁白：无\n")
return "".join(lines)
class FramePlanTests(SimpleTestCase):
def test_frame_count_scales_with_duration_within_bounds(self):
for duration, expected in [(3, MIN_FRAMES), (10, MIN_FRAMES), (30, 15), (60, 30), (180, MAX_FRAMES)]:
@@ -98,12 +106,15 @@ class MessageBuildTests(SimpleTestCase):
def test_system_prompt_is_the_digest_skill(self):
messages = build_digest_messages(self.frames, 10)
self.assertEqual(messages[0]["role"], "system")
self.assertIn("解说词", messages[0]["content"])
self.assertIn("台词/旁白", messages[0]["content"])
self.assertIn("覆盖全片", messages[0]["content"])
self.assertIn("【镜头 01】", messages[0]["content"])
def test_user_prompt_asks_for_full_timeline(self):
content = build_digest_messages(self.frames, 60)[1]["content"]
self.assertIn("全片", content[0]["text"])
self.assertIn("景别", content[0]["text"])
self.assertIn("背景音乐", content[0]["text"])
def test_every_frame_is_inlined_with_its_timestamp(self):
content = build_digest_messages(self.frames, 10)[1]["content"]
@@ -121,17 +132,19 @@ class MessageBuildTests(SimpleTestCase):
self.assertIn("蓝牙耳机 · 数码3C", content[0]["text"])
def test_skill_loads_from_disk(self):
self.assertIn("解说词", load_digest_skill())
self.assertIn("台词/旁白", load_digest_skill())
self.assertIn("景别", load_digest_skill())
class NativeVideoInputTests(SimpleTestCase):
def test_short_clip_is_sent_as_whole_video(self):
video, frames, duration = digest_input_from_upload(_synth_clip(seconds=6))
video, frames, duration, extras = digest_input_from_upload(_synth_clip(seconds=6))
self.assertIsNotNone(video)
self.assertEqual(frames, [])
self.assertTrue(video.mime.startswith("video/"))
self.assertGreater(len(video.data), 100)
self.assertAlmostEqual(duration, 6, delta=1)
self.assertIn("cover_jpeg", extras)
def test_native_video_message_inlines_video_data_uri(self):
video = DigestVideo(mime="video/mp4", data=b"\x00\x00fake")
@@ -140,6 +153,8 @@ class NativeVideoInputTests(SimpleTestCase):
self.assertEqual(len(media), 1)
self.assertTrue(media[0]["image_url"]["url"].startswith("data:video/mp4;base64,"))
self.assertIn("口播", content[0]["text"])
self.assertIn("景别", content[0]["text"])
self.assertIn("背景音乐", content[0]["text"])
class DigestValidationTests(SimpleTestCase):
@@ -153,11 +168,15 @@ class DigestValidationTests(SimpleTestCase):
validate_digest_text("这是一条很好的带货视频,节奏明快,画面精美。" * 5)
def test_accepts_well_formed_digest(self):
good = _shot_digest()
self.assertEqual(validate_digest_text(f" {good} "), good.strip())
def test_accepts_legacy_shot_heading(self):
good = "【整体结构】\n形式:口播\n" + "【第 1 镜】0-4 秒 · 钩子\n主体:一位女生\n" * 3
self.assertEqual(validate_digest_text(f" {good} "), good.strip())
def test_rejects_too_few_shots_for_a_long_clip(self):
stub = "【整体结构】\n形式:口播\n" + "【第 1 镜】0-4 秒 · 钩子\n主体:一位女生\n" * 3
stub = _shot_digest(3)
with self.assertRaises(ValueError):
validate_digest_text(stub, duration=60, frame_count=30)
@@ -324,7 +343,142 @@ class VideoDigestApiTests(TestCase):
self.assertTrue(response.data["detail"])
self.assertEqual(response.data["error"]["operation"], "video_digest")
def _seed_model(self):
from decimal import Decimal
provider, _ = ModelProvider.objects.get_or_create(
name="remix-hist-relay",
defaults={"display_name": "Gemini 官转", "status": ModelProvider.Status.ACTIVE},
)
return ModelConfig.objects.create(
provider=provider,
name=DIGEST_VISION_MODEL_NAME,
display_name="Gemini 3.1 Pro 官转",
capability=ModelConfig.Capability.TEXT,
status=ModelConfig.Status.ACTIVE,
unit_price=Decimal("10"),
)
def _seed_digest_task(self, *, team=None, status=None, is_deleted=False, prompt="示例提示词", extra_key=""):
from apps.ai.models import AITask
model = getattr(self, "_hist_model", None) or self._seed_model()
self._hist_model = model
task = AITask.objects.create(
team=team or self.team,
created_by=self.user,
project=None,
task_type=AITask.Type.VIDEO_DIGEST,
status=status or AITask.Status.SUCCEEDED,
model_config=model,
idempotency_key=f"digest-hist-{team.id if team else self.team.id}-{extra_key or AITask.objects.count()}",
request_payload={
"feature": "video_remix",
"file_name": "参考视频.mp4",
"title": "轻醒咖啡场景种草视频",
"duration_seconds": 15,
"ratio": "9:16 竖屏",
"shot_count": 5,
"video_url": "https://cdn.example/coffee.mp4",
"cover_url": "https://cdn.example/coffee.jpg",
},
response_payload={"digest": prompt, "prompt": prompt},
is_deleted=is_deleted,
)
return task
def test_history_empty_by_default(self):
response = self.client.get("/api/ai/video-digest/")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["results"], [])
self.assertEqual(response.data["total"], 0)
def test_history_lists_succeeded_team_digests(self):
from apps.ai.models import AITask
self._seed_digest_task(prompt="咖啡提示词", extra_key="ok")
self._seed_digest_task(status=AITask.Status.FAILED, extra_key="fail")
self._seed_digest_task(is_deleted=True, extra_key="del")
other_user = User.objects.create_user(username="remix-other", password="p")
other_team = Team.objects.create(name="Other", owner=other_user)
TeamMember.objects.create(team=other_team, user=other_user, role=TeamMember.Role.OWNER)
self._seed_digest_task(team=other_team, extra_key="other")
response = self.client.get("/api/ai/video-digest/")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["total"], 1)
item = response.data["results"][0]
self.assertEqual(item["title"], "轻醒咖啡场景种草视频")
self.assertEqual(item["status"], "已完成")
self.assertEqual(item["duration_label"], "00:15")
self.assertEqual(item["ratio"], "9:16 竖屏")
self.assertEqual(item["shots"], 5)
self.assertEqual(item["file_name"], "参考视频.mp4")
self.assertEqual(item["prompt"], "咖啡提示词")
self.assertEqual(item["video_url"], "https://cdn.example/coffee.mp4")
self.assertEqual(item["cover_url"], "https://cdn.example/coffee.jpg")
self.assertRegex(item["created_date"], r"^\d{4}-\d{2}-\d{2}$")
@patch("apps.assets.storage.TosStorage")
def test_history_signs_video_url_from_object_key(self, storage_cls):
storage = storage_cls.return_value
storage.presigned_get_url.return_value = "https://signed.example/play.mp4"
storage.public_url.return_value = "https://cdn.example/cover.jpg"
task = self._seed_digest_task(extra_key="signed")
task.request_payload = {
**task.request_payload,
"video_url": "",
"video_key": "teams/1/video-digest/abc.mp4",
"cover_url": "",
"cover_key": "teams/1/video-digest/abc.jpg",
}
task.save(update_fields=["request_payload"])
response = self.client.get("/api/ai/video-digest/")
self.assertEqual(response.status_code, 200)
item = response.data["results"][0]
self.assertEqual(item["video_url"], "https://signed.example/play.mp4")
self.assertEqual(item["cover_url"], "https://cdn.example/cover.jpg")
storage.presigned_get_url.assert_called()
def test_history_without_saved_source_has_empty_video_url(self):
task = self._seed_digest_task(extra_key="nosource")
payload = dict(task.request_payload)
payload.pop("video_url", None)
payload.pop("video_key", None)
task.request_payload = payload
task.save(update_fields=["request_payload"])
response = self.client.get("/api/ai/video-digest/")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["results"][0]["video_url"], "")
def test_save_prompt_updates_history(self):
task = self._seed_digest_task(prompt="旧提示词", extra_key="save")
response = self.client.patch(
f"/api/ai/video-digest/{task.id}/",
{"prompt": "新提示词"},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["prompt"], "新提示词")
task.refresh_from_db()
self.assertEqual(task.response_payload["prompt"], "新提示词")
self.assertEqual(task.response_payload["digest"], "旧提示词")
def test_save_prompt_rejects_empty_and_missing(self):
task = self._seed_digest_task(extra_key="empty")
empty = self.client.patch(f"/api/ai/video-digest/{task.id}/", {"prompt": " "}, format="json")
self.assertEqual(empty.status_code, 400)
missing = self.client.patch(
"/api/ai/video-digest/00000000-0000-0000-0000-000000000099/",
{"prompt": "还在"},
format="json",
)
self.assertEqual(missing.status_code, 404)
@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}})
class VideoDigestBillingTests(TestCase):
"""提炼提示词按功能价 30 积分结算,不跟 Gemini 文本单价。"""
@@ -362,7 +516,7 @@ class VideoDigestBillingTests(TestCase):
with (
patch(
"apps.ai.video_digest.digest_input_from_upload",
return_value=(None, [VideoFrame(at_seconds=2, jpeg=b"\xff\xd8x")], 15.0),
return_value=(None, [VideoFrame(at_seconds=2, jpeg=b"\xff\xd8x")], 15.0, {}),
),
patch(
"apps.ai.services.execute_routed_text_request",
@@ -380,6 +534,13 @@ class VideoDigestBillingTests(TestCase):
from decimal import Decimal
from apps.ai.models import AITask
from apps.billing.pricing import VIDEO_DIGEST_POINTS, quote_video_digest
self.assertEqual(VIDEO_DIGEST_POINTS, Decimal("30"))
self.assertEqual(quote_video_digest(model_config=self.model).points, Decimal("30"))
self.team.price_multiplier = Decimal("0.80")
self.assertEqual(quote_video_digest(team=self.team, model_config=self.model).points, Decimal("24"))
self.team.price_multiplier = Decimal("1")
result = self._run_digest()
self.account.refresh_from_db()
@@ -401,7 +562,7 @@ class VideoDigestBillingTests(TestCase):
with (
patch(
"apps.ai.video_digest.digest_input_from_upload",
return_value=(None, [VideoFrame(at_seconds=2, jpeg=b"\xff\xd8x")], 15.0),
return_value=(None, [VideoFrame(at_seconds=2, jpeg=b"\xff\xd8x")], 15.0, {}),
),
patch("apps.ai.services.execute_routed_text_request") as mocked,
):
core/backend/apps/ai/urls.py
+2
View File
@@ -3,6 +3,7 @@ from rest_framework.routers import DefaultRouter
from .views import (
AITaskViewSet,
VideoDigestDetailView,
VideoDigestView,
FreeVideoDetailView,
FreeVideoFavoriteView,
@@ -25,6 +26,7 @@ router.register("image-conversations", ImageConversationViewSet, basename="image
urlpatterns = [
path("generate-image/", GenerateImageView.as_view(), name="ai-generate-image"),
path("video-digest/", VideoDigestView.as_view(), name="ai-video-digest"),
path("video-digest/<uuid:task_id>/", VideoDigestDetailView.as_view(), name="ai-video-digest-detail"),
path("free-video/", FreeVideoView.as_view(), name="ai-free-video"),
path("free-video/trash/", FreeVideoTrashView.as_view(), name="ai-free-video-trash"),
path("free-video/upload/", FreeVideoUploadView.as_view(), name="ai-free-video-upload"),
core/backend/apps/ai/video_digest.py
+364 -21
View File
@@ -46,8 +46,8 @@ MIN_FRAMES = 8
MAX_FRAMES = 36
FRAME_WIDTH = 768
FRAME_QUALITY = 3
DIGEST_MAX_TOKENS = 8192
_SHOT_MARK = re.compile(r"【第\s*\d+\s*镜】")
DIGEST_MAX_TOKENS = 12288
_SHOT_MARK = re.compile(r"【(?:镜头\s*\d+|第\s*\d+\s*镜)】")
_FFMPEG_TIMEOUT = 60
@@ -93,9 +93,10 @@ def load_digest_skill() -> str:
return main.read_text(encoding="utf-8")
# 兜底:skill 丢了也别整条链路挂掉,退化成一句话提示词(产出会明显变差,交接文档已注明须带 skills 目录)
return (
"你是分镜拆解 agent。输入是一条电商短视频的完整文件,含画面和口播。"
"必须覆盖全片,从 0 秒写到片尾,有几镜写几镜,不要概括成几大段。"
"每镜写画面和解说词,解说词按听到的口播逐字写。输出中文纯文本。"
"你是分镜拆解 agent。输入是一条短视频的完整文件,含画面和音轨。"
"必须覆盖全片,从 00:00 写到片尾,有几镜写几镜,不要概括成几大段。"
"每镜按【镜头 01】写出时间、时长、景别、机位、运镜、画面、人物动作、人物表情、"
"台词/旁白、音效、背景音乐、字幕、备注。没有就写无。输出中文纯文本。"
)
@@ -130,6 +131,73 @@ def probe_duration(path: str | Path) -> float:
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)))
@@ -223,6 +291,60 @@ def _compress_video(path: str) -> bytes | None:
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:
@@ -233,17 +355,37 @@ def _native_video(path: str, size: int, suffix: str) -> DigestVideo | None:
return None
def digest_input_from_upload(upload) -> tuple[DigestVideo | None, list[VideoFrame], float]:
"""优先整段视频(含音轨);塞不进请求才抽帧。"""
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)
if video is not None:
return video, [], duration
return None, extract_frames(path, plan_frame_times(duration)), duration
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:
if path and not keep:
Path(path).unlink(missing_ok=True)
@@ -267,20 +409,28 @@ def build_digest_messages(
*,
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"这是一条时长约 {round(duration)} 秒的电商带货短视频的**完整文件**(含画面和口播音轨)。",
"请按技能还原全片分镜:从 0 秒写到片尾,有几镜写几镜。",
"解说词按听到的口播逐字写;画面上的花字一并写进画面。",
f"这是一条{measured}的短视频**完整文件**(含画面和口播音轨)。{title_hint}",
"请按技能还原全片导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
"每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、台词/旁白、音效、背景音乐、字幕、备注。",
"台词/旁白按听到的口播逐字写;画面上的花字写进字幕。",
]
else:
head = [
f"这是一条时长约 {round(duration)} 秒的电商带货短视频,",
f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。",
"请按技能还原**全片**分镜:从 0 秒写到片尾,有几镜写几镜。",
f"这是一条{measured}的短视频,",
f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。{title_hint}",
"请按技能还原**全片**导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
"每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、台词/旁白、音效、背景音乐、字幕、备注。",
]
if product_hint:
head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")
@@ -429,6 +579,127 @@ def digest_team_video(*, team, user, upload, model_config_id=None) -> dict:
)
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)
def _digest_video(*, team, user, upload, project=None, product_hint="", model_config_id=None) -> dict:
"""上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
from django.db import transaction
@@ -439,10 +710,20 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
from apps.billing.pricing import quote_video_digest
from apps.billing.services.ledger import charge_reserved_credit, reserve_credit
video, frames, duration = digest_input_from_upload(upload)
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 = 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",
@@ -456,7 +737,12 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
)
messages = build_digest_messages(
frames, duration, product_hint=product_hint, video=video
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,
@@ -467,6 +753,12 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
"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)
@@ -502,6 +794,8 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
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
@@ -532,15 +826,56 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
)
_text, _response, digest = routed.value
except Exception as exc: # noqa: BLE001
if source_path:
Path(source_path).unlink(missing_ok=True)
_fail_digest_task(task, reservation, str(exc))
raise
cover_key, cover_url, video_key, video_url = "", "", "", ""
if project is None:
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:
video_key, video_url = _store_digest_video(
team=team,
path=source_path,
suffix=str(extras.get("suffix") or ".mp4"),
)
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 = {
**(task.request_payload or {}),
**request_payload,
"shot_count": shots,
"cover_key": cover_key,
"cover_url": cover_url,
"video_key": video_key,
"video_url": video_url,
}
with transaction.atomic():
task.status = AITask.Status.SUCCEEDED
task.response_payload = {"digest": digest[:32000]}
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", "response_payload", "actual_cost", "completed_at", "updated_at"])
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 {
@@ -551,6 +886,14 @@ def _digest_video(*, team, user, upload, project=None, product_hint="", model_co
"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,
}
core/backend/apps/ai/views.py
+32 -2
View File
@@ -7,7 +7,7 @@ from django.utils import timezone
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.parsers import FormParser, MultiPartParser
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
@@ -700,9 +700,17 @@ class VideoDigestView(APIView):
"""视频提炼 · 上传参考视频提炼分镜稿(不绑项目)。
POST /api/ai/video-digest/ multipart file → 中文分镜稿。慢(30~60 秒),固定 30 积分/次,失败退还。
GET /api/ai/video-digest/ 本团队已完成的提炼历史(新→旧)。
"""
parser_classes = [MultiPartParser, FormParser]
parser_classes = [MultiPartParser, FormParser, JSONParser]
def get(self, request):
from .video_digest import list_team_digest_history
team = get_current_team(request.user)
results = list_team_digest_history(team=team)
return Response({"results": results, "total": len(results)})
def post(self, request):
from .video_digest import VideoDigestError, digest_team_video
@@ -730,6 +738,28 @@ class VideoDigestView(APIView):
return Response({"name": getattr(upload, "name", "") or "参考视频", **result})
class VideoDigestDetailView(APIView):
"""PATCH /api/ai/video-digest/<id>/ 保存编辑后的提示词到这条历史。"""
parser_classes = [JSONParser, FormParser]
def patch(self, request, task_id):
from .video_digest import VideoDigestError, save_digest_prompt
team = get_current_team(request.user)
try:
item = save_digest_prompt(
team=team,
task_id=task_id,
prompt=str(request.data.get("prompt") or ""),
)
except VideoDigestError as exc:
return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
if item is None:
return Response({"detail": "记录不存在"}, status=status.HTTP_404_NOT_FOUND)
return Response(item)
class FreeVideoView(APIView):
"""自由创作·视频生成(不绑项目,universal 全能参考 / keyframe 首尾帧)。