diff --git a/.gitignore b/.gitignore index 97a8cb6..9c8f917 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/app/models/schema.py b/app/models/schema.py index 5b16143..8bbc8a9 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -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 diff --git a/app/services/sonilo.py b/app/services/sonilo.py index 5b6e1fe..dab4ebe 100644 --- a/app/services/sonilo.py +++ b/app/services/sonilo.py @@ -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 Key（config.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 diff --git a/app/services/task.py b/app/services/task.py index ea5cdb3..d5e8fa9 100644 --- a/app/services/task.py +++ b/app/services/task.py @@ -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) # 获取优化的音量配置 diff --git a/config.example.toml b/config.example.toml index 57aeeed..5c4d177 100644 --- a/config.example.toml +++ b/config.example.toml @@ -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 获取地址： diff --git a/tests/test_sonilo_sfx_unittest.py b/tests/test_sonilo_sfx_unittest.py new file mode 100644 index 0000000..c227f19 --- /dev/null +++ b/tests/test_sonilo_sfx_unittest.py @@ -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() diff --git a/webui/components/audio_settings.py b/webui/components/audio_settings.py index d5e31a2..c0bf44c 100644 --- a/webui/components/audio_settings.py +++ b/webui/components/audio_settings.py @@ -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), } diff --git a/webui/i18n/en.json b/webui/i18n/en.json index ac1514b..11994ed 100644 --- a/webui/i18n/en.json +++ b/webui/i18n/en.json @@ -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.", diff --git a/webui/i18n/zh.json b/webui/i18n/zh.json index 076e76a..2ccac00 100644 --- a/webui/i18n/zh.json +++ b/webui/i18n/zh.json @@ -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": "未找到资源目录中的背景音乐，请上传背景音乐文件",