feat(sfx): 新增 Sonilo AI 音效模式(可选,默认关闭)

#270 配乐模式基础上新增可选的 "AI 音效(Sonilo)":将合并后的成片
上传到 Sonilo API(POST /v1/video-to-sfx,异步任务管线),根据画面内容
生成音效,再用 ffmpeg 混在成片现有音轨之下(视频流直接复制,不重编码
画面)。解说配音在后续合成步骤中单独混入,音量策略不受影响。生成的
音效为免版税素材。

- 复用 #270 的 sonilo_api_key 配置与 app/services/sonilo.py 模块,
  新增音效任务管线(提交 / 轮询 / 预签名 URL 下载)
- 上传前 ffprobe 本地校验时长(音效接口上限 3 分钟),避免白传计费
- 轮询是免费幂等 GET:网络抖动与 5xx 在截止时间内重试,不报废已计费任务
- 任何失败(超时、HTTP 错误、任务失败、混音失败)只记录日志并沿用
  原视频,绝不中断成片任务
- WebUI 音频设置新增独立开关面板(默认关闭),选项处提示视频将上传至
  Sonilo API
- 现有 BGM / 合成逻辑不做任何改动,未勾选时行为与 main 完全一致
This commit is contained in:
Cindy Xu 2026-07-16 14:17:16 -07:00 committed by cindyxu1030
parent d46d2ca2a5
commit 14828256de
9 changed files with 843 additions and 26 deletions

1
.gitignore vendored
View File

