# -*- coding: utf-8 -*- """RpaVideoSession：ffmpeg 桌面录屏、背景音乐循环淡出、旁白混音。""" from __future__ import annotations import asyncio import os import sys import tempfile import unittest from pathlib import Path from unittest.mock import MagicMock, patch from jiangchang_skill_core import config from jiangchang_skill_core.rpa import video_session as vs from jiangchang_skill_core.rpa.video_session import ( RpaVideoSession, _StepEntry, _compose_capture_mp4, _music_fade_params, _select_voiceover_clips, _voiceover_text_for_step, ) class TestRpaVideoSession(unittest.TestCase): def setUp(self) -> None: config.reset_cache() def test_disabled_is_noop(self) -> None: old = os.environ.pop("OPENCLAW_RECORD_VIDEO", None) try: with tempfile.TemporaryDirectory() as tmp: config.reset_cache() session = RpaVideoSession( skill_slug="receive-order", skill_data_dir=tmp, batch_id="batch_test", ) self.assertFalse(session.enabled) summ = session.summary() self.assertFalse(summ["enabled"]) self.assertIsNone(summ["path"]) self.assertIsNone(summ["capture_path"]) self.assertIsNone(summ["record_log_path"]) self.assertIsNone(summ["voiceover_path"]) self.assertIsNone(summ["music_path"]) finally: if old is not None: os.environ["OPENCLAW_RECORD_VIDEO"] = old config.reset_cache() def test_enabled_paths_use_ffmpeg_capture(self) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" try: with tempfile.TemporaryDirectory() as tmp: config.reset_cache() session = RpaVideoSession( skill_slug="receive-order", skill_data_dir=tmp, batch_id="exmail_20260602_120000", title="接收订单", ) self.assertTrue(session.enabled) self.assertTrue( session._capture_path.replace("\\", "/").endswith( "/rpa-artifacts/exmail_20260602_120000/capture.mp4" ) ) self.assertTrue( session._ffmpeg_record_log_path.replace("\\", "/").endswith( "/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log" ) ) self.assertTrue( session._voiceover_path.replace("\\", "/").endswith( "/rpa-artifacts/exmail_20260602_120000/voiceover/voiceover.wav" ) ) self.assertTrue( session._planned_output.replace("\\", "/").startswith( tmp.replace("\\", "/") + "/videos/" ) ) finally: os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture") @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe") @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen") def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, mock_resolve: MagicMock) -> None: mock_resolve.return_value = Path(r"C:\media\bin\win32-x64\ffmpeg.exe") proc = MagicMock() proc.poll.return_value = None proc.stdin = MagicMock() mock_popen.return_value = proc os.environ["OPENCLAW_RECORD_VIDEO"] = "1" try: with tempfile.TemporaryDirectory() as tmp: config.reset_cache() async def _run() -> None: async with RpaVideoSession( skill_slug="receive-order", skill_data_dir=tmp, batch_id="batch_ff", ) as video: video.add_step("测试步骤") with patch.object(RpaVideoSession, "finalize", return_value=None): asyncio.run(_run()) self.assertTrue(mock_popen.called) cmd = mock_popen.call_args[0][0] self.assertTrue(str(cmd[0]).endswith("ffmpeg.exe")) self.assertIn("gdigrab", cmd) self.assertIn("desktop", cmd) kwargs = mock_popen.call_args[1] self.assertIsNot(kwargs.get("stderr"), os.devnull) finally: os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe", return_value=None) def test_ffmpeg_not_found_warning(self, _resolve: MagicMock) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" try: with tempfile.TemporaryDirectory() as tmp: config.reset_cache() session = RpaVideoSession( skill_slug="receive-order", skill_data_dir=tmp, batch_id="batch_no_ff", ) session._start_ffmpeg_capture() self.assertIn("ffmpeg_not_found", session.warnings) finally: os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() def test_finalize_missing_capture_warning(self) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" try: with tempfile.TemporaryDirectory() as tmp: config.reset_cache() session = RpaVideoSession( skill_slug="receive-order", skill_data_dir=tmp, batch_id="batch_warn", ) asyncio.run(session.finalize()) self.assertTrue( any("ffmpeg_capture_missing" in w for w in session.warnings) ) self.assertIn("record_log=", "".join(session.warnings)) self.assertIsNone(session.output_video_path) finally: os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() class TestMusicFadeParams(unittest.TestCase): def test_normal_duration(self) -> None: start, dur = _music_fade_params(20.0) self.assertAlmostEqual(start, 17.0) self.assertAlmostEqual(dur, 3.0) def test_short_video(self) -> None: start, dur = _music_fade_params(4.0) self.assertAlmostEqual(dur, 1.0) self.assertAlmostEqual(start, 3.0) def test_invalid_duration(self) -> None: self.assertEqual(_music_fade_params(None), (None, None)) self.assertEqual(_music_fade_params(0), (None, None)) self.assertEqual(_music_fade_params(-1), (None, None)) class TestVoiceoverTextFilter(unittest.TestCase): def test_keeps_meaningful_steps(self) -> None: for text in ( "开始采集", "获取1688账号", "启动浏览器", "检查登录状态", "搜索关键词：螺丝", "解析第 1 页店铺列表", "打开店铺联系方式页", "提取联系人信息", "采集完成", ): self.assertIsNotNone(_voiceover_text_for_step(text), msg=text) def test_skips_noise(self) -> None: for text in ( "等待 1-5s", "跳过重复店铺", "写入联系人库", "探针模式", "操作异常", "登录失败", "2024-01-01 12:00:00 INFO: debug", "", " ", ): self.assertIsNone(_voiceover_text_for_step(text), msg=text) def test_truncates_long_text(self) -> None: long_text = "开" * 80 result = _voiceover_text_for_step(long_text) self.assertIsNotNone(result) assert result is not None self.assertEqual(len(result), 60) def test_dedup_within_ten_seconds(self) -> None: entries = [ _StepEntry(0.0, "启动浏览器"), _StepEntry(5.0, "启动浏览器"), _StepEntry(12.0, "启动浏览器"), ] clips = _select_voiceover_clips(entries) texts = [t for _, t in clips] self.assertEqual(texts.count("启动浏览器"), 2) def test_min_gap_two_seconds(self) -> None: entries = [ _StepEntry(0.0, "开始采集"), _StepEntry(1.0, "启动浏览器"), _StepEntry(3.0, "检查登录状态"), ] clips = _select_voiceover_clips(entries) self.assertEqual(len(clips), 2) self.assertEqual(clips[0][1], "开始采集") self.assertEqual(clips[1][1], "检查登录状态") class TestComposeCaptureMp4(unittest.TestCase): def setUp(self) -> None: self.ffmpeg = Path(r"C:\ffmpeg\ffmpeg.exe") self.capture = Path(r"C:\tmp\capture.mp4") self.srt = Path(r"C:\tmp\sub.srt") self.output = Path(r"C:\tmp\out.mp4") self.music = Path(r"C:\tmp\music.mp3") self.voice = Path(r"C:\tmp\voiceover.wav") def _run_compose(self, **kwargs) -> list[str]: captured: list[str] = [] def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]: captured.extend(cmd_tail) return True, "" with ( patch.object(vs, "_run_ffmpeg", side_effect=fake_run), patch.object(vs, "_resolve_ffprobe_exe", return_value=Path(r"C:\ffmpeg\ffprobe.exe")), patch.object(vs, "_probe_media_duration_seconds", return_value=kwargs.pop("duration", 20.0)), patch.object(Path, "is_file", return_value=True), ): _compose_capture_mp4( self.ffmpeg, self.capture, self.srt, self.output, music_file=kwargs.get("music_file"), voiceover_file=kwargs.get("voiceover_file"), warnings=kwargs.get("warnings"), ) return captured def test_music_loop_params(self) -> None: cmd = self._run_compose(music_file=self.music) cmd_str = " ".join(cmd) loop_idx = cmd.index("-stream_loop") minus_one_idx = cmd.index("-1", loop_idx) music_i_idx = cmd.index("-i", minus_one_idx) self.assertEqual(cmd[music_i_idx + 1], str(self.music)) self.assertIn("-shortest", cmd) filter_idx = cmd.index("-filter_complex") filt = cmd[filter_idx + 1] self.assertNotIn("apad", filt) def test_music_fade_filter_long_video(self) -> None: cmd = self._run_compose(music_file=self.music, duration=20.0) filt = cmd[cmd.index("-filter_complex") + 1] self.assertIn("afade=t=out:st=17", filt) self.assertIn("d=3", filt) self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt) def test_music_fade_filter_short_video(self) -> None: cmd = self._run_compose(music_file=self.music, duration=4.0) filt = cmd[cmd.index("-filter_complex") + 1] self.assertIn("afade=t=out:st=3", filt) self.assertIn("d=1", filt) def test_duration_probe_failed_no_afade(self) -> None: warnings: list[str] = [] captured: list[str] = [] def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]: captured.extend(cmd_tail) return True, "" with ( patch.object(vs, "_run_ffmpeg", side_effect=fake_run), patch.object(vs, "_resolve_ffprobe_exe", return_value=None), patch.object(vs, "_probe_media_duration_seconds", return_value=None), patch.object(Path, "is_file", return_value=True), ): ok, _ = _compose_capture_mp4( self.ffmpeg, self.capture, self.srt, self.output, music_file=self.music, warnings=warnings, ) self.assertTrue(ok) self.assertIn("video_duration_probe_failed_for_music_fade", warnings) filt = captured[captured.index("-filter_complex") + 1] self.assertNotIn("afade", filt) self.assertIn("-stream_loop", captured) def test_music_and_voice_amix(self) -> None: cmd = self._run_compose(music_file=self.music, voiceover_file=self.voice) filt = cmd[cmd.index("-filter_complex") + 1] self.assertIn("amix", filt) self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt) self.assertIn(f"volume={vs._VOICE_VOLUME}", filt) self.assertLess(vs._MUSIC_VOLUME, vs._VOICE_VOLUME) self.assertIn("afade", filt) self.assertNotIn("apad", filt) self.assertIn("-stream_loop", cmd) self.assertIn("-shortest", cmd) def test_voice_only_no_amix(self) -> None: cmd = self._run_compose(voiceover_file=self.voice) filt = cmd[cmd.index("-filter_complex") + 1] self.assertNotIn("amix", filt) self.assertIn(f"volume={vs._VOICE_VOLUME}", filt) self.assertIn("apad", filt) self.assertIn("-shortest", cmd) self.assertNotIn("-stream_loop", cmd) class TestFinalizeDegradation(unittest.TestCase): def setUp(self) -> None: config.reset_cache() def tearDown(self) -> None: os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe") @patch("jiangchang_skill_core.rpa.video_session._resolve_music_file") @patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover") @patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4") def test_tts_failure_still_composes( self, mock_compose: MagicMock, mock_tts: MagicMock, mock_music: MagicMock, mock_ffmpeg: MagicMock, ) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe") mock_music.return_value = Path(r"C:\tmp\music.mp3") mock_tts.return_value = None mock_compose.return_value = (True, "") with tempfile.TemporaryDirectory() as tmp: capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4" capture.parent.mkdir(parents=True) capture.write_bytes(b"fake") session = RpaVideoSession( skill_slug="test", skill_data_dir=tmp, batch_id="batch", ) session.add_step("启动浏览器") asyncio.run(session.finalize()) self.assertTrue(mock_compose.called) self.assertIsNotNone(session.output_video_path) @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe") @patch("jiangchang_skill_core.rpa.video_session._resolve_music_file") @patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover") @patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4") def test_compose_fallback_without_music( self, mock_compose: MagicMock, mock_tts: MagicMock, mock_music: MagicMock, mock_ffmpeg: MagicMock, ) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe") mock_music.return_value = Path(r"C:\tmp\music.mp3") mock_tts.return_value = None mock_compose.side_effect = [(False, "music err"), (True, "")] with tempfile.TemporaryDirectory() as tmp: capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4" capture.parent.mkdir(parents=True) capture.write_bytes(b"fake") session = RpaVideoSession( skill_slug="test", skill_data_dir=tmp, batch_id="batch", ) asyncio.run(session.finalize()) self.assertEqual(mock_compose.call_count, 2) self.assertTrue(any("ffmpeg_with_music_failed" in w for w in session.warnings)) if __name__ == "__main__": unittest.main()