mirror of
https://github.com/linyqh/NarratoAI.git
synced 2026-08-01 19:05:52 +00:00
14828256de
在 #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 完全一致
363 lines
15 KiB
Python
363 lines
15 KiB
Python
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()
|