Compare commits

...

4 Commits

Author SHA1 Message Date
frontend-dev 938ef0b8cc fix(ai-avatar): 端到端一致性修复(标题/封面/B-roll位置/B-roll时长) (#1869)
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Check push changed paths (push) Successful in 4s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 21s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 24s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 51s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 3m26s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 3m37s
CI/CD Pipeline / Validate - Style (push) Successful in 4m6s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 4m29s
CI/CD Pipeline / Validate - Security (push) Successful in 8m25s
CI/CD Pipeline / Unit Tests (push) Successful in 8m59s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 18m30s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m43s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 3m7s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m42s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
Co-authored-by: frontend-dev <frontend-dev@xiaoxiajianji.com>
Co-committed-by: frontend-dev <frontend-dev@xiaoxiajianji.com>
2026-09-12 02:41:11 +08:00
frontend-dev 982daac6e5 Merge pull request 'fix(ai-avatar): 全盘修复 FFmpeg 渲染滤镜链路(exit 234 P0)' (#1868) from fix/ffmpeg-filter-comprehensive-fix into develop
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 1s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 6s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m29s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 4m2s
CI/CD Pipeline / Integration Tests (push) Successful in 7m24s
CI/CD Pipeline / Validate - Style (push) Successful in 9m16s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 9m56s
CI/CD Pipeline / Unit Tests (push) Successful in 12m12s
CI/CD Pipeline / Validate - Security (push) Successful in 23m10s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 39m39s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 41m2s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 3m10s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m43s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m50s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Failing after 3m27s
2026-09-11 23:40:21 +08:00
CI Bot a7067c8171 style: auto-format with black + isort + ruff + prettier [skip ci-format-check]
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 32s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 33s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m48s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m42s
AI Code Review / AI Code Review (pull_request) Successful in 6m37s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 7m55s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 8m12s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 8m22s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 10m50s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 11m0s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 5s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 2m55s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 28m58s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 1s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
2026-09-11 15:20:35 +00:00
LingYing Agent 32c3d2f263 fix(ai-avatar): 全盘修复 FFmpeg 渲染滤镜链路(exit 234 P0)
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 27s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m43s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m55s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m44s
AI Code Review / AI Code Review (pull_request) Successful in 6m34s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 8m17s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 8m45s
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
根因分析(退出码 234 = Invalid argument):
1) drawtext 使用了无效参数 font=bold,导致整个 filter_complex 解析失败
   —— FFmpeg drawtext 没有 bold 参数;改为通过 borderw 模拟粗体视觉效果
2) DRAWTEXT_FONT_SEARCH_PATHS 未包含 Dockerfile 中 COPY 的
   NotoSansSC-VF.ttf 路径,且把不支持中文的 DejaVuSans 放在 fallback 首位,
   导致字体 fallback 到拉丁字体,中文渲染乱码/方框
3) build_broll_overlay_filter 硬编码 1280:720 横屏尺寸,AI 数字人是 9:16 竖屏
4) B-roll PIP 输入索引错误(用 len(sorted_segments) 而非原始下标映射),
   PIP 标签链断裂([pip0]→[vout0] 而非 [vout])
5) 渲染命令缺 -map 0:a?,合成后音频丢失
6) 标题滤镜与 B-roll 输出标签拼接用分号有缺陷,末尾分号处理偶发问题

修复内容(packages/domain/video_filter_builder.py):
- DRAWTEXT_FONT_SEARCH_PATHS: VF 字体置顶、移除 DejaVuSans、加 Bold.ttc 路径
- DRAWTEXT_FONT_MAP: 思源黑体等关键字改为 NotoSansSC(匹配 VF 文件名)
- 删除 font=bold 无效参数;bold 无显式描边时自动用 borderw=3+同色描边模拟粗体
- build_broll_overlay_filter 返回 (filter_str, final_label) 元组,解决标签问题
- 重写 fullscreen/PIP 拆分:原始列表下标决定 -i 输入序号,sorted 只用于时序处理
- PIP overlay 正确基于 fullscreen 输出([vout_fs])或主视频([0:v])链接
- output_width/output_height 贯穿所有 scale/pad/overlay,默认 1280x720 兼容旧调用

修复内容(apps/api/app/services/ai_avatar_render_service.py):
- 新增 _probe_video_resolution() 用 ffprobe 探测输入视频实际分辨率
- AI 数字人默认竖屏 720x1280,探测失败兜底不阻断渲染
- build_broll_overlay_filter 调用传实际 output_width/output_height
- 标题滤镜传实际尺寸,保证位置/坐标计算正确
- filter_complex 拼接重写:broll+title/broll-only/title-only/无滤镜四分支清晰
- _build_ffmpeg_command 补 -map 0:a? + -c:a aac,音频不再丢失
- 清理冗余内联 import

测试:
- 更新 test_ai_avatar_render_routes.py / test_video_filter_builder.py 适配新元组签名
- 新增 test_bold_true_does_not_use_font_bold_param 防回归
- 本地 ffmpeg 实测中文标题+B-roll+竖屏720x1280合成成功
- 相关 3524 个单测全通过
2026-09-11 23:09:22 +08:00
16 changed files with 773 additions and 160 deletions
@@ -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")
+1
View File
@@ -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,
]
+214 -2
View File
@@ -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)
+11 -1
View File
@@ -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"
}
+14 -2
View File
@@ -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,
+17 -18
View File
@@ -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)
+160 -86
View File
@@ -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(
+6 -3
View File
@@ -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
+171
View File
@@ -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()
+29 -7
View File
@@ -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):