@ -55,6 +55,7 @@ tests/*
!tests/test_generate_script_docu_unittest.py
!tests/test_streamlit_widget_session_state.py
!tests/test_sonilo_bgm_unittest.py
!tests/test_sonilo_sfx_unittest.py
docs/reddit-community
docs/wechat-0.8

View File

@ -183,6 +183,7 @@ class VideoClipParams(BaseModel):
bgm_name: Optional[str] = Field(default="random", description="背景音乐名称")
bgm_type: Optional[str] = Field(default="random", description="背景音乐类型")
bgm_file: Optional[str] = Field(default="", description="背景音乐文件")
sonilo_sfx_enabled: Optional[bool] = Field(default=False, description="是否启用 Sonilo AI 音效(可选功能,默认关闭)")
subtitle_enabled: bool = True
subtitle_mask_enabled: bool = False

View File

@ -1,31 +1,50 @@
"""
Sonilo (https://sonilo.com) AI 配乐BGM集成 可选功能默认关闭
Sonilo (https://sonilo.com) AI 配乐BGM AI 音效SFX集成
均为可选功能默认关闭
将合成完成的视频未加 BGM上传到 Sonilo API`POST /v1/video-to-music`
根据画面内容与剪辑节奏生成一段背景音乐作为普通音频文件交还给现有的
合成流程混音生成的音乐已获授权可商用以条款为准
配乐BGM将合成完成的视频未加 BGM上传到 Sonilo API
`POST /v1/video-to-music`根据画面内容与剪辑节奏生成一段背景音乐
作为普通音频文件交还给现有的合成流程混音生成的音乐已获授权
可商用以条款为准
设计约束完全不影响现有 BGM 逻辑
* 默认关闭仅当用户在 WebUI 中把背景音乐来源切换为 Sonilo且配置了
音效SFX将合成完成的视频上传到 Sonilo API
`POST /v1/video-to-sfx`根据画面内容生成贴合画面的音效该接口是
异步任务提交后返回 task_id轮询 `GET /v1/tasks/{task_id}` 直到终态
成功后从预签名 URL 下载音效音频再用 ffmpeg 混在成片现有音轨之下
解说配音在后续合成步骤中单独混入音量策略不受影响生成的音效为
免版税素材
设计约束完全不影响现有合成逻辑
* 默认关闭配乐仅当用户在 WebUI 中把背景音乐来源切换为 Sonilo 时启用
音效仅当用户勾选 "AI 音效Sonilo" 时启用两者都要求配置了
Sonilo API Keyconfig.toml `sonilo_api_key`或环境变量
`SONILO_API_KEY` 兜底时才会启用
* 本模块只负责生成音频文件音量淡出循环与混音全部复用
`SONILO_API_KEY` 兜底
* 配乐模块只负责生成音频文件音量淡出循环与混音全部复用
app/services/generate_video.py 中现有的音频处理逻辑解说配音的
音量压制策略保持不变
* 任何失败超时HTTP 错误流中断时长超限都只记录日志并返回
空字符串由调用方回退到现有的 BGM 逻辑绝不中断成片任务
* 上传属于计费操作上传前先用 ffprobe 在本地校验视频时长接口
目前拒绝超过 6 分钟的视频避免白传一次注定被拒绝的成片
* 任何失败超时HTTP 错误流中断时长超限混音失败都只记录
日志并返回空字符串由调用方回退到现有逻辑绝不中断成片任务
* 上传属于计费操作上传前先用 ffprobe 在本地校验视频时长配乐接口
目前拒绝超过 6 分钟的视频音效接口拒绝超过 3 分钟的视频避免
白传一次注定被拒绝的成片
接口返回 NDJSON 事件流`audio_chunk`base64 音频分片 stream_index
分组`title``complete`成功终止事件 `error`失败终止事件
进度事件与无法解析的行一律忽略生成的音频为 AAC 编码的 .m4a 文件
配乐接口返回 NDJSON 事件流`audio_chunk`base64 音频分片
stream_index 分组`title``complete`成功终止事件 `error`
失败终止事件进度事件与无法解析的行一律忽略生成的音频为 AAC
编码的 .m4a 文件
音效接口为异步任务管线`POST /v1/video-to-sfx` 受理后即计费并返回
`{"task_id": ...}``GET /v1/tasks/{task_id}` 返回
`{"status": "succeeded"/"failed"/..., "audio": {"url": ...}, "error": ...,
"refunded": ...}`结果地址是预签名 URL下载时绝不能携带 API Key
配置config.toml [app]
sonilo_api_key = "..." # 必填,启用开关
sonilo_api_key = "..." # 必填,启用开关(配乐与音效共用)
# sonilo_base_url = "https://api.sonilo.com"
# sonilo_timeout_seconds = 600
# sonilo_bgm_prompt = "" # 可选:配乐风格提示
# sonilo_sfx_prompt = "" # 可选:音效风格提示
# sonilo_sfx_volume = 0.6 # 音效混入原声之下的音量0-2]
"""
import base64
@ -33,6 +52,7 @@ import binascii
import json
import os
import subprocess
import time
from typing import Iterable, Optional
import requests
@ -42,12 +62,24 @@ from app.config import config
DEFAULT_BASE_URL = "https://api.sonilo.com"
VIDEO_TO_MUSIC_PATH = "/v1/video-to-music"
VIDEO_TO_SFX_PATH = "/v1/video-to-sfx"
TASKS_PATH = "/v1/tasks"
# 后端生成接口的读超时约为 600 秒。生成一旦开始就会计费,客户端过早超时
# 只会浪费一次已经付费的请求,所以默认读超时与后端保持一致,并允许覆盖。
DEFAULT_TIMEOUT_SECONDS = 600
_CONNECT_TIMEOUT_SECONDS = 15
# 接口目前拒绝超过 6 分钟的视频;上传前先在本地校验时长。
# 轮询任务状态是免费且幂等的 GET单次请求用短读超时即可。
_POLL_READ_TIMEOUT_SECONDS = 30
_SFX_POLL_INTERVAL_SECONDS = 5.0
# 测试接缝:单测里替换为 no-op避免真实等待。
_sleep = time.sleep
# 配乐接口目前拒绝超过 6 分钟的视频;上传前先在本地校验时长。
MAX_VIDEO_DURATION_SECONDS = 360
# 音效接口目前拒绝超过 3 分钟的视频。
MAX_SFX_VIDEO_DURATION_SECONDS = 180
# 音效混入原声之下的默认音量(解说配音在后续合成步骤中以 1.0 混入,
# 音效保持在其之下)。可通过 sonilo_sfx_volume 配置覆盖。
DEFAULT_SFX_VOLUME = 0.6
class SoniloError(Exception):
@ -65,6 +97,10 @@ def is_enabled() -> bool:
return bool(get_api_key())
def _get_base_url() -> str:
return str(config.app.get("sonilo_base_url", "") or DEFAULT_BASE_URL).rstrip("/")
def generate_bgm(video_path: str, save_path: str) -> str:
"""
上传合成完成的视频未加 BGM Sonilo生成配乐并保存到
@ -182,9 +218,7 @@ def _get_timeout_seconds() -> float:
def _request_video_to_music(video_path: str) -> bytes:
base_url = str(config.app.get("sonilo_base_url", "") or DEFAULT_BASE_URL).rstrip(
"/"
)
base_url = _get_base_url()
timeout_seconds = _get_timeout_seconds()
prompt = str(config.app.get("sonilo_bgm_prompt", "") or "").strip()
@ -271,3 +305,322 @@ def _consume_ndjson_stream(lines: Iterable[str]) -> bytes:
raise SoniloError("Sonilo 事件流已完成但未返回音频数据")
first_index = sorted(streams)[0]
return bytes(streams[first_index])
# ---------- AI 音效SFX可选功能默认关闭 ----------
def apply_sfx(video_path: str, output_path: str) -> str:
"""
为合成完成的视频生成 Sonilo 音效并用 ffmpeg 混在现有音轨之下
输出新视频到 `output_path`视频流直接复制不重编码画面
成功时返回输出视频路径任何失败都返回空字符串由调用方沿用
原视频本函数绝不抛出异常 音效问题绝不能中断成片任务
"""
sfx_audio_path = os.path.splitext(output_path)[0] + ".m4a"
if not generate_sfx(video_path, sfx_audio_path):
return ""
return _mix_sfx_under_original(video_path, sfx_audio_path, output_path)
def generate_sfx(video_path: str, save_path: str) -> str:
"""
上传合成完成的视频到 Sonilo生成音效音频并保存到 `save_path`.m4a
成功时返回音频文件路径任何失败都返回空字符串 generate_bgm
的约定一致本函数绝不抛出异常
"""
if not is_enabled():
logger.warning("Sonilo 音效已跳过: 未配置 API Key")
return ""
if not video_path or not os.path.isfile(video_path):
logger.warning(f"Sonilo 音效已跳过: 视频文件不存在: {video_path}")
return ""
# 任务受理即计费,先在本地校验时长,避免白传一次注定被拒绝的成片。
duration = _probe_video_duration(video_path)
if duration and duration > MAX_SFX_VIDEO_DURATION_SECONDS:
logger.warning(
f"Sonilo 音效已跳过: 视频时长 {duration:.1f}s 超过接口上限 "
f"{MAX_SFX_VIDEO_DURATION_SECONDS}s"
)
return ""
try:
audio = _request_video_to_sfx(video_path)
except Exception as e:
logger.error(f"Sonilo 音效生成失败: {str(e)}")
return ""
try:
with open(save_path, "wb") as f:
f.write(audio)
except OSError as e:
logger.error(f"Sonilo 音效文件保存失败: {str(e)}")
return ""
logger.success(f"Sonilo 音效已生成: {save_path}")
return save_path
def _request_video_to_sfx(video_path: str) -> bytes:
"""提交音效任务、轮询到终态、下载结果音频。失败抛出 SoniloError。"""
task_id = _submit_sfx_task(video_path)
body = _poll_sfx_task(task_id)
return _download_sfx_audio(_extract_sfx_audio_url(body, task_id))
def _submit_sfx_task(video_path: str) -> str:
"""POST /v1/video-to-sfx受理后返回 task_id受理即计费不做重试"""
timeout_seconds = _get_timeout_seconds()
prompt = str(config.app.get("sonilo_sfx_prompt", "") or "").strip()
data: Optional[dict] = {"prompt": prompt} if prompt else None
headers = {"Authorization": f"Bearer {get_api_key()}"}
logger.info(f"正在提交 Sonilo 音效任务, 视频: {video_path}")
try:
with open(video_path, "rb") as video_file:
files = {
"video": (os.path.basename(video_path), video_file, "video/mp4"),
}
response = requests.post(
f"{_get_base_url()}{VIDEO_TO_SFX_PATH}",
headers=headers,
data=data,
files=files,
timeout=(_CONNECT_TIMEOUT_SECONDS, timeout_seconds),
)
except requests.exceptions.Timeout as exc:
raise SoniloError(
f"Sonilo 音效任务提交超时 ({timeout_seconds:.0f}s)"
) from exc
except requests.exceptions.RequestException as exc:
raise SoniloError(f"Sonilo 音效任务提交失败: {str(exc)}") from exc
if response.status_code >= 400:
body = response.content.decode("utf-8", errors="replace")
raise SoniloError(_http_error_message(response.status_code, body))
try:
task_id = response.json().get("task_id")
except (ValueError, AttributeError):
task_id = None
if not task_id:
raise SoniloError("Sonilo 音效任务已受理但未返回 task_id")
task_id = str(task_id)
# 受理即计费;先把 task_id 落进日志,后续轮询失败时仍有据可查。
logger.info(f"Sonilo 音效任务已提交: {task_id}")
return task_id
def _poll_sfx_task(task_id: str) -> dict:
"""
轮询 GET /v1/tasks/{task_id} 直到任务终态succeeded/failed或超时
succeeded 时返回任务体failed / 超时 / 不可恢复的 HTTP 错误抛出
SoniloError轮询是免费且幂等的 GET网络抖动与 5xx 不该报废一次
已计费的任务在截止时间内继续重试
"""
headers = {"Authorization": f"Bearer {get_api_key()}"}
timeout_seconds = _get_timeout_seconds()
deadline = time.monotonic() + timeout_seconds
while True:
response = None
try:
response = requests.get(
f"{_get_base_url()}{TASKS_PATH}/{task_id}",
headers=headers,
timeout=(_CONNECT_TIMEOUT_SECONDS, _POLL_READ_TIMEOUT_SECONDS),
)
except requests.exceptions.RequestException as exc:
logger.warning(f"Sonilo 音效任务查询失败(将重试): {str(exc)}")
if response is not None:
if response.status_code >= 500:
logger.warning(
f"Sonilo 音效任务查询返回 {response.status_code}(将重试)"
)
elif response.status_code >= 400:
body = response.content.decode("utf-8", errors="replace")
raise SoniloError(
f"{_http_error_message(response.status_code, body)}"
f"(任务已提交, task_id: {task_id}"
)
else:
try:
body = response.json()
except ValueError:
body = None
if isinstance(body, dict):
status = body.get("status")
if status == "succeeded":
return body
if status == "failed":
raise SoniloError(_task_failure_message(body, task_id))
# 非终态pending / processing 等)继续等待。
if time.monotonic() >= deadline:
raise SoniloError(
f"等待 Sonilo 音效任务超时 ({timeout_seconds:.0f}s), "
f"task_id: {task_id}"
)
_sleep(_SFX_POLL_INTERVAL_SECONDS)
def _task_failure_message(body: dict, task_id: str) -> str:
err = body.get("error")
if isinstance(err, dict):
message = err.get("message") or err.get("code") or "生成失败"
elif isinstance(err, str) and err:
message = err
else:
message = "生成失败"
refund_note = ",费用已退还" if body.get("refunded") is True else ""
return f"Sonilo 音效生成失败: {message}task_id: {task_id}{refund_note}"
def _extract_sfx_audio_url(body: dict, task_id: str) -> str:
audio = body.get("audio")
if isinstance(audio, dict):
url = audio.get("url")
if isinstance(url, str) and url:
return url
raise SoniloError(
f"Sonilo 音效任务成功但未返回音频结果 (task_id: {task_id})"
)
def _download_sfx_audio(url: str) -> bytes:
"""下载任务结果音频。结果地址是预签名 URL自带鉴权 ——
绝不能把 API Key 发给存储域名因此这里不带任何鉴权头"""
try:
response = requests.get(
url, timeout=(_CONNECT_TIMEOUT_SECONDS, _get_timeout_seconds())
)
except requests.exceptions.RequestException as exc:
raise SoniloError(f"Sonilo 音效结果下载失败: {str(exc)}") from exc
if response.status_code >= 400:
raise SoniloError(
f"Sonilo 音效结果下载失败 (HTTP {response.status_code})"
)
if not response.content:
raise SoniloError("Sonilo 音效结果为空")
return response.content
def _get_sfx_volume() -> float:
"""音效混入原声之下的音量。非法值或 <=0 回退默认值,上限 2.0。"""
try:
volume = float(config.app.get("sonilo_sfx_volume", DEFAULT_SFX_VOLUME))
except (TypeError, ValueError):
return DEFAULT_SFX_VOLUME
if volume <= 0:
return DEFAULT_SFX_VOLUME
return min(volume, 2.0)
def _get_ffmpeg_binary() -> str:
"""与 generate_video 保持一致的 ffmpeg 查找逻辑(环境变量优先)。"""
for env_name in ("NARRATO_FFMPEG_EXE", "IMAGEIO_FFMPEG_EXE"):
candidate = os.environ.get(env_name, "").strip()
if candidate and os.path.isfile(candidate):
return candidate
try:
import imageio_ffmpeg
candidate = imageio_ffmpeg.get_ffmpeg_exe()
if candidate and os.path.isfile(candidate):
return candidate
except Exception:
pass
return "ffmpeg"
def _probe_has_audio_stream(video_path: str) -> bool:
"""尽力而为地探测视频是否带音轨。探测失败按无音轨处理。"""
try:
result = subprocess.run(
[
_get_ffprobe_binary(),
"-v",
"quiet",
"-print_format",
"json",
"-show_streams",
"-select_streams",
"a",
video_path,
],
capture_output=True,
timeout=30,
)
except (OSError, subprocess.TimeoutExpired):
return False
if result.returncode != 0:
return False
try:
return bool(json.loads(result.stdout).get("streams"))
except (json.JSONDecodeError, AttributeError, TypeError, ValueError):
return False
def _mix_sfx_under_original(
video_path: str, sfx_audio_path: str, output_path: str
) -> str:
"""
ffmpeg 把音效混在成片现有音轨之下音效音量默认 0.6原声音量
不变视频流直接复制不重编码成片没有音轨时音效直接作为音轨
写入成功返回 output_path任何失败返回空字符串
"""
volume = _get_sfx_volume()
has_audio = _probe_has_audio_stream(video_path)
if has_audio:
filter_complex = (
f"[1:a]volume={volume}[sfx];"
"[0:a][sfx]amix=inputs=2:duration=first:"
"dropout_transition=0:normalize=0[aout]"
)
else:
filter_complex = f"[1:a]volume={volume}[aout]"
cmd = [
_get_ffmpeg_binary(),
"-y",
"-i",
video_path,
"-i",
sfx_audio_path,
"-filter_complex",
filter_complex,
"-map",
"0:v",
"-map",
"[aout]",
"-c:v",
"copy",
"-c:a",
"aac",
"-b:a",
"192k",
]
if not has_audio:
cmd.append("-shortest")
cmd.append(output_path)
logger.info(f"正在混入 Sonilo 音效 (音量 {volume}): {output_path}")
try:
result = subprocess.run(cmd, capture_output=True, timeout=300)
except (OSError, subprocess.TimeoutExpired) as e:
logger.error(f"Sonilo 音效混音失败: {str(e)}")
return ""
if result.returncode != 0:
stderr_tail = (result.stderr or b"").decode("utf-8", errors="replace")[-500:]
logger.error(f"Sonilo 音效混音失败 (ffmpeg 退出码 {result.returncode}): {stderr_tail}")
return ""
logger.success(f"Sonilo 音效已混入: {output_path}")
return output_path

View File

@ -236,6 +236,24 @@ def _resolve_bgm_path(task_id: str, params: VideoClipParams, combined_video_path
)
def _apply_sonilo_sfx(task_id: str, params: VideoClipParams, combined_video_path: str) -> str:
"""为合并后的成片混入 Sonilo AI 音效(可选功能,默认关闭)。
仅当 params.sonilo_sfx_enabled True 时启用把合并后的成片上传到
Sonilo API 生成音效再用 ffmpeg 混在现有音轨之下返回新视频路径
解说配音在后续 merge_materials 中单独混入音量策略不受影响任何
失败都只记录日志并沿用原视频绝不中断成片任务
"""
if not getattr(params, "sonilo_sfx_enabled", False):
return combined_video_path
output_path = path.join(utils.task_dir(task_id), "merger_sfx.mp4")
sfx_video_path = sonilo.apply_sfx(combined_video_path, output_path)
if sfx_video_path:
return sfx_video_path
logger.warning("Sonilo 音效不可用,继续使用未加音效的成片")
return combined_video_path
def _transcribe_final_video(task_id: str, video_path: str, params: VideoClipParams) -> str:
"""Transcribe the fully merged video into an SRT file."""
from app.services import fun_asr_subtitle
@ -542,6 +560,9 @@ def start_subclip(task_id: str, params: VideoClipParams, subclip_path_videos: di
)
logger.info(f"\n\n## 6. 最后一步: 合并字幕/BGM/配音/视频 -> {merge_output_video_path}")
# 可选功能,默认关闭:混入 Sonilo AI 音效(失败时沿用原视频)
combined_video_path = _apply_sonilo_sfx(task_id, params, combined_video_path)
# bgm_path = '/Users/apple/Desktop/home/NarratoAI/resource/songs/bgm.mp3'
bgm_path = _resolve_bgm_path(task_id, params, combined_video_path)
@ -869,6 +890,9 @@ def start_subclip_unified(task_id: str, params: VideoClipParams):
ffmpeg_progress=0,
)
# 可选功能,默认关闭:混入 Sonilo AI 音效(失败时沿用原视频)
combined_video_path = _apply_sonilo_sfx(task_id, params, combined_video_path)
bgm_path = _resolve_bgm_path(task_id, params, combined_video_path)
# 获取优化的音量配置

View File

@ -68,14 +68,19 @@
tavily_search_depth = "basic" # basic / advanced / fast / ultra-fast
tavily_max_results = 5
# ===== 可选Sonilo AI 配乐BGM=====
# 在 WebUI 背景音乐来源中选择 "AI 生成配乐Sonilo" 即可启用(默认关闭,不影响现有 BGM 逻辑)。
# ===== 可选Sonilo AI 配乐BGM/ AI 音效SFX=====
# 配乐:在 WebUI 背景音乐来源中选择 "AI 生成配乐Sonilo" 即可启用(默认关闭,不影响现有 BGM 逻辑)。
# 启用后会将合成完成的视频(未加 BGM上传到 Sonilo API根据画面内容与剪辑节奏生成配乐
# 生成的音乐已获授权、可商用(以条款为准)。视频时长上限 6 分钟,生成失败时自动回退到随机背景音乐。
sonilo_api_key = "" # 获取地址https://sonilo.com
# 音效:在 WebUI 音频设置中勾选 "AI 音效Sonilo" 即可启用(默认关闭)。
# 启用后会将合成完成的视频上传到 Sonilo API根据画面内容生成音效并混在现有音轨之下解说不受影响
# 生成的音效为免版税素材。视频时长上限 3 分钟,生成失败时自动跳过音效。
sonilo_api_key = "" # 获取地址https://sonilo.com配乐与音效共用
# sonilo_base_url = "https://api.sonilo.com"
# sonilo_timeout_seconds = 600 # 生成接口读超时(秒)
# sonilo_timeout_seconds = 600 # 生成接口读超时 / 音效任务等待上限(秒)
# sonilo_bgm_prompt = "" # 可选:配乐风格提示,留空则完全根据画面生成
# sonilo_sfx_prompt = "" # 可选:音效风格提示,留空则完全根据画面生成
# sonilo_sfx_volume = 0.6 # 音效混入原声之下的音量,范围 (0, 2],默认 0.6
# ===== API Keys 参考 =====
# 主流 LLM Providers API Key 获取地址:

View File

@ -0,0 +1,362 @@
import os
import tempfile
import unittest
from unittest import mock
import requests
from app.services import sonilo
def _response(status_code=200, json_body=None, content=b"{}"):
resp = mock.Mock()
resp.status_code = status_code
resp.content = content
if json_body is None:
resp.json.side_effect = ValueError("no json")
else:
resp.json.return_value = json_body
return resp
class SubmitSfxTaskTests(unittest.TestCase):
def setUp(self):
self._tmp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp_dir.cleanup)
self.video_path = os.path.join(self._tmp_dir.name, "combined.mp4")
with open(self.video_path, "wb") as f:
f.write(b"fake video")
def test_returns_task_id(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(
sonilo.requests,
"post",
return_value=_response(202, {"task_id": "task-123"}),
) as post_mock:
self.assertEqual("task-123", sonilo._submit_sfx_task(self.video_path))
args, kwargs = post_mock.call_args
self.assertTrue(args[0].endswith("/v1/video-to-sfx"))
self.assertEqual("Bearer sk-test", kwargs["headers"]["Authorization"])
def test_http_error_raises(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(
sonilo.requests,
"post",
return_value=_response(402, content=b'{"detail": "no credits"}'),
):
with self.assertRaises(sonilo.SoniloError):
sonilo._submit_sfx_task(self.video_path)
def test_missing_task_id_raises(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(
sonilo.requests, "post", return_value=_response(202, {})
):
with self.assertRaises(sonilo.SoniloError):
sonilo._submit_sfx_task(self.video_path)
class PollSfxTaskTests(unittest.TestCase):
def test_returns_body_when_succeeded(self):
responses = [
_response(200, {"status": "processing"}),
_response(
200,
{"status": "succeeded", "audio": {"url": "https://cdn/x.m4a"}},
),
]
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_sleep"), \
mock.patch.object(sonilo.requests, "get", side_effect=responses):
body = sonilo._poll_sfx_task("task-123")
self.assertEqual("succeeded", body["status"])
def test_failed_status_raises_with_task_id(self):
response = _response(
200,
{
"status": "failed",
"error": {"code": "GENERATION_FAILED", "message": "boom"},
"refunded": True,
},
)
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_sleep"), \
mock.patch.object(sonilo.requests, "get", return_value=response):
with self.assertRaises(sonilo.SoniloError) as ctx:
sonilo._poll_sfx_task("task-123")
self.assertIn("task-123", str(ctx.exception))
self.assertIn("boom", str(ctx.exception))
def test_transient_error_retries_until_succeeded(self):
responses = [
requests.exceptions.ConnectionError("blip"),
_response(
200,
{"status": "succeeded", "audio": {"url": "https://cdn/x.m4a"}},
),
]
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_sleep"), \
mock.patch.object(sonilo.requests, "get", side_effect=responses):
body = sonilo._poll_sfx_task("task-123")
self.assertEqual("succeeded", body["status"])
def test_non_recoverable_http_error_raises(self):
response = _response(401, content=b'{"detail": "bad key"}')
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_sleep"), \
mock.patch.object(sonilo.requests, "get", return_value=response):
with self.assertRaises(sonilo.SoniloError):
sonilo._poll_sfx_task("task-123")
def test_timeout_raises_with_task_id(self):
response = _response(200, {"status": "processing"})
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_sleep"), \
mock.patch.object(sonilo, "_get_timeout_seconds", return_value=0.0), \
mock.patch.object(sonilo.requests, "get", return_value=response):
with self.assertRaises(sonilo.SoniloError) as ctx:
sonilo._poll_sfx_task("task-123")
self.assertIn("task-123", str(ctx.exception))
class DownloadSfxAudioTests(unittest.TestCase):
def test_returns_content_without_auth_headers(self):
response = _response(200, content=b"audio-bytes")
with mock.patch.object(
sonilo.requests, "get", return_value=response
) as get_mock:
self.assertEqual(
b"audio-bytes", sonilo._download_sfx_audio("https://cdn/x.m4a")
)
# 预签名 URL 自带鉴权,绝不能把 API Key 发给存储域名。
_, kwargs = get_mock.call_args
self.assertNotIn("headers", kwargs)
def test_http_error_raises(self):
with mock.patch.object(
sonilo.requests, "get", return_value=_response(403, content=b"denied")
):
with self.assertRaises(sonilo.SoniloError):
sonilo._download_sfx_audio("https://cdn/x.m4a")
def test_empty_content_raises(self):
with mock.patch.object(
sonilo.requests, "get", return_value=_response(200, content=b"")
):
with self.assertRaises(sonilo.SoniloError):
sonilo._download_sfx_audio("https://cdn/x.m4a")
class ExtractSfxAudioUrlTests(unittest.TestCase):
def test_returns_audio_url(self):
body = {"status": "succeeded", "audio": {"url": "https://cdn/x.m4a"}}
self.assertEqual(
"https://cdn/x.m4a", sonilo._extract_sfx_audio_url(body, "task-123")
)
def test_missing_audio_raises(self):
with self.assertRaises(sonilo.SoniloError):
sonilo._extract_sfx_audio_url({"status": "succeeded"}, "task-123")
class GenerateSfxTests(unittest.TestCase):
def setUp(self):
self._tmp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp_dir.cleanup)
self.video_path = os.path.join(self._tmp_dir.name, "combined.mp4")
with open(self.video_path, "wb") as f:
f.write(b"fake video")
self.save_path = os.path.join(self._tmp_dir.name, "sonilo_sfx.m4a")
def test_returns_empty_without_api_key(self):
with mock.patch.object(sonilo, "get_api_key", return_value=""):
self.assertEqual("", sonilo.generate_sfx(self.video_path, self.save_path))
def test_returns_empty_when_video_missing(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"):
missing = os.path.join(self._tmp_dir.name, "missing.mp4")
self.assertEqual("", sonilo.generate_sfx(missing, self.save_path))
def test_skips_upload_when_duration_exceeds_limit(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_probe_video_duration", return_value=181.0), \
mock.patch.object(sonilo, "_request_video_to_sfx") as request_mock:
self.assertEqual("", sonilo.generate_sfx(self.video_path, self.save_path))
request_mock.assert_not_called()
def test_saves_audio_on_success(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_probe_video_duration", return_value=60.0), \
mock.patch.object(sonilo, "_request_video_to_sfx", return_value=b"audio-bytes"):
self.assertEqual(self.save_path, sonilo.generate_sfx(self.video_path, self.save_path))
with open(self.save_path, "rb") as f:
self.assertEqual(b"audio-bytes", f.read())
def test_request_failure_degrades_to_empty(self):
with mock.patch.object(sonilo, "get_api_key", return_value="sk-test"), \
mock.patch.object(sonilo, "_probe_video_duration", return_value=60.0), \
mock.patch.object(
sonilo,
"_request_video_to_sfx",
side_effect=sonilo.SoniloError("timeout"),
):
self.assertEqual("", sonilo.generate_sfx(self.video_path, self.save_path))
self.assertFalse(os.path.exists(self.save_path))
class GetSfxVolumeTests(unittest.TestCase):
def test_default_when_unset(self):
with mock.patch.object(sonilo.config, "app", {}):
self.assertEqual(sonilo.DEFAULT_SFX_VOLUME, sonilo._get_sfx_volume())
def test_default_when_invalid(self):
with mock.patch.object(sonilo.config, "app", {"sonilo_sfx_volume": "abc"}):
self.assertEqual(sonilo.DEFAULT_SFX_VOLUME, sonilo._get_sfx_volume())
def test_default_when_non_positive(self):
with mock.patch.object(sonilo.config, "app", {"sonilo_sfx_volume": 0}):
self.assertEqual(sonilo.DEFAULT_SFX_VOLUME, sonilo._get_sfx_volume())
def test_clamped_to_upper_bound(self):
with mock.patch.object(sonilo.config, "app", {"sonilo_sfx_volume": 5}):
self.assertEqual(2.0, sonilo._get_sfx_volume())
def test_valid_value_passes_through(self):
with mock.patch.object(sonilo.config, "app", {"sonilo_sfx_volume": 0.8}):
self.assertEqual(0.8, sonilo._get_sfx_volume())
class MixSfxUnderOriginalTests(unittest.TestCase):
def _run(self, has_audio, returncode=0, run_side_effect=None):
run_mock = mock.Mock(return_value=mock.Mock(returncode=returncode, stderr=b"err"))
if run_side_effect is not None:
run_mock = mock.Mock(side_effect=run_side_effect)
with mock.patch.object(sonilo, "_get_ffmpeg_binary", return_value="ffmpeg"), \
mock.patch.object(sonilo, "_probe_has_audio_stream", return_value=has_audio), \
mock.patch.object(sonilo, "_get_sfx_volume", return_value=0.6), \
mock.patch.object(sonilo.subprocess, "run", run_mock):
result = sonilo._mix_sfx_under_original(
"/tmp/combined.mp4", "/tmp/sfx.m4a", "/tmp/merger_sfx.mp4"
)
return result, run_mock
def test_mixes_under_existing_audio_with_amix(self):
result, run_mock = self._run(has_audio=True)
self.assertEqual("/tmp/merger_sfx.mp4", result)
cmd = run_mock.call_args[0][0]
filter_complex = cmd[cmd.index("-filter_complex") + 1]
self.assertIn("volume=0.6", filter_complex)
self.assertIn("amix", filter_complex)
# 视频流直接复制,不重编码画面。
self.assertIn("copy", cmd[cmd.index("-c:v") + 1])
self.assertNotIn("-shortest", cmd)
def test_sfx_becomes_audio_track_when_video_has_no_audio(self):
result, run_mock = self._run(has_audio=False)
self.assertEqual("/tmp/merger_sfx.mp4", result)
cmd = run_mock.call_args[0][0]
filter_complex = cmd[cmd.index("-filter_complex") + 1]
self.assertNotIn("amix", filter_complex)
self.assertIn("-shortest", cmd)
def test_ffmpeg_failure_returns_empty(self):
result, _ = self._run(has_audio=True, returncode=1)
self.assertEqual("", result)
def test_ffmpeg_oserror_returns_empty(self):
result, _ = self._run(has_audio=True, run_side_effect=OSError("no ffmpeg"))
self.assertEqual("", result)
class ApplySfxTests(unittest.TestCase):
def test_success_returns_output_path(self):
with mock.patch.object(
sonilo, "generate_sfx", return_value="/tmp/merger_sfx.m4a"
) as generate_mock, mock.patch.object(
sonilo, "_mix_sfx_under_original", return_value="/tmp/merger_sfx.mp4"
) as mix_mock:
result = sonilo.apply_sfx("/tmp/combined.mp4", "/tmp/merger_sfx.mp4")
self.assertEqual("/tmp/merger_sfx.mp4", result)
generate_mock.assert_called_once_with("/tmp/combined.mp4", "/tmp/merger_sfx.m4a")
mix_mock.assert_called_once_with(
"/tmp/combined.mp4", "/tmp/merger_sfx.m4a", "/tmp/merger_sfx.mp4"
)
def test_generation_failure_skips_mixing(self):
with mock.patch.object(sonilo, "generate_sfx", return_value=""), \
mock.patch.object(sonilo, "_mix_sfx_under_original") as mix_mock:
self.assertEqual(
"", sonilo.apply_sfx("/tmp/combined.mp4", "/tmp/merger_sfx.mp4")
)
mix_mock.assert_not_called()
def test_mixing_failure_returns_empty(self):
with mock.patch.object(
sonilo, "generate_sfx", return_value="/tmp/merger_sfx.m4a"
), mock.patch.object(sonilo, "_mix_sfx_under_original", return_value=""):
self.assertEqual(
"", sonilo.apply_sfx("/tmp/combined.mp4", "/tmp/merger_sfx.mp4")
)
class ApplySoniloSfxTaskTests(unittest.TestCase):
"""任务层的音效挂载:默认关闭,失败时沿用原视频,绝不中断成片任务。"""
def _make_params(self, sfx_enabled):
params = mock.Mock()
params.sonilo_sfx_enabled = sfx_enabled
return params
def test_disabled_returns_original_path_without_calling_sonilo(self):
from app.services import task
params = self._make_params(False)
with mock.patch.object(task.sonilo, "apply_sfx") as apply_mock:
result = task._apply_sonilo_sfx("task-id", params, "/tmp/merger.mp4")
self.assertEqual("/tmp/merger.mp4", result)
apply_mock.assert_not_called()
def test_enabled_returns_sfx_video_path(self):
from app.services import task
params = self._make_params(True)
with mock.patch.object(task.utils, "task_dir", return_value="/tmp/task-id"), \
mock.patch.object(
task.sonilo, "apply_sfx", return_value="/tmp/task-id/merger_sfx.mp4"
) as apply_mock:
result = task._apply_sonilo_sfx("task-id", params, "/tmp/merger.mp4")
self.assertEqual("/tmp/task-id/merger_sfx.mp4", result)
apply_mock.assert_called_once_with(
"/tmp/merger.mp4", os.path.join("/tmp/task-id", "merger_sfx.mp4")
)
def test_failure_falls_back_to_original_path(self):
from app.services import task
params = self._make_params(True)
with mock.patch.object(task.utils, "task_dir", return_value="/tmp/task-id"), \
mock.patch.object(task.sonilo, "apply_sfx", return_value=""):
result = task._apply_sonilo_sfx("task-id", params, "/tmp/merger.mp4")
self.assertEqual("/tmp/merger.mp4", result)
if __name__ == "__main__":
unittest.main()

View File

@ -549,6 +549,9 @@ def render_audio_panel(tr):
# 背景音乐独立成框,放在音频设置下方
render_bgm_panel(tr)
# AI 音效独立成框(可选功能,默认关闭)
render_sonilo_sfx_panel(tr)
def render_bgm_panel(tr):
"""渲染背景音乐设置面板"""
@ -556,6 +559,12 @@ def render_bgm_panel(tr):
render_bgm_settings(tr)
def render_sonilo_sfx_panel(tr):
"""渲染 Sonilo AI 音效设置面板(可选功能,默认关闭)"""
with st.container(border=True):
render_sonilo_sfx_settings(tr)
def render_tts_settings(tr):
"""渲染TTS(文本转语音)设置"""
@ -2208,6 +2217,55 @@ def render_sonilo_bgm_settings(tr):
st.warning(tr("Sonilo API Key Required"))
def render_sonilo_sfx_settings(tr):
"""渲染 Sonilo AI 音效设置(可选功能,默认关闭)"""
# 避免在本模块顶层引入 basic_settings 的重依赖链,按需导入。
from webui.components.basic_settings import update_app_config_if_changed
sfx_enabled = st.checkbox(
tr("Sonilo AI Sound Effects"),
value=bool(st.session_state.get("sonilo_sfx_enabled", False)),
help=tr("Sonilo SFX Help"),
key="sonilo_sfx_enabled_checkbox",
)
st.session_state["sonilo_sfx_enabled"] = bool(sfx_enabled)
if not sfx_enabled:
return
st.info(tr("Sonilo SFX Notice"))
sonilo_api_key = st.text_input(
tr("Sonilo API Key"),
value=config.app.get("sonilo_api_key", ""),
type="password",
help=tr("Sonilo API Key Help"),
key="sonilo_sfx_api_key_input",
)
sonilo_sfx_prompt = st.text_input(
tr("Sonilo SFX Prompt"),
value=config.app.get("sonilo_sfx_prompt", ""),
help=tr("Sonilo SFX Prompt Help"),
key="sonilo_sfx_prompt_input",
)
api_key_changed = update_app_config_if_changed(
"sonilo_api_key", str(sonilo_api_key or "").strip()
)
prompt_changed = update_app_config_if_changed(
"sonilo_sfx_prompt", str(sonilo_sfx_prompt or "").strip()
)
if api_key_changed or prompt_changed:
try:
config.save_config()
st.success(tr("Sonilo config saved"))
except Exception as e:
st.error(f"{tr('Failed to save config')}: {str(e)}")
logger.error(f"保存 Sonilo 配置失败: {str(e)}")
if not sonilo.is_enabled():
st.warning(tr("Sonilo SFX API Key Required"))
def render_bgm_settings(tr):
"""渲染背景音乐设置"""
saved_bgm_file = st.session_state.get('bgm_file', '')
@ -2343,5 +2401,6 @@ def get_audio_params():
'bgm_type': st.session_state.get('bgm_type', 'random'),
'bgm_file': st.session_state.get('bgm_file', ''),
'bgm_volume': st.session_state.get('bgm_volume', AudioVolumeDefaults.BGM_VOLUME),
'sonilo_sfx_enabled': bool(st.session_state.get('sonilo_sfx_enabled', False)),
'tts_engine': st.session_state.get('tts_engine', config.INDEXTTS_ENGINE),
}

View File

@ -58,11 +58,17 @@
"Sonilo AI Background Music": "AI-Generated Background Music (Sonilo)",
"Sonilo BGM Notice": "When enabled, the assembled video (without background music) is uploaded to the Sonilo API to generate background music that follows the visuals and editing pace; the narration keeps its existing volume settings and stays clearly audible. Generated music is licensed for commercial use (terms apply). Videos longer than 6 minutes are not supported; if generation fails, the task falls back to random background music.",
"Sonilo API Key": "Sonilo API Key",
"Sonilo API Key Help": "Get an API key at https://sonilo.com. Only used when Sonilo background music is selected.",
"Sonilo API Key Help": "Get an API key at https://sonilo.com. Only used when Sonilo background music or sound effects are enabled.",
"Sonilo API Key Required": "Please enter a Sonilo API Key first, otherwise the task will fall back to random background music.",
"Sonilo BGM Prompt": "Music Style Hint (Optional)",
"Sonilo BGM Prompt Help": "Optional: describe the desired music style, e.g. \"calm piano\" or \"tense suspense\". Leave empty to generate purely from the visuals.",
"Sonilo config saved": "Sonilo configuration saved",
"Sonilo AI Sound Effects": "AI Sound Effects (Sonilo)",
"Sonilo SFX Help": "Automatically generate sound effects that match the visuals of the assembled video (optional, disabled by default).",
"Sonilo SFX Notice": "When enabled, the assembled video is uploaded to the Sonilo API to generate sound effects based on the visuals; the effects are mixed underneath the existing audio track, and the narration keeps its existing volume settings and stays clearly audible. Generated sound effects are royalty-free. Videos longer than 3 minutes are not supported; if generation fails, the sound-effects step is skipped and the video is produced as usual.",
"Sonilo SFX Prompt": "Sound Effects Hint (Optional)",
"Sonilo SFX Prompt Help": "Optional: describe the desired sound effects, e.g. \"rain with distant thunder\" or \"metal clanking\". Leave empty to generate purely from the visuals.",
"Sonilo SFX API Key Required": "Please enter a Sonilo API Key first, otherwise the sound-effects step will be skipped.",
"Upload Background Music": "Upload Background Music",
"Background Music Path Help": "Choose the background music used for video synthesis.",
"No Background Music Resources Found": "No background music resources found. Please upload a background music file.",

View File

@ -46,11 +46,17 @@
"Sonilo AI Background Music": "AI 生成配乐Sonilo",
"Sonilo BGM Notice": "启用后,合成完成的视频(未加背景音乐)将上传至 Sonilo API根据画面内容与剪辑节奏生成配乐解说配音仍按现有音量设置保持清晰。生成的音乐已获授权、可商用以条款为准。视频时长上限 6 分钟;生成失败时自动回退到随机背景音乐,不影响成片。",
"Sonilo API Key": "Sonilo API Key",
"Sonilo API Key Help": "获取地址https://sonilo.com仅在选择 Sonilo 配乐时使用",
"Sonilo API Key Help": "获取地址https://sonilo.com仅在启用 Sonilo 配乐或音效时使用",
"Sonilo API Key Required": "请先填写 Sonilo API Key否则生成时将回退到随机背景音乐",
"Sonilo BGM Prompt": "配乐风格提示(可选)",
"Sonilo BGM Prompt Help": "可选:描述期望的配乐风格,例如“舒缓钢琴”“紧张悬疑”,留空则完全根据画面生成",
"Sonilo config saved": "Sonilo 配置已保存",
"Sonilo AI Sound Effects": "AI 音效Sonilo",
"Sonilo SFX Help": "为成片自动生成贴合画面的音效(可选功能,默认关闭)",
"Sonilo SFX Notice": "启用后,合成完成的视频将上传至 Sonilo API根据画面内容生成音效并混在现有音轨之下解说配音仍按现有音量设置保持清晰。生成的音效为免版税素材。视频时长上限 3 分钟;生成失败时自动跳过音效,不影响成片。",
"Sonilo SFX Prompt": "音效风格提示(可选)",
"Sonilo SFX Prompt Help": "可选:描述期望的音效,例如“雨声和远处雷声”“金属碰撞”,留空则完全根据画面生成",
"Sonilo SFX API Key Required": "请先填写 Sonilo API Key否则生成时将跳过音效",
"Upload Background Music": "上传背景音乐",
"Background Music Path Help": "选择用于视频合成的背景音乐",
"No Background Music Resources Found": "未找到资源目录中的背景音乐,请上传背景音乐文件",