From 0e2b73e1ad374017a3de60098c834567ca9f95aa Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Mon, 8 Jun 2026 10:14:31 +0800 Subject: [PATCH] feat: improve rpa video audio composition Co-authored-by: Cursor --- pyproject.toml | 2 +- src/jiangchang_desktop_sdk/__init__.py | 2 +- .../rpa/video_session.py | 528 +++++++++++++++--- tests/test_video_session.py | 277 ++++++++- 4 files changed, 738 insertions(+), 71 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ae0e2e8..231063d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.0.11" +version = "1.0.12" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_desktop_sdk/__init__.py b/src/jiangchang_desktop_sdk/__init__.py index ef9f258..87c2562 100644 --- a/src/jiangchang_desktop_sdk/__init__.py +++ b/src/jiangchang_desktop_sdk/__init__.py @@ -23,7 +23,7 @@ __all__ = [ "LaunchError", "GatewayDownError", ] -__version__ = "1.0.11" +__version__ = "1.0.12" def __getattr__(name: str): diff --git a/src/jiangchang_skill_core/rpa/video_session.py b/src/jiangchang_skill_core/rpa/video_session.py index 45d2d71..12ec5c9 100644 --- a/src/jiangchang_skill_core/rpa/video_session.py +++ b/src/jiangchang_skill_core/rpa/video_session.py @@ -1,9 +1,14 @@ -"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。""" +"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 旁白 + 背景音乐 → 最终 MP4。 + +背景音乐默认循环铺满录屏时长并在结尾淡出；旁白使用 Windows 本地 SAPI 合成， +失败时静默降级，不影响视频生成。Skill 无需额外配置。 +""" from __future__ import annotations import asyncio import os +import re import subprocess import sys import time @@ -18,6 +23,7 @@ from ..media_assets import ( ensure_media_assets, pick_background_music, resolve_ffmpeg, + resolve_ffprobe, ) _JIANGCHANG_SUBTITLE_STYLE = ( @@ -28,6 +34,31 @@ _JIANGCHANG_SUBTITLE_STYLE = ( _CAPTURE_FRAMERATE = 15 +# 背景音乐循环 + 结尾淡出（默认策略，非 env 配置） +_MUSIC_VOLUME = 0.12 +_VOICE_VOLUME = 1.0 +_MUSIC_FADE_OUT_SECONDS = 3.0 +_MIN_MUSIC_FADE_OUT_SECONDS = 1.0 +_SHORT_VIDEO_FADE_RATIO = 0.25 + +_VOICEOVER_MAX_CHARS = 60 +_VOICEOVER_DEDUP_SECONDS = 10.0 +_VOICEOVER_MIN_GAP_SECONDS = 2.0 + +_VOICEOVER_SKIP_SUBSTRINGS = ( + "等待 1-5s", + "跳过重复", + "写入联系人库", + "探针模式", + "异常", + "失败", +) + +# 明显技术日志：时间戳前缀、纯数字进度等 +_TECH_LOG_RE = re.compile( + r"^\d{4}-\d{2}-\d{2}|^\[\w+\]|^DEBUG:|^INFO:|^WARN:|^ERROR:" +) + def _record_video_enabled() -> bool: return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False) @@ -41,6 +72,68 @@ def _resolve_ffmpeg_exe() -> Optional[Path]: return path if path.is_file() else None +def _resolve_ffprobe_exe() -> Optional[Path]: + """定位 ffprobe；优先 media_assets bundle，否则尝试 ffmpeg 同目录。""" + ffprobe = resolve_ffprobe() + if ffprobe is not None: + path = Path(ffprobe) + if path.is_file(): + return path + ffmpeg = _resolve_ffmpeg_exe() + if ffmpeg is not None: + sibling = ffmpeg.parent / ("ffprobe.exe" if sys.platform == "win32" else "ffprobe") + if sibling.is_file(): + return sibling + return None + + +def _probe_media_duration_seconds( + ffprobe_exe: Optional[Path], media_path: Path +) -> Optional[float]: + if ffprobe_exe is None or not media_path.is_file(): + return None + try: + result = subprocess.run( + [ + str(ffprobe_exe), + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=nk=1:nw=1", + str(media_path), + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=30, + ) + if result.returncode != 0: + return None + raw = (result.stdout or "").strip() + if not raw: + return None + value = float(raw) + return value if value > 0 else None + except Exception: + return None + + +def _music_fade_params( + duration_seconds: Optional[float], +) -> tuple[Optional[float], Optional[float]]: + if duration_seconds is None or duration_seconds <= 0: + return (None, None) + fade = min( + _MUSIC_FADE_OUT_SECONDS, + max(_MIN_MUSIC_FADE_OUT_SECONDS, duration_seconds * _SHORT_VIDEO_FADE_RATIO), + ) + fade_start = max(0.0, duration_seconds - fade) + return (fade_start, fade) + + def _resolve_music_file(warnings: List[str]) -> Optional[Path]: music = pick_background_music() if music is not None: @@ -59,6 +152,173 @@ class _StepEntry: duration: float = 4.0 +@dataclass +class _VoiceClip: + path: Path + delay_ms: int + + +def _voiceover_text_for_step(text: str) -> Optional[str]: + """过滤适合朗读的步骤文本；噪音/技术日志返回 None。""" + cleaned = (text or "").strip() + if not cleaned: + return None + if _TECH_LOG_RE.search(cleaned): + return None + for skip in _VOICEOVER_SKIP_SUBSTRINGS: + if skip in cleaned: + return None + if len(cleaned) > _VOICEOVER_MAX_CHARS: + cleaned = cleaned[:_VOICEOVER_MAX_CHARS] + return cleaned + + +def _select_voiceover_clips( + entries: List[_StepEntry], +) -> List[tuple[_StepEntry, str]]: + """按时间排序，去重并保证旁白间隔。""" + result: List[tuple[_StepEntry, str]] = [] + last_text_at: Dict[str, float] = {} + last_spoken_at = -_VOICEOVER_MIN_GAP_SECONDS + + for entry in sorted(entries, key=lambda e: e.start): + spoken = _voiceover_text_for_step(entry.text) + if not spoken: + continue + prev = last_text_at.get(spoken) + if prev is not None and entry.start - prev < _VOICEOVER_DEDUP_SECONDS: + continue + if entry.start - last_spoken_at < _VOICEOVER_MIN_GAP_SECONDS: + continue + last_text_at[spoken] = entry.start + last_spoken_at = entry.start + result.append((entry, spoken)) + return result + + +def _tts_available() -> bool: + return sys.platform == "win32" + + +def _sapi_synthesize_wav(text: str, output_wav: Path) -> bool: + if not _tts_available(): + return False + output_wav.parent.mkdir(parents=True, exist_ok=True) + out_str = str(output_wav.resolve()).replace("'", "''") + text_esc = text.replace("'", "''") + ps_script = ( + "Add-Type -AssemblyName System.Speech; " + "$s = New-Object System.Speech.Synthesis.SpeechSynthesizer; " + f"$s.SetOutputToWaveFile('{out_str}'); " + f"$s.Speak('{text_esc}'); " + "$s.Dispose()" + ) + try: + result = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_script], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=120, + ) + return ( + result.returncode == 0 + and output_wav.is_file() + and output_wav.stat().st_size > 0 + ) + except Exception: + return False + + +def _mix_voice_clips( + ffmpeg_exe: Path, clips: List[_VoiceClip], output_wav: Path +) -> bool: + if not clips: + return False + output_wav.parent.mkdir(parents=True, exist_ok=True) + + filter_parts: List[str] = [] + mix_labels: List[str] = [] + cmd_inputs: List[str] = [] + + for i, clip in enumerate(clips): + cmd_inputs.extend(["-i", str(clip.path)]) + label = f"v{i}" + filter_parts.append( + f"[{i}:a]adelay={clip.delay_ms}|{clip.delay_ms}," + f"volume={_VOICE_VOLUME}[{label}]" + ) + mix_labels.append(f"[{label}]") + + if len(clips) == 1: + filter_complex = filter_parts[0].replace(f"[v0]", "[out]") + # single clip: relabel output + filter_complex = ( + f"[0:a]adelay={clips[0].delay_ms}|{clips[0].delay_ms}," + f"volume={_VOICE_VOLUME}[out]" + ) + else: + filter_complex = ( + ";".join(filter_parts) + + ";" + + "".join(mix_labels) + + f"amix=inputs={len(clips)}:duration=longest:dropout_transition=0:normalize=0[out]" + ) + + cmd_tail = [ + "-y", + *cmd_inputs, + "-filter_complex", + filter_complex, + "-map", + "[out]", + str(output_wav), + ] + ok, _ = _run_ffmpeg(ffmpeg_exe, cmd_tail) + return ok and output_wav.is_file() + + +def _synthesize_step_voiceover( + entries: List[_StepEntry], + output_wav: Path, + warnings: List[str], + *, + ffmpeg_exe: Optional[Path] = None, +) -> Optional[Path]: + """本地 Windows SAPI 逐条合成旁白并按步骤 start 时间对齐；失败返回 None。""" + if not _tts_available(): + warnings.append("tts_unsupported_platform") + return None + + clips_to_speak = _select_voiceover_clips(entries) + if not clips_to_speak: + return None + + voice_dir = output_wav.parent + voice_dir.mkdir(parents=True, exist_ok=True) + temp_clips: List[_VoiceClip] = [] + + for i, (entry, spoken) in enumerate(clips_to_speak): + clip_path = voice_dir / f"voice_{i:03d}.wav" + if not _sapi_synthesize_wav(spoken, clip_path): + warnings.append("tts_synthesis_failed") + return None + delay_ms = max(0, int(entry.start * 1000)) + temp_clips.append(_VoiceClip(path=clip_path, delay_ms=delay_ms)) + + if ffmpeg_exe is None: + ffmpeg_exe = _resolve_ffmpeg_exe() + if ffmpeg_exe is None: + warnings.append("tts_mix_ffmpeg_missing") + return None + + if not _mix_voice_clips(ffmpeg_exe, temp_clips, output_wav): + warnings.append("tts_mix_failed") + return None + return output_wav + + def _generate_srt(entries: List[_StepEntry], output_path: str) -> None: MIN_DURATION = 1.0 GAP = 0.1 @@ -117,27 +377,16 @@ def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: List[str]) -> tuple[bool, str]: return True, "" -def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]: - """Windows gdigrab 全桌面录制；非 Windows 返回 None。""" - if sys.platform != "win32": - return None - return _ffmpeg_cmd( - ffmpeg_exe, - "-y", - "-f", - "gdigrab", - "-framerate", - str(_CAPTURE_FRAMERATE), - "-i", - "desktop", - "-c:v", - "libx264", - "-preset", - "ultrafast", - "-pix_fmt", - "yuv420p", - capture_path, - ) +def _build_music_filter( + audio_input_idx: int, + fade_start: Optional[float], + fade_duration: Optional[float], + output_label: str = "music", +) -> str: + filt = f"[{audio_input_idx}:a]volume={_MUSIC_VOLUME}" + if fade_start is not None and fade_duration is not None: + filt += f",afade=t=out:st={fade_start}:d={fade_duration}" + return filt + f"[{output_label}]" def _compose_capture_mp4( @@ -147,7 +396,8 @@ def _compose_capture_mp4( output_path: Path, *, music_file: Optional[Path] = None, - music_volume: float = 0.15, + voiceover_file: Optional[Path] = None, + warnings: Optional[List[str]] = None, ) -> tuple[bool, str]: output_path.parent.mkdir(parents=True, exist_ok=True) srt_esc = _escape_srt_path(str(srt_path.resolve())) @@ -156,38 +406,77 @@ def _compose_capture_mp4( f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'" ) - if music_file and music_file.is_file(): - filter_complex = ( - f"[0:v]{subtitle_filter}[vout];" - f"[1:a]volume={music_volume},apad[aout]" + has_music = music_file is not None and music_file.is_file() + has_voice = voiceover_file is not None and voiceover_file.is_file() + + fade_start: Optional[float] = None + fade_duration: Optional[float] = None + if has_music: + ffprobe = _resolve_ffprobe_exe() + duration = _probe_media_duration_seconds(ffprobe, capture_path) + fade_start, fade_duration = _music_fade_params(duration) + if fade_start is None and warnings is not None: + warnings.append("video_duration_probe_failed_for_music_fade") + + if has_music or has_voice: + cmd_tail: List[str] = ["-y", "-i", str(capture_path)] + input_idx = 1 + filter_parts: List[str] = [f"[0:v]{subtitle_filter}[vout]"] + audio_labels: List[str] = [] + + if has_music: + cmd_tail.extend(["-stream_loop", "-1", "-i", str(music_file)]) + music_label = "aout" if not has_voice else "music" + filter_parts.append( + _build_music_filter(input_idx, fade_start, fade_duration, music_label) + ) + if has_voice: + audio_labels.append(f"[{music_label}]") + input_idx += 1 + + if has_voice: + cmd_tail.extend(["-i", str(voiceover_file)]) + if has_music: + filter_parts.append( + f"[{input_idx}:a]volume={_VOICE_VOLUME}[voice]" + ) + audio_labels.append("[voice]") + else: + # voice-only：apad 补静音，避免 -shortest 按旁白长度截短视频 + filter_parts.append( + f"[{input_idx}:a]volume={_VOICE_VOLUME},apad[aout]" + ) + input_idx += 1 + + if len(audio_labels) == 2: + filter_parts.append( + "".join(audio_labels) + "amix=inputs=2:normalize=0[aout]" + ) + + cmd_tail.extend( + [ + "-filter_complex", + ";".join(filter_parts), + "-map", + "[vout]", + "-map", + "[aout]", + "-c:v", + "libx264", + "-preset", + "fast", + "-crf", + "23", + "-c:a", + "aac", + "-b:a", + "128k", + "-shortest", + "-pix_fmt", + "yuv420p", + str(output_path), + ] ) - cmd_tail = [ - "-y", - "-i", - str(capture_path), - "-i", - str(music_file), - "-filter_complex", - filter_complex, - "-map", - "[vout]", - "-map", - "[aout]", - "-c:v", - "libx264", - "-preset", - "fast", - "-crf", - "23", - "-c:a", - "aac", - "-b:a", - "128k", - "-shortest", - "-pix_fmt", - "yuv420p", - str(output_path), - ] else: cmd_tail = [ "-y", @@ -212,6 +501,29 @@ def _compose_capture_mp4( return _run_ffmpeg(ffmpeg_exe, cmd_tail) +def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]: + """Windows gdigrab 全桌面录制；非 Windows 返回 None。""" + if sys.platform != "win32": + return None + return _ffmpeg_cmd( + ffmpeg_exe, + "-y", + "-f", + "gdigrab", + "-framerate", + str(_CAPTURE_FRAMERATE), + "-i", + "desktop", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-pix_fmt", + "yuv420p", + capture_path, + ) + + class RpaVideoSession: """Service/RPA 层统一 ffmpeg 桌面录屏；OPENCLAW_RECORD_VIDEO=1 时启用成片流程。""" @@ -242,12 +554,17 @@ class RpaVideoSession: self.skill_data_dir, "rpa-artifacts", batch_id ) self._subtitles_dir = os.path.join(self._artifact_root, "subtitles") + self._voiceover_dir = os.path.join(self._artifact_root, "voiceover") self._logs_dir = os.path.join(self._artifact_root, "logs") self._ffmpeg_record_log_path = os.path.join(self._logs_dir, "ffmpeg-record.log") self._capture_path = os.path.join(self._artifact_root, "capture.mp4") self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt") + self._voiceover_path = os.path.join(self._voiceover_dir, "voiceover.wav") self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4") + self._music_path: Optional[str] = None + self._voiceover_output: Optional[str] = None + self._steps: List[_StepEntry] = [] self._t0: Optional[float] = None self._ffmpeg_proc: Optional[subprocess.Popen] = None @@ -256,6 +573,7 @@ class RpaVideoSession: if self.enabled: os.makedirs(self._videos_dir, exist_ok=True) os.makedirs(self._subtitles_dir, exist_ok=True) + os.makedirs(self._voiceover_dir, exist_ok=True) os.makedirs(self._logs_dir, exist_ok=True) os.makedirs(self._artifact_root, exist_ok=True) @@ -379,6 +697,54 @@ class RpaVideoSession: self._stop_ffmpeg_capture() await self.finalize() + def _try_compose_variants( + self, + ffmpeg_exe: Path, + capture: Path, + srt_file: Path, + output: Path, + *, + music: Optional[Path], + voiceover: Optional[Path], + use_srt: bool, + ) -> tuple[bool, str]: + """按 music/voice 组合尝试合成，失败时逐级降级。""" + srt = srt_file if use_srt else Path(self._srt_path) + variants: List[tuple[Optional[Path], Optional[Path]]] = [ + (music, voiceover), + ] + if music and voiceover: + variants.append((None, voiceover)) + if music: + variants.append((music, None)) + variants.append((None, None)) + + seen: set[tuple[Optional[str], Optional[str]]] = set() + last_err = "" + for m, v in variants: + key = (str(m) if m else None, str(v) if v else None) + if key in seen: + continue + seen.add(key) + ok, err = _compose_capture_mp4( + ffmpeg_exe, + capture, + srt, + output, + music_file=m, + voiceover_file=v, + warnings=self.warnings, + ) + if ok: + return True, "" + last_err = err + if m is not None and v is not None: + self.warnings.append(f"ffmpeg_with_music_and_voice_failed: {err}") + elif m is not None: + self.warnings.append(f"ffmpeg_with_music_failed: {err}") + + return False, last_err + async def finalize(self) -> None: """停止录制后合成最终 MP4；失败只记 warning，不抛异常。""" if not self.enabled: @@ -406,7 +772,12 @@ class RpaVideoSession: self.warnings.append(f"subtitle_write_failed: {exc}") music = _resolve_music_file(self.warnings) - if music is None and "background_music_lfs_pointer_only" not in self.warnings and "background_music_invalid" not in self.warnings: + if music is not None: + self._music_path = str(music.resolve()) + elif ( + "background_music_lfs_pointer_only" not in self.warnings + and "background_music_invalid" not in self.warnings + ): self.warnings.append("background_music_missing") ffmpeg_exe = _resolve_ffmpeg_exe() @@ -415,25 +786,32 @@ class RpaVideoSession: self.artifacts = self.summary() return + voiceover: Optional[Path] = None + try: + vo = _synthesize_step_voiceover( + entries, + Path(self._voiceover_path), + self.warnings, + ffmpeg_exe=ffmpeg_exe, + ) + if vo is not None: + voiceover = vo + self._voiceover_output = str(vo.resolve()) + except Exception as exc: + self.warnings.append(f"tts_unexpected_error: {exc}") + srt_file = Path(self._srt_path) use_srt = srt_file.is_file() and srt_file.stat().st_size > 0 - ok, err = _compose_capture_mp4( + ok, err = self._try_compose_variants( ffmpeg_exe, capture, - srt_file if use_srt else Path(self._srt_path), + srt_file, Path(self._planned_output), - music_file=music, + music=music, + voiceover=voiceover, + use_srt=use_srt, ) - if not ok and music is not None: - self.warnings.append(f"ffmpeg_with_music_failed: {err}") - ok, err = _compose_capture_mp4( - ffmpeg_exe, - capture, - srt_file if use_srt else Path(self._srt_path), - Path(self._planned_output), - music_file=None, - ) if not ok and not use_srt: ok, err = _run_ffmpeg( ffmpeg_exe, @@ -469,12 +847,28 @@ class RpaVideoSession: if self.enabled and self._ffmpeg_record_log_path else None ) + audio_warnings = [ + w + for w in self.warnings + if w.startswith( + ( + "tts_", + "video_duration_probe_failed", + "ffmpeg_with_music", + "background_music_", + ) + ) + or w == "background_music_missing" + ] return { "enabled": self.enabled, "path": self.output_video_path, "capture_path": self.capture_path, "record_log_path": record_log, "warnings": list(self.warnings), + "voiceover_path": self._voiceover_output, + "music_path": self._music_path, + "audio_warnings": audio_warnings, } def to_artifact_dict(self) -> Dict[str, Any]: diff --git a/tests/test_video_session.py b/tests/test_video_session.py index be41f28..086fd9a 100644 --- a/tests/test_video_session.py +++ b/tests/test_video_session.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""RpaVideoSession：ffmpeg 桌面录屏。""" +"""RpaVideoSession：ffmpeg 桌面录屏、背景音乐循环淡出、旁白混音。""" from __future__ import annotations import asyncio @@ -9,8 +9,17 @@ 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.video_session import RpaVideoSession +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): @@ -33,6 +42,8 @@ class TestRpaVideoSession(unittest.TestCase): 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 @@ -60,6 +71,11 @@ class TestRpaVideoSession(unittest.TestCase): "/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/" @@ -144,5 +160,262 @@ class TestRpaVideoSession(unittest.TestCase): 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()