Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 938ef0b8cc | |||
| 982daac6e5 | |||
| a7067c8171 | |||
| 32c3d2f263 |
@@ -0,0 +1,27 @@
|
||||
"""add sentence_timings to lipsync_jobs
|
||||
|
||||
Revision ID: 075_add_sentence_timings
|
||||
Revises: 074_ai_avatar_render_script_id_optional
|
||||
Create Date: 2026-09-12
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "075_add_sentence_timings"
|
||||
down_revision = "074_render_script_id_optional"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table("lipsync_jobs") as batch:
|
||||
batch.add_column(
|
||||
sa.Column("sentence_timings", sa.JSON(), nullable=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("lipsync_jobs") as batch:
|
||||
batch.drop_column("sentence_timings")
|
||||
@@ -33,6 +33,7 @@ class LipsyncJobResponse(BaseModel):
|
||||
output_duration: float
|
||||
error_message: str
|
||||
error_code: str
|
||||
sentence_timings: Optional[list] = None
|
||||
submitted_at: Optional[datetime] = None
|
||||
completed_at: Optional[datetime] = None
|
||||
created_at: datetime
|
||||
|
||||
@@ -25,6 +25,7 @@ from packages.adapters.sqlalchemy_impl.models import (
|
||||
ScriptModel,
|
||||
)
|
||||
from packages.domain.video_filter_builder import (
|
||||
build_broll_overlay_filter,
|
||||
build_title_drawtext_filter,
|
||||
)
|
||||
from packages.shared.storage import get_shared_storage_service
|
||||
@@ -228,27 +229,49 @@ class AiAvatarRenderService:
|
||||
self.db.commit()
|
||||
|
||||
# 2. 构建 FFmpeg 滤镜链 (40%)
|
||||
from packages.domain.video_filter_builder import build_broll_overlay_filter
|
||||
# 用 ffprobe 探测输入视频分辨率,确保 B-roll 缩放与标题位置与实际输出一致。
|
||||
# AI 数字人对口型输出为 9:16 竖屏,默认兜底 720x1280;探测失败时使用默认值不阻断渲染。
|
||||
output_width, output_height = self._probe_video_resolution(input_video_path)
|
||||
if output_width <= 0 or output_height <= 0:
|
||||
output_width, output_height = 720, 1280
|
||||
logger.info(
|
||||
"[数字人渲染] ffprobe 探测分辨率失败或无效,使用默认竖屏尺寸 %sx%s",
|
||||
output_width,
|
||||
output_height,
|
||||
)
|
||||
else:
|
||||
logger.info("[数字人渲染] 探测输入视频分辨率: %sx%s", output_width, output_height)
|
||||
|
||||
filter_complex = build_broll_overlay_filter(
|
||||
broll_filter, broll_label = build_broll_overlay_filter(
|
||||
b_roll_segments=job.b_roll_segments,
|
||||
video_duration=lipsync_job.output_duration,
|
||||
output_width=output_width,
|
||||
output_height=output_height,
|
||||
)
|
||||
|
||||
# 标题叠加
|
||||
title_filter = build_title_drawtext_filter(job.title_config)
|
||||
if title_filter:
|
||||
if filter_complex:
|
||||
filter_complex += f"[vout]{title_filter}[vout_titled];"
|
||||
else:
|
||||
filter_complex = f"[0:v]{title_filter}[vout_titled];"
|
||||
# 标题叠加(传入实际输出尺寸,保证位置计算正确)
|
||||
title_filter = build_title_drawtext_filter(
|
||||
job.title_config,
|
||||
output_width=output_width,
|
||||
output_height=output_height,
|
||||
)
|
||||
|
||||
# 清理末尾分号
|
||||
if filter_complex.endswith(";"):
|
||||
filter_complex = filter_complex[:-1]
|
||||
|
||||
# 最终输出标签
|
||||
final_label = "vout_titled" if title_filter else ("vout" if filter_complex else None)
|
||||
filter_complex = ""
|
||||
final_label = None
|
||||
if broll_filter and title_filter:
|
||||
# B-roll → 标题叠在 B-roll 输出上
|
||||
filter_complex = broll_filter + f";[{broll_label}]{title_filter}[vout_titled]"
|
||||
final_label = "vout_titled"
|
||||
elif broll_filter:
|
||||
filter_complex = broll_filter
|
||||
final_label = broll_label
|
||||
elif title_filter:
|
||||
filter_complex = f"[0:v]{title_filter}[vout_titled]"
|
||||
final_label = "vout_titled"
|
||||
else:
|
||||
# 无滤镜:直接拷贝视频流
|
||||
filter_complex = ""
|
||||
final_label = None
|
||||
|
||||
job.progress = 40
|
||||
self.db.commit()
|
||||
@@ -427,6 +450,37 @@ class AiAvatarRenderService:
|
||||
os.unlink(tmp.name)
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def _probe_video_resolution(video_path: str) -> tuple[int, int]:
|
||||
"""用 ffprobe 探测视频分辨率,返回 (width, height);失败返回 (0, 0)。"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"v:0",
|
||||
"-show_entries",
|
||||
"stream=width,height",
|
||||
"-of",
|
||||
"csv=p=0:s=x",
|
||||
video_path,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
parts = result.stdout.strip().split("x")
|
||||
if len(parts) == 2:
|
||||
w, h = int(parts[0]), int(parts[1])
|
||||
if w > 0 and h > 0:
|
||||
return w, h
|
||||
except Exception as exc:
|
||||
logger.warning("[数字人渲染] ffprobe 探测分辨率失败: %s", exc)
|
||||
return 0, 0
|
||||
|
||||
def _build_ffmpeg_command(
|
||||
self,
|
||||
*,
|
||||
@@ -450,7 +504,16 @@ class AiAvatarRenderService:
|
||||
cmd.extend(["-i", asset_url])
|
||||
|
||||
if filter_complex and final_label:
|
||||
cmd.extend(["-filter_complex", filter_complex, "-map", f"[{final_label}]"])
|
||||
cmd.extend(
|
||||
[
|
||||
"-filter_complex",
|
||||
filter_complex,
|
||||
"-map",
|
||||
f"[{final_label}]",
|
||||
"-map",
|
||||
"0:a?",
|
||||
]
|
||||
)
|
||||
elif filter_complex:
|
||||
cmd.extend(["-filter_complex", filter_complex])
|
||||
|
||||
@@ -462,6 +525,10 @@ class AiAvatarRenderService:
|
||||
"veryfast",
|
||||
"-crf",
|
||||
"23",
|
||||
"-c:a",
|
||||
"aac",
|
||||
"-b:a",
|
||||
"128k",
|
||||
"-y",
|
||||
output_path,
|
||||
]
|
||||
|
||||
@@ -54,6 +54,160 @@ def _sign_media_url(url: str) -> str:
|
||||
return url
|
||||
|
||||
|
||||
def _split_script_into_sentences(script_text: str) -> list[str]:
|
||||
"""按句号/问号/感叹号/分号/换行分句(与前端 splitScriptIntoSentences 一致)."""
|
||||
import re
|
||||
|
||||
text = (script_text or "").strip()
|
||||
if not text:
|
||||
return []
|
||||
parts = re.split(r"[。!?!?;;\n\r]+", text)
|
||||
return [p.strip() for p in parts if p.strip()]
|
||||
|
||||
|
||||
def _compute_sentence_timings(audio_data: bytes, script_text: str, total_duration: float) -> list[dict]:
|
||||
"""基于 TTS 音频的静音检测,精确计算每句文案的起止时间.
|
||||
|
||||
使用 ffmpeg silencedetect 检测静音段,将静音点与句子边界对齐。
|
||||
比字数比例估算准确得多。
|
||||
|
||||
Args:
|
||||
audio_data: TTS 音频二进制数据(MP3)
|
||||
script_text: 文案全文
|
||||
total_duration: 音频总时长(秒)
|
||||
|
||||
Returns:
|
||||
list[{"index": int, "text": str, "start_time": float, "end_time": float}]
|
||||
"""
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
sentences = _split_script_into_sentences(script_text)
|
||||
if not sentences:
|
||||
return []
|
||||
|
||||
# 写入临时音频文件
|
||||
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp:
|
||||
tmp.write(audio_data)
|
||||
tmp_path = tmp.name
|
||||
|
||||
try:
|
||||
# 用 ffmpeg silencedetect 检测静音段
|
||||
result = subprocess.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-i",
|
||||
tmp_path,
|
||||
"-af",
|
||||
"silencedetect=noise=-25dB:d=0.3",
|
||||
"-f",
|
||||
"null",
|
||||
"-",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
stderr = result.stderr or ""
|
||||
|
||||
# 解析静音结束时间点(silence_end: X.XXX)
|
||||
silence_ends = []
|
||||
for match in re.finditer(r"silence_end:\s*([\d.]+)", stderr):
|
||||
t = float(match.group(1))
|
||||
if 0 < t < total_duration:
|
||||
silence_ends.append(t)
|
||||
|
||||
# 如果没有检测到足够的静音点,降级为字数比例估算
|
||||
if len(silence_ends) < len(sentences) - 1:
|
||||
logger.warning(
|
||||
"[sentence_timings] 静音点不足(%d < %d),降级为字数比例估算",
|
||||
len(silence_ends),
|
||||
len(sentences) - 1,
|
||||
)
|
||||
return _estimate_sentence_timings_by_chars(sentences, total_duration)
|
||||
|
||||
# 贪心匹配:N-1 个句子边界对应 N-1 个静音点
|
||||
# 按时间均匀分布期望值,选择最近的静音点
|
||||
n_boundaries = len(sentences) - 1
|
||||
boundaries = []
|
||||
used_indices = set()
|
||||
|
||||
for i in range(n_boundaries):
|
||||
# 期望的边界位置(按句子数量均匀分布)
|
||||
expected_pos = (i + 1) / len(sentences) * total_duration
|
||||
# 找最近的未使用静音点
|
||||
best_idx = None
|
||||
best_dist = float("inf")
|
||||
for j, t in enumerate(silence_ends):
|
||||
if j in used_indices:
|
||||
continue
|
||||
dist = abs(t - expected_pos)
|
||||
if dist < best_dist:
|
||||
best_dist = dist
|
||||
best_idx = j
|
||||
if best_idx is not None:
|
||||
used_indices.add(best_idx)
|
||||
boundaries.append(silence_ends[best_idx])
|
||||
|
||||
boundaries.sort()
|
||||
|
||||
# 构建 sentence_timings
|
||||
timings = []
|
||||
prev_end = 0.0
|
||||
for i, sent in enumerate(sentences):
|
||||
start = prev_end
|
||||
end = boundaries[i] if i < len(boundaries) else total_duration
|
||||
timings.append(
|
||||
{
|
||||
"index": i,
|
||||
"text": sent,
|
||||
"start_time": round(start, 2),
|
||||
"end_time": round(end, 2),
|
||||
}
|
||||
)
|
||||
prev_end = end
|
||||
|
||||
return timings
|
||||
|
||||
except Exception as exc:
|
||||
logger.warning("[sentence_timings] 静音检测异常,降级为字数比例估算: %s", exc)
|
||||
return _estimate_sentence_timings_by_chars(sentences, total_duration)
|
||||
finally:
|
||||
import os
|
||||
|
||||
try:
|
||||
os.unlink(tmp_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _estimate_sentence_timings_by_chars(sentences: list[str], total_duration: float) -> list[dict]:
|
||||
"""降级方案:按字数比例估算句子时间(与原前端逻辑一致)."""
|
||||
if not sentences or total_duration <= 0:
|
||||
return []
|
||||
total_chars = sum(len(s.replace(r"\s", "")) for s in sentences)
|
||||
if total_chars == 0:
|
||||
return []
|
||||
|
||||
timings = []
|
||||
acc = 0
|
||||
for i, sent in enumerate(sentences):
|
||||
chars = len(sent.replace(r"\s", ""))
|
||||
start = (acc / total_chars) * total_duration
|
||||
end = ((acc + chars) / total_chars) * total_duration
|
||||
timings.append(
|
||||
{
|
||||
"index": i,
|
||||
"text": sent,
|
||||
"start_time": round(start, 2),
|
||||
"end_time": round(end, 2),
|
||||
}
|
||||
)
|
||||
acc += chars
|
||||
return timings
|
||||
|
||||
|
||||
@shared_task(
|
||||
bind=True,
|
||||
name="lipsync_tts.synthesize_and_submit",
|
||||
@@ -152,14 +306,14 @@ def tts_synthesize_and_submit(
|
||||
audio_data = safe_download_bytes(
|
||||
temp_url,
|
||||
purpose="lipsync_tts_audio",
|
||||
allowed_mime_types=(
|
||||
allowed_mime_types={
|
||||
"audio/mpeg",
|
||||
"audio/mp3",
|
||||
"audio/wav",
|
||||
"audio/x-wav", # CosyVoice 部分接口返回 audio/x-wav,与 audio/wav 等价(RIFF/WAVE)
|
||||
"audio/mp4",
|
||||
"audio/x-m4a",
|
||||
),
|
||||
},
|
||||
timeout=60.0,
|
||||
)
|
||||
from packages.shared.storage import get_shared_storage_service
|
||||
@@ -179,6 +333,64 @@ def tts_synthesize_and_submit(
|
||||
|
||||
db.commit()
|
||||
|
||||
# 2.5 计算精确句子时间戳(基于 TTS 音频静音检测)
|
||||
import os as _os
|
||||
|
||||
_st_tmp_path = None
|
||||
try:
|
||||
import subprocess as _sp
|
||||
import tempfile as _tmpf
|
||||
|
||||
# 下载音频用于探测时长和静音检测
|
||||
if isinstance(job.audio_url, str) and job.audio_url:
|
||||
from packages.shared.url_security import safe_download_bytes as _sdl
|
||||
|
||||
_audio_bytes = _sdl(job.audio_url, purpose="sentence_timings", timeout=30.0)
|
||||
else:
|
||||
_audio_bytes = audio_data
|
||||
|
||||
# ffprobe 获取音频时长
|
||||
with _tmpf.NamedTemporaryFile(suffix=".mp3", delete=False) as _atmp:
|
||||
_atmp.write(_audio_bytes)
|
||||
_st_tmp_path = _atmp.name
|
||||
|
||||
_probe_result = _sp.run(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"error",
|
||||
"-show_entries",
|
||||
"format=duration",
|
||||
"-of",
|
||||
"default=noprint_wrappers=1:nokey=1",
|
||||
_st_tmp_path,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
_audio_duration = float(_probe_result.stdout.strip()) if _probe_result.stdout.strip() else 0.0
|
||||
|
||||
if _audio_duration > 0:
|
||||
_timings = _compute_sentence_timings(_audio_bytes, script_text, _audio_duration)
|
||||
if _timings:
|
||||
job.sentence_timings = _timings
|
||||
logger.info(
|
||||
"[lipsync_tts] 句子时间戳已计算: job_id=%s sentences=%d duration=%.1f",
|
||||
job_id,
|
||||
len(_timings),
|
||||
_audio_duration,
|
||||
)
|
||||
db.commit()
|
||||
except Exception as _st_err:
|
||||
logger.warning("[lipsync_tts] 句子时间戳计算失败(不影响主流程): job_id=%s err=%s", job_id, _st_err)
|
||||
finally:
|
||||
if _st_tmp_path:
|
||||
try:
|
||||
_os.unlink(_st_tmp_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 3. 签名 URL 并提交到 MediaKit(复用模块内 _sign_media_url,避免对 LipsyncService 的耦合)
|
||||
audio_url = _sign_media_url(job.audio_url)
|
||||
video_url = _sign_media_url(job.video_url)
|
||||
|
||||
@@ -241,6 +241,15 @@ const AiAvatarPage: React.FC = () => {
|
||||
if (renderTimerRef.current) clearInterval(renderTimerRef.current)
|
||||
renderTimerRef.current = null
|
||||
setRenderStatus("completed")
|
||||
// 渲染完成后,用最终视频的封面更新前端封面配置
|
||||
if (updated.output_cover_url) {
|
||||
state.setCoverConfig((prev) => ({
|
||||
...prev,
|
||||
mode: "auto_frame",
|
||||
smart_cover_url: updated.output_cover_url,
|
||||
thumbnail_url: updated.output_cover_url,
|
||||
}))
|
||||
}
|
||||
message.success("视频已生成并保存到成片库")
|
||||
} else if (updated.status === "failed") {
|
||||
if (renderTimerRef.current) clearInterval(renderTimerRef.current)
|
||||
@@ -488,8 +497,9 @@ const AiAvatarPage: React.FC = () => {
|
||||
open={state.showBRollModal}
|
||||
onClose={() => state.setShowBRollModal(false)}
|
||||
existingSegments={state.bRollSegments}
|
||||
scriptText={state.scriptText}
|
||||
scriptText={state.lipsyncJob?.script_text || state.scriptText}
|
||||
outputDuration={state.lipsyncJob?.output_duration ?? 0}
|
||||
sentenceTimings={state.lipsyncJob?.sentence_timings}
|
||||
onConfirm={state.addBRollSegment}
|
||||
onRemove={state.removeBRollSegment}
|
||||
/>
|
||||
|
||||
@@ -5,12 +5,12 @@
|
||||
* - 左侧:先选素材库(video 库)→ 再选该库视频素材(已被其他 segment 使用的素材
|
||||
* 标灰 + "已选择" 遮罩,pointer-events:none 防重复选择)
|
||||
* - 右侧:文案句子列表(点选对应段落,替代原数字索引框)/ 全屏 or 画中画 / 四角位置+大小
|
||||
* (开始/结束时间已删除,按句子字数占比 × 口播总时长自动估算)
|
||||
* (开始/结束时间来自后端精确句子时间戳,基于 TTS 音频静音检测)
|
||||
* - 底部:已配置的画面插入列表(可删除)
|
||||
*/
|
||||
import React, { useEffect, useMemo, useState } from "react"
|
||||
import { getAssets, getAssetLibraries, type AssetItem, type AssetLibraryItem } from "@/api/assets"
|
||||
import type { BRollSegment, BRollInsertMode, PipPosition } from "../types"
|
||||
import type { BRollSegment, BRollInsertMode, PipPosition, SentenceTiming } from "../types"
|
||||
import { splitScriptIntoSentences, type ScriptSentence } from "../utils/sentences"
|
||||
|
||||
interface ModalBRollEditorProps {
|
||||
@@ -18,10 +18,12 @@ interface ModalBRollEditorProps {
|
||||
onClose: () => void
|
||||
/** 当前已有的 B-roll segments(用于标灰已选素材) */
|
||||
existingSegments: BRollSegment[]
|
||||
/** 当前文案全文(用于分句) */
|
||||
/** 文案全文(优先使用对口型时锁定的 scriptText) */
|
||||
scriptText: string
|
||||
/** 对口型成片总时长(秒),用于时间自动估算 */
|
||||
/** 对口型成片总时长(秒) */
|
||||
outputDuration: number
|
||||
/** 后端精确句子时间戳(来自 lipsyncJob.sentence_timings) */
|
||||
sentenceTimings?: SentenceTiming[] | null
|
||||
onConfirm: (segment: BRollSegment) => void
|
||||
onRemove: (id: string) => void
|
||||
}
|
||||
@@ -43,7 +45,8 @@ const ModalBRollEditor: React.FC<ModalBRollEditorProps> = ({
|
||||
onClose,
|
||||
existingSegments,
|
||||
scriptText,
|
||||
outputDuration,
|
||||
outputDuration: _outputDuration,
|
||||
sentenceTimings,
|
||||
onConfirm,
|
||||
onRemove,
|
||||
}) => {
|
||||
@@ -62,10 +65,10 @@ const ModalBRollEditor: React.FC<ModalBRollEditorProps> = ({
|
||||
const [pipPosition, setPipPosition] = useState<PipPosition>("top-right")
|
||||
const [pipScale, setPipScale] = useState(0.3)
|
||||
|
||||
/** 文案分句(⑤) */
|
||||
/** 文案分句(使用后端精确时间戳) */
|
||||
const sentences = useMemo(
|
||||
() => splitScriptIntoSentences(scriptText, outputDuration),
|
||||
[scriptText, outputDuration],
|
||||
() => splitScriptIntoSentences(scriptText, sentenceTimings),
|
||||
[scriptText, sentenceTimings],
|
||||
)
|
||||
|
||||
/** 已被现有 segments 占用的素材 id 集合(标灰、禁止重复选择) */
|
||||
@@ -264,7 +267,7 @@ const ModalBRollEditor: React.FC<ModalBRollEditorProps> = ({
|
||||
>
|
||||
<span className="aa-sentence-item__idx">{sent.index + 1}</span>
|
||||
<span className="aa-sentence-item__text">{sent.text}</span>
|
||||
{outputDuration > 0 && (
|
||||
{sent.endTime > 0 && (
|
||||
<span className="aa-sentence-item__time">
|
||||
{sent.startTime.toFixed(1)}-{sent.endTime.toFixed(1)}s
|
||||
</span>
|
||||
@@ -349,7 +352,7 @@ const ModalBRollEditor: React.FC<ModalBRollEditorProps> = ({
|
||||
selectedSentence.endTime,
|
||||
selectedSentence.startTime + 0.5,
|
||||
).toFixed(1)}
|
||||
s (按字数自动估算)
|
||||
s
|
||||
</div>
|
||||
</>
|
||||
) : (
|
||||
|
||||
@@ -120,7 +120,7 @@ const PanelCoverAndGenerate: React.FC<PanelCoverAndGenerateProps> = ({
|
||||
wordBreak: "break-word",
|
||||
whiteSpace: "pre-wrap",
|
||||
color: titleConfig.color || "#ffffff",
|
||||
fontSize: `${titleConfig.size}px`,
|
||||
fontSize: `${(titleConfig.size || 28) * 0.55}px`, // 预览容器缩放,与 PanelLipsyncPreview 对齐
|
||||
fontFamily: getFontFamily(titleConfig.font),
|
||||
fontWeight: titleConfig.bold ? "bold" : "normal",
|
||||
fontStyle: titleConfig.italic ? "italic" : "normal",
|
||||
@@ -153,8 +153,8 @@ const PanelCoverAndGenerate: React.FC<PanelCoverAndGenerateProps> = ({
|
||||
} else if (titleConfig.shadow) {
|
||||
style.textShadow = "0 2px 8px rgba(0,0,0,0.7), 0 0 2px rgba(0,0,0,0.5)"
|
||||
} else {
|
||||
// 默认给轻微阴影保证白字在亮背景可读
|
||||
style.textShadow = "0 2px 6px rgba(0,0,0,0.6)"
|
||||
// 无描边无阴影时,不加额外效果(与后端 drawtext 对齐:无 stroke/shadow 则不加)
|
||||
style.textShadow = "none"
|
||||
}
|
||||
|
||||
return style
|
||||
|
||||
@@ -14,8 +14,8 @@ interface PanelLipsyncPreviewProps {
|
||||
onRemoveBRoll: (id: string) => void
|
||||
/** 标题配置(实时叠加预览用) */
|
||||
titleConfig?: AiAvatarTitleConfig
|
||||
/** 标题位置变更回调(拖拽结束时调用) */
|
||||
onTitlePositionChange?: (pos: { pos_x: number; pos_y: number }) => void
|
||||
/** 标题位置变更回调(拖拽结束时调用,发送百分比坐标 + position:"custom") */
|
||||
onTitlePositionChange?: (pos: { pos_x: number; pos_y: number; position: string }) => void
|
||||
}
|
||||
|
||||
const BROLL_MODE_LABEL: Record<BRollSegment["mode"], string> = {
|
||||
@@ -56,11 +56,9 @@ export function PanelLipsyncPreview({
|
||||
const titleOverlayStyle: React.CSSProperties | null = titleConfig?.title
|
||||
? {
|
||||
position: "absolute",
|
||||
left: "50%",
|
||||
transform: "translateX(-50%)",
|
||||
color: titleConfig.color || "#ffffff",
|
||||
fontFamily: titleConfig.font || "思源黑体",
|
||||
fontSize: `${(titleConfig.size || 36) * 0.55}px`, // 预览等比缩
|
||||
fontSize: `${(titleConfig.size || 28) * 0.55}px`,
|
||||
fontWeight: titleConfig.bold ? 700 : 400,
|
||||
fontStyle: titleConfig.italic ? "italic" : "normal",
|
||||
textAlign: "center",
|
||||
@@ -68,11 +66,19 @@ export function PanelLipsyncPreview({
|
||||
padding: "4px 8px",
|
||||
textShadow: titleConfig.shadow ? "0 2px 4px rgba(0,0,0,0.8)" : undefined,
|
||||
WebkitTextStroke: titleConfig.stroke ? "1.5px #000" : undefined,
|
||||
...(titleConfig.position === "top"
|
||||
? { top: 8 }
|
||||
: titleConfig.position === "bottom"
|
||||
? { bottom: 8 }
|
||||
: { top: "50%", transform: "translateX(-50%) translateY(-50%)" }),
|
||||
...(titleConfig.position === "custom" &&
|
||||
titleConfig.pos_x != null &&
|
||||
titleConfig.pos_y != null
|
||||
? {
|
||||
left: `${titleConfig.pos_x}%`,
|
||||
top: `${titleConfig.pos_y}%`,
|
||||
transform: "translateX(-50%) translateY(-50%)",
|
||||
}
|
||||
: titleConfig.position === "top"
|
||||
? { left: "50%", top: 8, transform: "translateX(-50%)" }
|
||||
: titleConfig.position === "bottom"
|
||||
? { left: "50%", bottom: 8, transform: "translateX(-50%)" }
|
||||
: { left: "50%", top: "50%", transform: "translateX(-50%) translateY(-50%)" }),
|
||||
}
|
||||
: null
|
||||
|
||||
@@ -105,7 +111,10 @@ export function PanelLipsyncPreview({
|
||||
const rect = previewContainerRef.current.getBoundingClientRect()
|
||||
const relX = Math.max(0, Math.min(rect.width, e.clientX - rect.left))
|
||||
const relY = Math.max(0, Math.min(rect.height, e.clientY - rect.top))
|
||||
onTitlePositionChange({ pos_x: relX, pos_y: relY })
|
||||
// 发送百分比坐标(0-100),与后端 drawtext 百分比表达式对齐
|
||||
const xpct = Math.round((relX / rect.width) * 1000) / 10
|
||||
const ypct = Math.round((relY / rect.height) * 1000) / 10
|
||||
onTitlePositionChange({ pos_x: xpct, pos_y: ypct, position: "custom" })
|
||||
}
|
||||
;(e.currentTarget as HTMLDivElement).style.cursor = "grab"
|
||||
}
|
||||
|
||||
@@ -44,12 +44,23 @@ export interface LipsyncJob {
|
||||
status: LipsyncStatus
|
||||
progress: number
|
||||
output_video_url: string | null
|
||||
/** 对口型成片总时长(秒),后端返回;用于 B-roll 时间自动估算(#1809 ⑥) */
|
||||
/** 对口型成片总时长(秒),后端返回 */
|
||||
script_text: string
|
||||
output_duration?: number
|
||||
/** 精确句子时间戳(后端基于 TTS 音频静音检测计算) */
|
||||
sentence_timings?: SentenceTiming[] | null
|
||||
error_message: string | null
|
||||
created_at: string
|
||||
}
|
||||
|
||||
/* ── 句子时间戳(后端精确计算) ── */
|
||||
export interface SentenceTiming {
|
||||
index: number
|
||||
text: string
|
||||
start_time: number
|
||||
end_time: number
|
||||
}
|
||||
|
||||
/* ── B-roll 画面插入 ── */
|
||||
export type BRollInsertMode = "fullscreen" | "pip"
|
||||
export type PipPosition = "top-left" | "top-right" | "bottom-left" | "bottom-right"
|
||||
@@ -77,7 +88,7 @@ export interface AiAvatarTitleConfig {
|
||||
shadow: boolean
|
||||
color: string
|
||||
auto_subtitle: boolean
|
||||
/** 自定义位置坐标(position=custom 时生效,像素) */
|
||||
/** 自定义位置坐标(position=custom 时生效,百分比 0-100) */
|
||||
pos_x?: number
|
||||
pos_y?: number
|
||||
}
|
||||
@@ -101,6 +112,7 @@ export interface RenderJob {
|
||||
status: RenderStatus
|
||||
progress: number
|
||||
output_video_url: string | null
|
||||
output_cover_url: string | null
|
||||
error_message: string | null
|
||||
created_at: string
|
||||
}
|
||||
|
||||
@@ -39,7 +39,7 @@ export function buildTitleConfigPayload(cfg: AiAvatarTitleConfig): Record<string
|
||||
text,
|
||||
enabled: true,
|
||||
font: cfg.font || "思源黑体",
|
||||
font_size: Math.round(cfg.size) || 36,
|
||||
font_size: Math.round(cfg.size) || 28,
|
||||
font_color: cfg.color || "#ffffff",
|
||||
position,
|
||||
bold: !!cfg.bold,
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
/**
|
||||
* AI数字人 — 文案分句 & B-roll 时间自动估算(#1809 ⑤⑥)
|
||||
* AI数字人 — 文案分句工具
|
||||
*
|
||||
* 分句规则与后端 _split_script_into_sentences 保持一致。
|
||||
* 时间戳由后端基于 TTS 音频静音检测精确计算,前端不再做字数比例估算。
|
||||
*/
|
||||
import type { SentenceTiming } from "../types"
|
||||
|
||||
export interface ScriptSentence {
|
||||
/** 句子序号(从 0 开始,对应提交给后端的 script_segment_index) */
|
||||
@@ -9,21 +13,21 @@ export interface ScriptSentence {
|
||||
text: string
|
||||
/** 句子字数(按中文/字符计,去除空白) */
|
||||
charCount: number
|
||||
/** 累计起始字数(用于时间估算) */
|
||||
/** 累计起始字数 */
|
||||
startChar: number
|
||||
/** 估算的对口型视频内起始时间(秒) */
|
||||
/** 精确起始时间(秒),来自后端 sentence_timings;无数据时为 0 */
|
||||
startTime: number
|
||||
/** 估算的对口型视频内结束时间(秒) */
|
||||
/** 精确结束时间(秒),来自后端 sentence_timings;无数据时为 0 */
|
||||
endTime: number
|
||||
}
|
||||
|
||||
/**
|
||||
* 按句号/问号/感叹号/分号/换行分句(兼容中英文标点)。
|
||||
* 空文案返回空数组。时间按「该句字数 ÷ 全文总字数 × 口播总时长」线性估算。
|
||||
* 时间戳从后端 sentence_timings 获取(精确);若无则返回 0(由调用方降级处理)。
|
||||
*/
|
||||
export function splitScriptIntoSentences(
|
||||
scriptText: string,
|
||||
outputDuration: number,
|
||||
sentenceTimings?: SentenceTiming[] | null,
|
||||
): ScriptSentence[] {
|
||||
const text = (scriptText || "").trim()
|
||||
if (!text) return []
|
||||
@@ -33,30 +37,25 @@ export function splitScriptIntoSentences(
|
||||
.map((part) => part.trim())
|
||||
.filter((part) => part.length > 0)
|
||||
|
||||
const totalChars = rawParts.reduce((sum, part) => sum + part.replace(/\s/g, "").length, 0)
|
||||
const duration = outputDuration > 0 ? outputDuration : 0
|
||||
|
||||
const sentences: ScriptSentence[] = []
|
||||
let accChar = 0
|
||||
rawParts.forEach((part, i) => {
|
||||
const charCount = part.replace(/\s/g, "").length
|
||||
const startTime = duration > 0 && totalChars > 0 ? (accChar / totalChars) * duration : 0
|
||||
const endTime =
|
||||
duration > 0 && totalChars > 0 ? ((accChar + charCount) / totalChars) * duration : 0
|
||||
// 从后端精确时间戳获取;无数据时返回 0
|
||||
const timing = sentenceTimings?.[i]
|
||||
const startTime = timing?.start_time ?? 0
|
||||
const endTime = timing?.end_time ?? 0
|
||||
|
||||
sentences.push({
|
||||
index: i,
|
||||
text: part,
|
||||
charCount,
|
||||
startChar: accChar,
|
||||
startTime: round1(startTime),
|
||||
endTime: round1(endTime),
|
||||
startTime,
|
||||
endTime,
|
||||
})
|
||||
accChar += charCount
|
||||
})
|
||||
|
||||
return sentences
|
||||
}
|
||||
|
||||
function round1(n: number): number {
|
||||
return Math.round(n * 10) / 10
|
||||
}
|
||||
|
||||
@@ -703,6 +703,9 @@ class LipsyncJobModel(Base):
|
||||
error_message = Column(Text, nullable=False, default="")
|
||||
error_code = Column(String(100), nullable=False, default="")
|
||||
|
||||
# 精确句子时间戳(TTS 合成后由 silencedetect 计算,用于 B-roll 精确定位)
|
||||
sentence_timings = Column(JSON, nullable=True) # list[{index,text,start_time,end_time}]
|
||||
|
||||
# 时间戳
|
||||
submitted_at = Column(DateTime, nullable=True)
|
||||
completed_at = Column(DateTime, nullable=True)
|
||||
|
||||
@@ -377,26 +377,30 @@ def _append_audio_concat(parts: list[str], clip_chains: list[ClipFilterChain]) -
|
||||
|
||||
# ── 标题 drawtext 滤镜构建(#1789)─────────────────────────────────────────────
|
||||
|
||||
# drawtext 字体搜索路径:按优先级列出常见安装位置
|
||||
# 服务器使用 Noto Sans SC(思源黑体)作为默认字体
|
||||
# drawtext 字体搜索路径:按优先级从高到低排列
|
||||
# 服务器使用 Noto Sans SC(思源黑体)作为默认字体。
|
||||
# - NotoSansSC-VF.ttf 是 worker-base.Dockerfile 中 COPY 的 VF 字体(含所有字重,无 Mono 变体),优先级最高
|
||||
# - .ttc 系列为 fonts-noto-cjk 包预装字体(Dockerfile 已删除含 Mono 变体的旧 .ttc,存在时作为 fallback)
|
||||
# - DejaVuSans 仅含拉丁字符不支持中文,已移除
|
||||
DRAWTEXT_FONT_SEARCH_PATHS: list[str] = [
|
||||
"/usr/share/fonts/opentype/noto/NotoSansSC-VF.ttf",
|
||||
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
|
||||
"/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
|
||||
"/usr/share/fonts/noto-cjk/NotoSansCJK-Regular.ttc",
|
||||
"/usr/share/fonts/google-noto-cjk/NotoSansCJK-Regular.ttc",
|
||||
"/usr/share/fonts/truetype/noto/NotoSansSC-Regular.ttf",
|
||||
"/usr/share/fonts/noto/NotoSansSC-Regular.ttf",
|
||||
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
||||
]
|
||||
|
||||
# 前端字体名 → drawtext 字体搜索关键字
|
||||
# 前端字体名 → drawtext 字体搜索关键字(匹配 DRAWTEXT_FONT_SEARCH_PATHS 中的文件名关键字)
|
||||
DRAWTEXT_FONT_MAP: dict[str, str] = {
|
||||
"思源黑体": "NotoSansCJK",
|
||||
"思源黑体": "NotoSansSC",
|
||||
"思源宋体": "NotoSerifCJK",
|
||||
"苹方": "NotoSansCJK",
|
||||
"PingFang": "NotoSansCJK",
|
||||
"微软雅黑": "NotoSansCJK",
|
||||
"苹方": "NotoSansSC",
|
||||
"PingFang": "NotoSansSC",
|
||||
"微软雅黑": "NotoSansSC",
|
||||
"楷体": "NotoSerifCJK",
|
||||
"华康俪金黑": "NotoSansCJK",
|
||||
"华康俪金黑": "NotoSansSC",
|
||||
}
|
||||
|
||||
|
||||
@@ -477,13 +481,13 @@ def build_title_drawtext_filter(
|
||||
|
||||
# ── 样式参数 ──
|
||||
font_name = title_config.get("font") or title_config.get("font_preset") or "思源黑体"
|
||||
font_size = int(title_config.get("font_size") or title_config.get("size") or 36)
|
||||
font_size = int(title_config.get("font_size") or title_config.get("size") or 28)
|
||||
font_color = title_config.get("font_color") or title_config.get("color") or "#ffffff"
|
||||
# 去掉 # 前缀(drawtext 用纯 hex 或颜色名)
|
||||
if font_color.startswith("#"):
|
||||
font_color = font_color[1:]
|
||||
|
||||
position = title_config.get("position", "top")
|
||||
position = title_config.get("position") or "bottom"
|
||||
bold = bool(title_config.get("bold", True))
|
||||
stroke = title_config.get("stroke")
|
||||
shadow = title_config.get("shadow")
|
||||
@@ -504,26 +508,30 @@ def build_title_drawtext_filter(
|
||||
params.append(f"fontsize={font_size}")
|
||||
params.append(f"fontcolor={font_color}")
|
||||
|
||||
# 粗体:bold 在 drawtext 中通过 font 的 Bold 变体实现
|
||||
# 若字体有 Bold 变体可用 fontfont=bold;否则通过 borderw 模拟
|
||||
if bold:
|
||||
# 使用 font 参数尝试加载 Bold 变体(Noto Sans SC 有 Bold 变体文件)
|
||||
params.append("font=bold")
|
||||
# 粗体:drawtext 没有独立的 bold 参数,通过加大 borderw 模拟视觉粗体效果。
|
||||
# 注意:不能使用 `font=bold`——FFmpeg drawtext 的 font 参数需要 fontconfig 能解析的
|
||||
# 字体族名,而 "bold" 不是合法族名,会导致整个 filter_complex 解析失败(exit code 234)。
|
||||
# 当用户未显式配置描边宽度时,bold 模式自动将 borderw 提升到 3 以模拟粗体。
|
||||
|
||||
# 描边(borderw 需要 libfreetype 支持)
|
||||
# 粗体无显式描边时,自动用 borderw=3 + 近色描边模拟粗体;显式 stroke 按用户配置走
|
||||
border_width = 0
|
||||
border_color = "000000"
|
||||
if stroke:
|
||||
if isinstance(stroke, bool):
|
||||
border_width = 2
|
||||
border_color = "black"
|
||||
border_color = "000000"
|
||||
elif isinstance(stroke, dict):
|
||||
border_width = int(stroke.get("width", 2)) if stroke.get("enabled", True) else 0
|
||||
border_color = (stroke.get("color") or "#000000").lstrip("#")
|
||||
else:
|
||||
border_width = 0
|
||||
border_color = "black"
|
||||
if border_width > 0:
|
||||
params.append(f"borderw={border_width}")
|
||||
params.append(f"bordercolor={border_color}")
|
||||
if stroke.get("enabled", True):
|
||||
border_width = int(stroke.get("width", 2))
|
||||
border_color = (stroke.get("color") or "#000000").lstrip("#")
|
||||
elif bold:
|
||||
# 粗体模式且未配描边:加大描边宽度模拟粗体效果
|
||||
border_width = 3
|
||||
border_color = font_color # 用字体同色描边,视觉上加粗字形而非黑边
|
||||
if border_width > 0:
|
||||
params.append(f"borderw={border_width}")
|
||||
params.append(f"bordercolor={border_color}")
|
||||
|
||||
# 阴影(shadowcolor + shadowx/y)
|
||||
if shadow:
|
||||
@@ -548,8 +556,13 @@ def build_title_drawtext_filter(
|
||||
and not isinstance(pos_x, bool)
|
||||
and not isinstance(pos_y, bool)
|
||||
):
|
||||
params.append(f"x={int(pos_x)}")
|
||||
params.append(f"y={int(pos_y)}")
|
||||
# pos_x/pos_y 为百分比坐标(0-100),转换为 drawtext 表达式
|
||||
# 例如 pos_x=50 → x=(w-text_w)*0.50(水平居中偏50%)
|
||||
# pos_y=30 → y=(h-text_h)*0.30
|
||||
pct_x = max(0.0, min(100.0, float(pos_x))) / 100.0
|
||||
pct_y = max(0.0, min(100.0, float(pos_y))) / 100.0
|
||||
params.append(f"x=(w-text_w)*{pct_x:.4f}")
|
||||
params.append(f"y=(h-text_h)*{pct_y:.4f}")
|
||||
else:
|
||||
# 三档预设位置:top / center / bottom
|
||||
# x 始终水平居中:(w-text_w)/2
|
||||
@@ -573,7 +586,7 @@ def build_broll_overlay_filter(
|
||||
video_duration: float,
|
||||
output_width: int = DEFAULT_OUTPUT_WIDTH,
|
||||
output_height: int = DEFAULT_OUTPUT_HEIGHT,
|
||||
) -> str:
|
||||
) -> tuple[str, str | None]:
|
||||
"""构建 B-roll 叠加滤镜链。
|
||||
|
||||
支持两种模式:
|
||||
@@ -581,121 +594,182 @@ def build_broll_overlay_filter(
|
||||
- pip: 在对口型视频上叠加画中画 B-roll
|
||||
|
||||
Args:
|
||||
b_roll_segments: B-roll 片段配置列表
|
||||
b_roll_segments: B-roll 片段配置列表(原始顺序,决定 FFmpeg -i 输入顺序)
|
||||
video_duration: 对口型视频总时长(秒)
|
||||
output_width: 输出宽度
|
||||
output_height: 输出高度
|
||||
output_width: 输出宽度(默认 1280;AI 数字人竖屏传 720)
|
||||
output_height: 输出高度(默认 720;AI 数字人竖屏传 1280)
|
||||
|
||||
Returns:
|
||||
FFmpeg filter_complex 滤镜字符串片段
|
||||
(filter_complex_str, final_label)
|
||||
- filter_complex_str: filter_complex 片段字符串(末尾无分号)
|
||||
- final_label: 最终输出 pad 标签名,如 "vout";无 B-roll 时返回 None
|
||||
"""
|
||||
if not b_roll_segments:
|
||||
return ""
|
||||
return "", None
|
||||
|
||||
# 建立原始列表下标 → FFmpeg 输入下标的映射:
|
||||
# cmd 中 [0:v] 是主视频,随后按 b_roll_segments 原始顺序追加 -i,
|
||||
# 因此第 i 个 segment 的输入是 [{i+1}:v]
|
||||
def _input_label(seg: dict[str, Any]) -> str:
|
||||
# seg 必须来自 b_roll_segments;通过 id() 在原列表中查找
|
||||
for i, s in enumerate(b_roll_segments):
|
||||
if s is seg:
|
||||
return f"[{i + 1}:v]"
|
||||
# fallback: 找不到时不应发生,保守返回
|
||||
return "[1:v]"
|
||||
|
||||
parts: list[str] = []
|
||||
sorted_segments = sorted(b_roll_segments, key=lambda s: s.get("start_time", 0))
|
||||
|
||||
# 按模式分组处理
|
||||
# 按模式分组
|
||||
fullscreen_segments = [s for s in sorted_segments if s.get("mode") == "fullscreen"]
|
||||
pip_segments = [s for s in sorted_segments if s.get("mode") == "pip"]
|
||||
|
||||
final_label = None
|
||||
|
||||
# ── fullscreen 模式: 切分 + concat ──
|
||||
if fullscreen_segments:
|
||||
parts.append(_build_fullscreen_filters(fullscreen_segments, video_duration, output_width, output_height))
|
||||
fs_filter, fs_label = _build_fullscreen_filters(
|
||||
fullscreen_segments, b_roll_segments, video_duration, output_width, output_height, _input_label
|
||||
)
|
||||
parts.append(fs_filter)
|
||||
final_label = fs_label
|
||||
else:
|
||||
fs_label = None
|
||||
|
||||
# ── pip 模式: overlay 滤镜 ──
|
||||
if pip_segments:
|
||||
for idx, seg in enumerate(pip_segments):
|
||||
start = seg.get("start_time", 0)
|
||||
end = seg.get("end_time", video_duration)
|
||||
scale = seg.get("pip_scale", 0.3)
|
||||
position = seg.get("pip_position", "bottom_right")
|
||||
|
||||
pip_w = int(output_width * scale)
|
||||
pip_h = int(output_height * scale)
|
||||
|
||||
# 位置映射
|
||||
pos_map = {
|
||||
"top_left": "10:10",
|
||||
"top_right": "W-w-10:10",
|
||||
"bottom_left": "10:H-h-10",
|
||||
"bottom_right": "W-w-10:H-h-10",
|
||||
"center": "(W-w)/2:(H-h)/2",
|
||||
}
|
||||
pos_expr = pos_map.get(position, pos_map["bottom_right"])
|
||||
|
||||
broll_input_idx = len(sorted_segments) # placeholder for input index
|
||||
parts.append(
|
||||
f"[{broll_input_idx + idx}:v]scale={pip_w}:{pip_h}," f"enable='between(t,{start},{end})'[pip{idx}];"
|
||||
)
|
||||
# overlay onto main stream
|
||||
if idx == 0:
|
||||
base_label = "[vout]" if fullscreen_segments else "[0:v]"
|
||||
else:
|
||||
base_label = f"[pip{idx - 1}]"
|
||||
parts.append(f"{base_label}[pip{idx}]overlay={pos_expr}:enable='between(t,{start},{end})'[vout{idx}];")
|
||||
pip_filter, pip_label = _build_pip_filters(
|
||||
pip_segments, output_width, output_height, _input_label, base_label=fs_label
|
||||
)
|
||||
parts.append(pip_filter)
|
||||
final_label = pip_label
|
||||
|
||||
result = "".join(parts)
|
||||
# 清理末尾多余分号
|
||||
if result.endswith(";"):
|
||||
result = result[:-1]
|
||||
return result
|
||||
return result, final_label
|
||||
|
||||
|
||||
def _build_fullscreen_filters(
|
||||
segments: list[dict[str, Any]],
|
||||
sorted_fs_segments: list[dict[str, Any]],
|
||||
all_segments: list[dict[str, Any]],
|
||||
video_duration: float,
|
||||
output_width: int,
|
||||
output_height: int,
|
||||
) -> str:
|
||||
"""构建 fullscreen 模式的切分 + concat 滤镜.
|
||||
input_label_fn,
|
||||
) -> tuple[str, str]:
|
||||
"""构建 fullscreen 模式的切分 + concat 滤镜。
|
||||
|
||||
将对口型视频按 B-roll 时间段切分,然后用 concat 拼接 B-roll 片段。
|
||||
将主视频按 B-roll 时间段切分,然后用 concat 拼接主视频片段和 B-roll 片段。
|
||||
|
||||
Returns:
|
||||
(filter_str, final_label) 其中 final_label 是 concat 输出的 pad 标签
|
||||
"""
|
||||
parts: list[str] = []
|
||||
prev_end = 0.0
|
||||
|
||||
for idx, seg in enumerate(segments):
|
||||
# 注意:这里的 idx 是 sorted_fs_segments 中的下标;
|
||||
# 实际 FFmpeg 输入下标必须通过 input_label_fn 查询
|
||||
for idx, seg in enumerate(sorted_fs_segments):
|
||||
start = seg.get("start_time", 0)
|
||||
end = seg.get("end_time", video_duration)
|
||||
|
||||
# 保持原视频片段(B-roll 之前的部分)
|
||||
# 主视频片段(B-roll 之前)
|
||||
if prev_end < start:
|
||||
parts.append(f"[0:v]trim=start={prev_end}:end={start},setpts=PTS-STARTPTS[main{idx}];")
|
||||
|
||||
# B-roll 片段:缩放至目标分辨率
|
||||
# B-roll 片段:缩放到输出分辨率并裁到对应时长
|
||||
in_lbl = input_label_fn(seg)
|
||||
parts.append(
|
||||
f"[{idx + 1}:v]scale={output_width}:{output_height}"
|
||||
f"{in_lbl}scale={output_width}:{output_height}"
|
||||
f":force_original_aspect_ratio=decrease,"
|
||||
f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2,"
|
||||
f"trim=start=0:end={end - start},setpts=PTS-STARTPTS[br{idx}];"
|
||||
)
|
||||
prev_end = end
|
||||
|
||||
# 尾部片段
|
||||
# 尾部主视频片段
|
||||
if prev_end < video_duration:
|
||||
last_idx = len(segments)
|
||||
last_idx = len(sorted_fs_segments)
|
||||
parts.append(f"[0:v]trim=start={prev_end}:end={video_duration},setpts=PTS-STARTPTS[main{last_idx}];")
|
||||
|
||||
# concat 所有片段
|
||||
segment_labels = []
|
||||
for idx in range(len(segments)):
|
||||
start = segments[idx].get("start_time", 0)
|
||||
if (idx == 0 and segments[0].get("start_time", 0) > 0) or idx > 0:
|
||||
prev_end_prev = segments[idx - 1].get("end_time", 0) if idx > 0 else 0
|
||||
if prev_end_prev < start:
|
||||
segment_labels.append(f"[main{idx}]")
|
||||
segment_labels: list[str] = []
|
||||
for idx, seg in enumerate(sorted_fs_segments):
|
||||
start = seg.get("start_time", 0)
|
||||
# 每段 B-roll 之前是否有主视频片段?
|
||||
has_main_before = (idx == 0 and start > 0) or (
|
||||
idx > 0 and sorted_fs_segments[idx - 1].get("end_time", 0) < start
|
||||
)
|
||||
if has_main_before:
|
||||
segment_labels.append(f"[main{idx}]")
|
||||
segment_labels.append(f"[br{idx}]")
|
||||
|
||||
if prev_end < video_duration:
|
||||
segment_labels.append(f"[main{len(segments)}]")
|
||||
segment_labels.append(f"[main{len(sorted_fs_segments)}]")
|
||||
|
||||
final_lbl = "vout_fs"
|
||||
n = len(segment_labels)
|
||||
if n > 0:
|
||||
concat_inputs = "".join(segment_labels)
|
||||
parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[vout];")
|
||||
parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[{final_lbl}];")
|
||||
|
||||
return "".join(parts)
|
||||
return "".join(parts), final_lbl
|
||||
|
||||
|
||||
def _build_pip_filters(
|
||||
pip_segments: list[dict[str, Any]],
|
||||
output_width: int,
|
||||
output_height: int,
|
||||
input_label_fn,
|
||||
base_label: str | None,
|
||||
) -> tuple[str, str]:
|
||||
"""构建 PIP(画中画)overlay 滤镜链。
|
||||
|
||||
Args:
|
||||
pip_segments: 按时间排序的 pip 片段
|
||||
output_width: 输出宽度
|
||||
output_height: 输出高度
|
||||
input_label_fn: 片段 → 输入标签的映射函数
|
||||
base_label: 前序滤镜链输出的标签(如 fullscreen 的 vout_fs),为 None 则基于 [0:v]
|
||||
|
||||
Returns:
|
||||
(filter_str, final_label)
|
||||
"""
|
||||
parts: list[str] = []
|
||||
cur_label = base_label # 当前叠加到的标签
|
||||
|
||||
pos_map = {
|
||||
"top_left": "10:10",
|
||||
"top_right": "W-w-10:10",
|
||||
"bottom_left": "10:H-h-10",
|
||||
"bottom_right": "W-w-10:H-h-10",
|
||||
"center": "(W-w)/2:(H-h)/2",
|
||||
}
|
||||
|
||||
for idx, seg in enumerate(pip_segments):
|
||||
start = seg.get("start_time", 0)
|
||||
end = seg.get("end_time", 0)
|
||||
scale = seg.get("pip_scale", 0.3)
|
||||
position = seg.get("pip_position", "bottom_right")
|
||||
pos_expr = pos_map.get(position, pos_map["bottom_right"])
|
||||
|
||||
pip_w = max(1, int(output_width * scale))
|
||||
pip_h = max(1, int(output_height * scale))
|
||||
enable_expr = f"enable='between(t,{start},{end})'"
|
||||
|
||||
in_lbl = input_label_fn(seg)
|
||||
pip_scaled = f"pip{idx}"
|
||||
parts.append(f"{in_lbl}scale={pip_w}:{pip_h},{enable_expr}[{pip_scaled}];")
|
||||
|
||||
# overlay onto the current base
|
||||
base = f"[{cur_label}]" if cur_label else "[0:v]"
|
||||
out_lbl = f"vout_pip{idx}" if idx < len(pip_segments) - 1 else "vout"
|
||||
parts.append(f"{base}[{pip_scaled}]overlay={pos_expr}:{enable_expr}[{out_lbl}];")
|
||||
cur_label = out_lbl
|
||||
|
||||
return "".join(parts), cur_label or "vout"
|
||||
|
||||
|
||||
def build_cover_extract_command(
|
||||
|
||||
@@ -258,8 +258,9 @@ class TestBrollOverlayFilter:
|
||||
def test_empty_segments_returns_empty(self):
|
||||
from packages.domain.video_filter_builder import build_broll_overlay_filter
|
||||
|
||||
result = build_broll_overlay_filter([], 30.0)
|
||||
result, label = build_broll_overlay_filter([], 30.0)
|
||||
assert result == ""
|
||||
assert label is None
|
||||
|
||||
def test_pip_mode_generates_overlay(self):
|
||||
from packages.domain.video_filter_builder import build_broll_overlay_filter
|
||||
@@ -275,8 +276,9 @@ class TestBrollOverlayFilter:
|
||||
"pip_scale": 0.3,
|
||||
}
|
||||
]
|
||||
result = build_broll_overlay_filter(segments, 30.0)
|
||||
result, label = build_broll_overlay_filter(segments, 30.0)
|
||||
assert "overlay" in result or "scale=" in result
|
||||
assert label == "vout"
|
||||
|
||||
def test_fullscreen_mode_generates_concat(self):
|
||||
from packages.domain.video_filter_builder import build_broll_overlay_filter
|
||||
@@ -290,8 +292,9 @@ class TestBrollOverlayFilter:
|
||||
"end_time": 10.0,
|
||||
}
|
||||
]
|
||||
result = build_broll_overlay_filter(segments, 30.0)
|
||||
result, label = build_broll_overlay_filter(segments, 30.0)
|
||||
assert "trim" in result or "concat" in result
|
||||
assert label == "vout_fs"
|
||||
|
||||
def test_cover_extract_command(self):
|
||||
from packages.domain.video_filter_builder import build_cover_extract_command
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Tests for sentence timing functions in lipsync_tts."""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from apps.api.app.tasks.lipsync_tts import (
|
||||
_compute_sentence_timings,
|
||||
_estimate_sentence_timings_by_chars,
|
||||
_split_script_into_sentences,
|
||||
)
|
||||
|
||||
|
||||
class TestSplitScriptIntoSentences(unittest.TestCase):
|
||||
"""Tests for _split_script_into_sentences."""
|
||||
|
||||
def test_empty_string(self):
|
||||
self.assertEqual(_split_script_into_sentences(""), [])
|
||||
|
||||
def test_none(self):
|
||||
self.assertEqual(_split_script_into_sentences(None), [])
|
||||
|
||||
def test_whitespace_only(self):
|
||||
self.assertEqual(_split_script_into_sentences(" \n "), [])
|
||||
|
||||
def test_single_sentence(self):
|
||||
self.assertEqual(_split_script_into_sentences("你好世界。"), ["你好世界"])
|
||||
|
||||
def test_multiple_sentences_chinese(self):
|
||||
result = _split_script_into_sentences("第一句。第二句!第三句?")
|
||||
self.assertEqual(result, ["第一句", "第二句", "第三句"])
|
||||
|
||||
def test_english_punctuation(self):
|
||||
result = _split_script_into_sentences("Hello World! How are you?")
|
||||
self.assertEqual(result, ["Hello World", "How are you"])
|
||||
|
||||
def test_semicolons(self):
|
||||
result = _split_script_into_sentences("第一部分;第二部分;第三部分")
|
||||
self.assertEqual(result, ["第一部分", "第二部分", "第三部分"])
|
||||
|
||||
def test_newlines(self):
|
||||
result = _split_script_into_sentences("第一行\n第二行\n第三行")
|
||||
self.assertEqual(result, ["第一行", "第二行", "第三行"])
|
||||
|
||||
def test_no_trailing_punctuation(self):
|
||||
result = _split_script_into_sentences("没有标点的句子")
|
||||
self.assertEqual(result, ["没有标点的句子"])
|
||||
|
||||
|
||||
class TestEstimateSentenceTimingsByChars(unittest.TestCase):
|
||||
"""Tests for _estimate_sentence_timings_by_chars."""
|
||||
|
||||
def test_empty_sentences(self):
|
||||
self.assertEqual(_estimate_sentence_timings_by_chars([], 10.0), [])
|
||||
|
||||
def test_zero_duration(self):
|
||||
self.assertEqual(_estimate_sentence_timings_by_chars(["hello"], 0), [])
|
||||
|
||||
def test_negative_duration(self):
|
||||
self.assertEqual(_estimate_sentence_timings_by_chars(["hello"], -5.0), [])
|
||||
|
||||
def test_single_sentence(self):
|
||||
result = _estimate_sentence_timings_by_chars(["hello"], 10.0)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
self.assertAlmostEqual(result[0]["end_time"], 10.0)
|
||||
|
||||
def test_two_equal_sentences(self):
|
||||
result = _estimate_sentence_timings_by_chars(["你好", "世界"], 10.0)
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
self.assertAlmostEqual(result[0]["end_time"], 5.0)
|
||||
self.assertAlmostEqual(result[1]["start_time"], 5.0)
|
||||
self.assertAlmostEqual(result[1]["end_time"], 10.0)
|
||||
|
||||
def test_unequal_char_distribution(self):
|
||||
result = _estimate_sentence_timings_by_chars(["ABCD", "EF"], 9.0)
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
self.assertAlmostEqual(result[0]["end_time"], 6.0) # 4/6 * 9 = 6
|
||||
self.assertAlmostEqual(result[1]["start_time"], 6.0)
|
||||
self.assertAlmostEqual(result[1]["end_time"], 9.0)
|
||||
|
||||
def test_timing_structure(self):
|
||||
result = _estimate_sentence_timings_by_chars(["句子一", "句子二"], 6.0)
|
||||
for item in result:
|
||||
self.assertIn("index", item)
|
||||
self.assertIn("text", item)
|
||||
self.assertIn("start_time", item)
|
||||
self.assertIn("end_time", item)
|
||||
|
||||
|
||||
class TestComputeSentenceTimings(unittest.TestCase):
|
||||
"""Tests for _compute_sentence_timings."""
|
||||
|
||||
def test_empty_script_returns_empty(self):
|
||||
self.assertEqual(_compute_sentence_timings(b"fake_audio", "", 10.0), [])
|
||||
|
||||
def test_none_script_returns_empty(self):
|
||||
self.assertEqual(_compute_sentence_timings(b"fake_audio", None, 10.0), [])
|
||||
|
||||
@patch("os.unlink")
|
||||
@patch.object(tempfile, "NamedTemporaryFile")
|
||||
@patch.object(subprocess, "run")
|
||||
def test_silence_detection_insufficient_fallback(self, mock_run, mock_tmpfile, mock_unlink):
|
||||
"""When silence detection finds too few points, fallback to char estimation."""
|
||||
mock_run.return_value = MagicMock(stderr="", returncode=0)
|
||||
mock_tmp = MagicMock()
|
||||
mock_tmp.name = "/tmp/fake.mp3"
|
||||
mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
|
||||
mock_tmp.__exit__ = MagicMock(return_value=False)
|
||||
mock_tmpfile.return_value = mock_tmp
|
||||
|
||||
result = _compute_sentence_timings(b"fake_audio", "第一句。第二句。第三句。", 10.0)
|
||||
|
||||
# Should fallback to char estimation with 3 sentences
|
||||
self.assertEqual(len(result), 3)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
|
||||
@patch("os.unlink")
|
||||
@patch.object(tempfile, "NamedTemporaryFile")
|
||||
@patch.object(subprocess, "run")
|
||||
def test_silence_detection_with_enough_points(self, mock_run, mock_tmpfile, mock_unlink):
|
||||
"""When silence detection finds enough points, use them for boundaries."""
|
||||
mock_run.return_value = MagicMock(
|
||||
stderr="[silencedetect] silence_end: 3.5 | silence_duration: 0.4\n"
|
||||
"[silencedetect] silence_end: 7.0 | silence_duration: 0.3\n",
|
||||
returncode=0,
|
||||
)
|
||||
mock_tmp = MagicMock()
|
||||
mock_tmp.name = "/tmp/fake.mp3"
|
||||
mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
|
||||
mock_tmp.__exit__ = MagicMock(return_value=False)
|
||||
mock_tmpfile.return_value = mock_tmp
|
||||
|
||||
result = _compute_sentence_timings(b"fake_audio", "第一句。第二句。第三句。", 10.0)
|
||||
|
||||
self.assertEqual(len(result), 3)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
self.assertAlmostEqual(result[0]["end_time"], 3.5)
|
||||
self.assertAlmostEqual(result[1]["start_time"], 3.5)
|
||||
self.assertAlmostEqual(result[1]["end_time"], 7.0)
|
||||
self.assertAlmostEqual(result[2]["start_time"], 7.0)
|
||||
self.assertAlmostEqual(result[2]["end_time"], 10.0)
|
||||
|
||||
@patch("os.unlink")
|
||||
@patch.object(tempfile, "NamedTemporaryFile")
|
||||
@patch.object(subprocess, "run")
|
||||
def test_ffmpeg_exception_fallback(self, mock_run, mock_tmpfile, mock_unlink):
|
||||
"""When ffmpeg raises an exception, fallback to char estimation."""
|
||||
mock_run.side_effect = Exception("ffmpeg not found")
|
||||
mock_tmp = MagicMock()
|
||||
mock_tmp.name = "/tmp/fake.mp3"
|
||||
mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
|
||||
mock_tmp.__exit__ = MagicMock(return_value=False)
|
||||
mock_tmpfile.return_value = mock_tmp
|
||||
|
||||
result = _compute_sentence_timings(b"fake_audio", "句子一。句子二。", 6.0)
|
||||
|
||||
# Should fallback to char estimation
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertAlmostEqual(result[0]["start_time"], 0.0)
|
||||
self.assertAlmostEqual(result[0]["end_time"], 3.0)
|
||||
self.assertAlmostEqual(result[1]["start_time"], 3.0)
|
||||
self.assertAlmostEqual(result[1]["end_time"], 6.0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -902,9 +902,10 @@ class TestResolveFontPath(unittest.TestCase):
|
||||
|
||||
@patch("os.path.isfile")
|
||||
def test_unknown_font_fallback(self, mock_isfile):
|
||||
mock_isfile.side_effect = lambda p: "DejaVu" in p
|
||||
# DejaVuSans 已从 fallback 列表移除(不支持 CJK),用 VF 路径模拟
|
||||
mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p
|
||||
result = _resolve_font_path("UnknownFont")
|
||||
self.assertIn("DejaVu", result)
|
||||
self.assertIn("NotoSansSC-VF", result)
|
||||
|
||||
@patch("os.path.isfile")
|
||||
def test_no_fonts_available(self, mock_isfile):
|
||||
@@ -927,9 +928,11 @@ class TestResolveFontPath(unittest.TestCase):
|
||||
|
||||
@patch("os.path.isfile")
|
||||
def test_font_fallback_skips_nonexistent(self, mock_isfile):
|
||||
mock_isfile.side_effect = lambda p: "DejaVu" in p
|
||||
# 所有中文字体路径都不存在时,fallback 返回第一个存在的文件;
|
||||
# DejaVuSans 已从列表移除(不支持 CJK),使用 VF 字体路径模拟存在文件
|
||||
mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p
|
||||
result = _resolve_font_path("不存在字体")
|
||||
self.assertIn("DejaVu", result)
|
||||
self.assertIn("NotoSansSC-VF", result)
|
||||
|
||||
|
||||
class TestDrawtextFontFileIncluded(unittest.TestCase):
|
||||
@@ -1028,6 +1031,14 @@ class TestDrawtextBoldFalse(unittest.TestCase):
|
||||
self.assertIsNotNone(result)
|
||||
self.assertNotIn("font=bold", result)
|
||||
|
||||
def test_bold_true_does_not_use_font_bold_param(self):
|
||||
"""粗体模式不得使用 `font=bold`——该参数无效,会导致 filter_complex 解析失败(exit 234)。"""
|
||||
result = build_title_drawtext_filter({"text": "标题", "bold": True})
|
||||
self.assertIsNotNone(result)
|
||||
self.assertNotIn("font=bold", result)
|
||||
# 粗体应通过 borderw 实现
|
||||
self.assertIn("borderw=", result)
|
||||
|
||||
|
||||
class TestDrawtextPositionBranches(unittest.TestCase):
|
||||
"""位置相关分支覆盖。"""
|
||||
@@ -1054,12 +1065,23 @@ class TestDrawtextPositionBranches(unittest.TestCase):
|
||||
self.assertIn("y=h-text_h-50", result)
|
||||
|
||||
@patch("packages.domain.video_filter_builder._resolve_font_path")
|
||||
def test_position_custom_with_float_coords(self, mock_font):
|
||||
def test_position_custom_with_percentage_coords(self, mock_font):
|
||||
"""自定义位置:百分比坐标转换为 drawtext 表达式."""
|
||||
mock_font.return_value = ""
|
||||
# pos_x=50, pos_y=30 → x=(w-text_w)*0.5000, y=(h-text_h)*0.3000
|
||||
result = build_title_drawtext_filter({"text": "标题", "position": "custom", "pos_x": 50, "pos_y": 30})
|
||||
self.assertIsNotNone(result)
|
||||
self.assertIn("x=(w-text_w)*0.5000", result)
|
||||
self.assertIn("y=(h-text_h)*0.3000", result)
|
||||
|
||||
@patch("packages.domain.video_filter_builder._resolve_font_path")
|
||||
def test_position_custom_clamped_to_100(self, mock_font):
|
||||
"""自定义位置:超过100的坐标被截断到100%."""
|
||||
mock_font.return_value = ""
|
||||
result = build_title_drawtext_filter({"text": "标题", "position": "custom", "pos_x": 100.7, "pos_y": 200.3})
|
||||
self.assertIsNotNone(result)
|
||||
self.assertIn("x=100", result)
|
||||
self.assertIn("y=200", result)
|
||||
self.assertIn("x=(w-text_w)*1.0000", result)
|
||||
self.assertIn("y=(h-text_h)*1.0000", result)
|
||||
|
||||
@patch("packages.domain.video_filter_builder._resolve_font_path")
|
||||
def test_position_custom_bool_coords_fallback(self, mock_font):
|
||||
|
||||
Reference in New Issue
Block a user