From f3fc7b888b5853cc0fc1b82f5f9b1ff72449bd0e Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Tue, 2 Jun 2026 13:29:49 +0800 Subject: [PATCH] feat: add RpaVideoSession for service-layer RPA video output Co-authored-by: Cursor --- src/jiangchang_skill_core/rpa/__init__.py | 6 + .../rpa/video_session.py | 289 ++++++++++++++++++ tests/test_video_session.py | 86 ++++++ 3 files changed, 381 insertions(+) create mode 100644 src/jiangchang_skill_core/rpa/video_session.py create mode 100644 tests/test_video_session.py diff --git a/src/jiangchang_skill_core/rpa/__init__.py b/src/jiangchang_skill_core/rpa/__init__.py index 284b12c..3979200 100644 --- a/src/jiangchang_skill_core/rpa/__init__.py +++ b/src/jiangchang_skill_core/rpa/__init__.py @@ -9,6 +9,11 @@ try: except ImportError: launch_persistent_browser = None # type: ignore[misc, assignment] +try: + from .video_session import RpaVideoSession +except ImportError: + RpaVideoSession = None # type: ignore[misc, assignment] + __all__ = [ "anti_detect", "artifacts", @@ -16,6 +21,7 @@ __all__ = [ "human_login", "stealth", "launch_persistent_browser", + "RpaVideoSession", # re-export common anti_detect helpers "random_delay", "human_delay_short", diff --git a/src/jiangchang_skill_core/rpa/video_session.py b/src/jiangchang_skill_core/rpa/video_session.py new file mode 100644 index 0000000..bcc4ea2 --- /dev/null +++ b/src/jiangchang_skill_core/rpa/video_session.py @@ -0,0 +1,289 @@ +"""RPA 运行录屏会话：Playwright raw video → 字幕 + 背景音乐 → 最终 MP4。""" + +from __future__ import annotations + +import asyncio +import os +import random +import subprocess +import sys +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .. import config + +# 与 screencast/composer.py 保持一致的字幕样式 +_JIANGCHANG_SUBTITLE_STYLE = ( + "FontName=Arial,FontSize=14," + "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000," + "Outline=2,Shadow=1,Alignment=2" +) + +_DEFAULT_MUSIC_ROOT = os.environ.get( + "MEDIA_ASSETS_ROOT", + r"D:\OpenClaw\client-commons\media-assets", +) + + +def _record_video_enabled() -> bool: + return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False) + + +def _resolve_music_file() -> Optional[Path]: + music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music" + if not music_dir.is_dir(): + return None + files = list(music_dir.rglob("*.mp3")) + if not files: + return None + return random.choice(files) + + +@dataclass +class _StepEntry: + start: float + text: str + duration: float = 4.0 + + +def _generate_srt(entries: List[_StepEntry], output_path: str) -> None: + """根据步骤时间线生成 SRT（防重叠）。""" + MIN_DURATION = 1.0 + GAP = 0.1 + steps = sorted(entries, key=lambda e: e.start) + if not steps: + steps = [_StepEntry(0.0, "任务执行中", 4.0)] + + for i in range(len(steps) - 1): + cur, nxt = steps[i], steps[i + 1] + max_end = nxt.start - GAP + new_dur = max_end - cur.start + if new_dur < MIN_DURATION: + cur.duration = MIN_DURATION + nxt.start = cur.start + MIN_DURATION + GAP + else: + cur.duration = min(cur.duration, new_dur) + + def _fmt(s: float) -> str: + h = int(s // 3600) + m = int((s % 3600) // 60) + sec = int(s % 60) + ms = int((s % 1) * 1000) + return f"{h:02d}:{m:02d}:{sec:02d},{ms:03d}" + + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w", encoding="utf-8") as f: + for i, e in enumerate(steps, 1): + f.write(f"{i}\n") + f.write(f"{_fmt(e.start)} --> {_fmt(e.start + e.duration)}\n") + f.write(f"{e.text}\n\n") + + +def _escape_srt_path(path: str) -> str: + return path.replace("\\", "/").replace(":", "\\:") + + +def _find_raw_webm(raw_dir: str) -> Optional[Path]: + root = Path(raw_dir) + if not root.is_dir(): + return None + webms = list(root.rglob("*.webm")) + if not webms: + return None + return max(webms, key=lambda p: p.stat().st_size) + + +def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]: + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + except FileNotFoundError: + return False, "ffmpeg not found" + if result.returncode != 0: + err = (result.stderr or result.stdout or "").strip() + return False, err[:500] if err else f"exit {result.returncode}" + return True, "" + + +def _compose_webm_mp4( + webm_path: Path, + srt_path: Path, + output_path: Path, + *, + music_file: Optional[Path] = None, + music_volume: float = 0.15, +) -> tuple[bool, str]: + output_path.parent.mkdir(parents=True, exist_ok=True) + srt_esc = _escape_srt_path(str(srt_path.resolve())) + subtitle_filter = ( + f"subtitles='{srt_esc}'" + 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]" + ) + cmd = [ + "ffmpeg", "-y", + "-i", str(webm_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 = [ + "ffmpeg", "-y", + "-i", str(webm_path), + "-vf", subtitle_filter, + "-c:v", "libx264", + "-preset", "fast", + "-crf", "23", + "-c:a", "aac", + "-b:a", "128k", + "-pix_fmt", "yuv420p", + str(output_path), + ] + return _run_ffmpeg(cmd) + + +class RpaVideoSession: + """Service/RPA 层统一录屏会话；OPENCLAW_RECORD_VIDEO=1 时启用完整成片流程。""" + + def __init__( + self, + *, + skill_slug: str, + skill_data_dir: str, + batch_id: str, + title: str = "", + ) -> None: + self.skill_slug = skill_slug + self.skill_data_dir = os.path.abspath(skill_data_dir) + self.batch_id = batch_id + self.title = title + self.enabled = _record_video_enabled() + self.warnings: List[str] = [] + self.output_video_path: Optional[str] = None + self.record_video_dir: Optional[str] = None + self.artifacts: Dict[str, Any] = {} + + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + safe_batch = "".join(c if c.isalnum() or c in "-_" else "_" for c in batch_id) + self._video_stem = f"{skill_slug}_{ts}_{safe_batch}" + + self._videos_dir = os.path.join(self.skill_data_dir, "videos") + self._artifact_root = os.path.join( + self.skill_data_dir, "rpa-artifacts", batch_id + ) + self._raw_video_dir = os.path.join(self._artifact_root, "raw-video") + self._subtitles_dir = os.path.join(self._artifact_root, "subtitles") + self._logs_dir = os.path.join(self._artifact_root, "logs") + self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt") + self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4") + + self._steps: List[_StepEntry] = [] + self._t0: Optional[float] = None + + if self.enabled: + os.makedirs(self._videos_dir, exist_ok=True) + os.makedirs(self._raw_video_dir, exist_ok=True) + os.makedirs(self._subtitles_dir, exist_ok=True) + os.makedirs(self._logs_dir, exist_ok=True) + self.record_video_dir = self._raw_video_dir + + def step(self, text: str, *, duration: float = 4.0) -> None: + """记录一步骤字幕（中文说明）。""" + if not self.enabled: + return + if self._t0 is None: + self._t0 = time.monotonic() + elapsed = time.monotonic() - self._t0 + self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration)) + + async def __aenter__(self) -> "RpaVideoSession": + if self.enabled: + self._t0 = time.monotonic() + if self.title: + self.step(self.title, duration=3.0) + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + if self.enabled: + await self.finalize() + + async def finalize(self) -> None: + """在 Playwright context 已 close 后合成最终 MP4；失败只记 warning。""" + if not self.enabled: + return + + if not self._steps and self.title: + self.step(self.title) + + try: + await asyncio.sleep(0.5) + webm = _find_raw_webm(self._raw_video_dir) + if webm is None: + self.warnings.append("no_playwright_raw_video") + self.artifacts = self.to_artifact_dict() + return + + entries = list(self._steps) + if entries: + entries[-1].duration = max(entries[-1].duration, 5.0) + _generate_srt(entries, self._srt_path) + + music = _resolve_music_file() + if music is None: + self.warnings.append("background_music_missing") + + ok, err = _compose_webm_mp4( + webm, + Path(self._srt_path), + Path(self._planned_output), + music_file=music, + ) + if not ok and music is not None: + self.warnings.append(f"ffmpeg_with_music_failed: {err}") + ok, err = _compose_webm_mp4( + webm, + Path(self._srt_path), + Path(self._planned_output), + music_file=None, + ) + if not ok: + self.warnings.append(f"ffmpeg_compose_failed: {err}") + self.artifacts = self.to_artifact_dict() + return + + self.output_video_path = str(Path(self._planned_output).resolve()) + self.artifacts = self.to_artifact_dict() + except Exception as exc: + self.warnings.append(f"video_finalize_error: {exc}") + self.artifacts = self.to_artifact_dict() + + def to_artifact_dict(self) -> Dict[str, Any]: + return { + "enabled": self.enabled, + "path": self.output_video_path, + "warnings": list(self.warnings), + "batch_id": self.batch_id, + } diff --git a/tests/test_video_session.py b/tests/test_video_session.py new file mode 100644 index 0000000..0547ed2 --- /dev/null +++ b/tests/test_video_session.py @@ -0,0 +1,86 @@ +# -*- coding: utf-8 -*- +"""RpaVideoSession 路径与 no-op 行为。""" +from __future__ import annotations + +import os +import tempfile +import unittest + +from jiangchang_skill_core import config +from jiangchang_skill_core.rpa.video_session import RpaVideoSession + + +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) + self.assertIsNone(session.record_video_dir) + self.assertIsNone(session.output_video_path) + finally: + if old is not None: + os.environ["OPENCLAW_RECORD_VIDEO"] = old + config.reset_cache() + + def test_enabled_paths(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.record_video_dir.replace("\\", "/").endswith( + "/rpa-artifacts/exmail_20260602_120000/raw-video" + ) + ) + self.assertTrue( + session._planned_output.replace("\\", "/").startswith( + tmp.replace("\\", "/") + "/videos/" + ) + ) + self.assertNotIn( + "rpa-artifacts", + session._planned_output.replace("\\", "/").split("/videos/")[-1], + ) + finally: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + config.reset_cache() + + def test_finalize_missing_webm_only_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", + ) + import asyncio + + asyncio.run(session.finalize()) + self.assertIn("no_playwright_raw_video", session.warnings) + self.assertIsNone(session.output_video_path) + finally: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + config.reset_cache() + + +if __name__ == "__main__": + unittest.main()