refactor: ffmpeg desktop capture replaces Playwright recording

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_video_session.py

@@ -1,10 +1,13 @@
 # -*- coding: utf-8 -*-
-"""RpaVideoSession 路径与 no-op 行为。"""
+"""RpaVideoSession：ffmpeg 桌面录屏，无 Playwright record_video_dir。"""
 from __future__ import annotations
 
+import asyncio
 import os
+import sys
 import tempfile
 import unittest
+from unittest.mock import MagicMock, patch
 
 from jiangchang_skill_core import config
 from jiangchang_skill_core.rpa.video_session import RpaVideoSession
@@ -27,12 +30,16 @@ class TestRpaVideoSession(unittest.TestCase):
                 self.assertFalse(session.enabled)
                 self.assertIsNone(session.record_video_dir)
                 self.assertIsNone(session.output_video_path)
+                self.assertIsNone(session.capture_path)
+                summ = session.summary()
+                self.assertFalse(summ["enabled"])
+                self.assertIsNone(summ["path"])
         finally:
             if old is not None:
                 os.environ["OPENCLAW_RECORD_VIDEO"] = old
             config.reset_cache()
 
-    def test_enabled_paths(self) -> None:
+    def test_enabled_paths_no_raw_video(self) -> None:
         os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
         try:
             with tempfile.TemporaryDirectory() as tmp:
@@ -44,9 +51,10 @@ class TestRpaVideoSession(unittest.TestCase):
                     title="接收订单",
                 )
                 self.assertTrue(session.enabled)
+                self.assertIsNone(session.record_video_dir)
                 self.assertTrue(
-                    session.record_video_dir.replace("\\", "/").endswith(
-                        "/rpa-artifacts/exmail_20260602_120000/raw-video"
+                    session._capture_path.replace("\\", "/").endswith(
+                        "/rpa-artifacts/exmail_20260602_120000/capture.mp4"
                     )
                 )
                 self.assertTrue(
@@ -54,15 +62,72 @@ class TestRpaVideoSession(unittest.TestCase):
                         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:
+    @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture")
+    @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value="ffmpeg")
+    @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen")
+    def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, _which: MagicMock) -> None:
+        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(video, "finalize", return_value=None):
+                            pass
+
+                asyncio.run(_run())
+                self.assertTrue(mock_popen.called)
+                cmd = mock_popen.call_args[0][0]
+                self.assertIn("ffmpeg", cmd)
+                self.assertIn("gdigrab", cmd)
+                self.assertIn("desktop", cmd)
+        finally:
+            os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+            config.reset_cache()
+
+    @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value=None)
+    def test_ffmpeg_not_found_warning(self, _which: MagicMock) -> None:
+        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_no_ff",
+                    ):
+                        pass
+
+                asyncio.run(_run())
+                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:
@@ -72,11 +137,11 @@ class TestRpaVideoSession(unittest.TestCase):
                     skill_data_dir=tmp,
                     batch_id="batch_warn",
                 )
-                import asyncio
-
                 asyncio.run(session.finalize())
-                self.assertIn("no_playwright_raw_video", session.warnings)
+                self.assertIn("ffmpeg_capture_missing", session.warnings)
                 self.assertIsNone(session.output_video_path)
+                summ = session.summary()
+                self.assertIsNone(summ["path"])
         finally:
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
             config.reset_cache()