feat(tools): add screencast recording engine

Adds tools/screencast/ — a reusable screen-recording pipeline for
skill demo videos. Wraps a pytest desktop E2E run with:
  - ScreenRecorder: mss-based full-screen frame capture (background thread)
  - SubtitleEngine: stdout keyword → timestamped SRT generation
  - compose_video: FFmpeg filter_complex merge (video + subtitles + BGM)
  - run_screencast: main orchestrator wiring all three layers

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tools/screencast/__init__.py

@@ -0,0 +1,3 @@
+from .runner import run_screencast
+
+__all__ = ["run_screencast"]

tools/screencast/composer.py

@@ -0,0 +1,71 @@
+"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
+from __future__ import annotations
+
+import random
+import subprocess
+import sys
+from pathlib import Path
+
+
+def compose_video(
+    frames_dir: str,
+    fps: int,
+    subtitle_path: str,
+    music_dir: str,
+    output_path: str,
+    music_volume: float = 0.15,
+) -> str:
+    frames_dir = Path(frames_dir)
+    subtitle_path = Path(subtitle_path)
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # 随机选一首背景音乐
+    music_files = list(Path(music_dir).rglob("*.mp3"))
+    if not music_files:
+        raise FileNotFoundError(f"music_dir 下没有找到 MP3 文件: {music_dir}")
+    music_file = random.choice(music_files)
+    print(f"[screencast] 背景音乐: {music_file.name}")
+
+    # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法）
+    srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:")
+
+    subtitle_filter = (
+        f"subtitles='{srt_path_str}'"
+        ":force_style='FontName=Arial,FontSize=28,"
+        "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
+        "Outline=2,Shadow=1,Alignment=2'"
+    )
+
+    # 视频和音频都放进 filter_complex，避免 -vf 与 -filter_complex 混用报错
+    filter_complex = (
+        f"[0:v]{subtitle_filter}[vout];"
+        f"[1:a]volume={music_volume},apad[aout]"
+    )
+
+    cmd = [
+        "ffmpeg", "-y",
+        "-framerate", str(fps),
+        "-i", str(frames_dir / "frame_%06d.png"),
+        "-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),
+    ]
+
+    print(f"[screencast] 运行 FFmpeg 合成...")
+    result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
+    if result.returncode != 0:
+        print(result.stderr, file=sys.stderr)
+        raise RuntimeError(f"FFmpeg 合成失败，退出码 {result.returncode}")
+
+    print(f"[screencast] 输出: {output_path}")
+    return str(output_path)

tools/screencast/recorder.py

@@ -0,0 +1,48 @@
+"""全屏截帧引擎，后台线程持续截图存为 PNG 帧序列。"""
+from __future__ import annotations
+
+import threading
+import time
+from pathlib import Path
+
+import mss
+import mss.tools
+
+
+class ScreenRecorder:
+    def __init__(self, frames_dir: str, fps: int = 10):
+        self._frames_dir = Path(frames_dir)
+        self._fps = fps
+        self._stop_event = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._frame_count = 0
+
+    def start(self) -> None:
+        self._frames_dir.mkdir(parents=True, exist_ok=True)
+        self._stop_event.clear()
+        self._frame_count = 0
+        self._thread = threading.Thread(target=self._capture_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop_event.set()
+        if self._thread:
+            self._thread.join(timeout=10)
+
+    @property
+    def frame_count(self) -> int:
+        return self._frame_count
+
+    def _capture_loop(self) -> None:
+        interval = 1.0 / self._fps
+        with mss.mss() as sct:
+            monitor = sct.monitors[0]  # 全屏（monitor[0] 是所有显示器合并区域）
+            while not self._stop_event.is_set():
+                t0 = time.perf_counter()
+                img = sct.grab(monitor)
+                path = self._frames_dir / f"frame_{self._frame_count:06d}.png"
+                mss.tools.to_png(img.rgb, img.size, output=str(path))
+                self._frame_count += 1
+                elapsed = time.perf_counter() - t0
+                sleep = max(0.0, interval - elapsed)
+                time.sleep(sleep)

tools/screencast/requirements.txt

@@ -0,0 +1 @@
+mss>=9.0.1

tools/screencast/runner.py

@@ -0,0 +1,100 @@
+"""主编排器：录屏 + 运行 pytest + 生成字幕 + FFmpeg 合成。"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from .composer import compose_video
+from .recorder import ScreenRecorder
+from .subtitle import SubtitleEngine
+
+
+def run_screencast(
+    skill_slug: str,
+    subtitle_script: List[Tuple[str, str]],
+    pytest_args: List[str],
+    output_dir: str,
+    media_assets_root: Optional[str] = None,
+    music_subdir: str = "music",
+    fps: int = 10,
+) -> str:
+    """
+    录制一次技能桌面 E2E 演示视频。
+
+    Args:
+        skill_slug:       技能唯一标识，如 "query-balance-icbc"
+        subtitle_script:  [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout
+        pytest_args:      传递给 pytest 的参数列表
+        output_dir:       最终 MP4 输出目录
+        media_assets_root: media-assets 仓库本地路径；
+                           不传时读 MEDIA_ASSETS_ROOT 环境变量，
+                           再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
+        music_subdir:     music_assets_root 下音乐子目录，默认 "music"
+        fps:              截帧帧率，默认 10
+
+    Returns:
+        输出 MP4 的绝对路径
+    """
+    if media_assets_root is None:
+        media_assets_root = os.environ.get(
+            "MEDIA_ASSETS_ROOT",
+            r"D:\OpenClaw\client-commons\media-assets",
+        )
+
+    music_dir = str(Path(media_assets_root) / music_subdir)
+    output_dir_path = Path(output_dir)
+    output_dir_path.mkdir(parents=True, exist_ok=True)
+
+    date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
+    base_name = f"{skill_slug}_{date_str}"
+    output_mp4 = output_dir_path / f"{base_name}.mp4"
+    subtitle_file = output_dir_path / f"{base_name}.srt"
+
+    engine = SubtitleEngine(subtitle_script)
+
+    with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir:
+        recorder = ScreenRecorder(frames_dir, fps=fps)
+
+        print(f"[screencast] 启动录制 → {output_mp4.name}")
+        recorder.start()
+        engine.set_start_time(time.time())
+
+        try:
+            proc = subprocess.Popen(
+                [sys.executable, "-m", "pytest", *pytest_args],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+            )
+            assert proc.stdout is not None
+            for line in proc.stdout:
+                sys.stdout.write(line)
+                sys.stdout.flush()
+                engine.process_line(line)
+            exit_code = proc.wait()
+            if exit_code != 0:
+                print(f"[screencast] pytest 退出码 {exit_code}，录制仍继续合成")
+        finally:
+            recorder.stop()
+            print(f"[screencast] 录制停止，共 {recorder.frame_count} 帧")
+
+        engine.generate_srt(str(subtitle_file))
+        print(f"[screencast] 字幕 → {subtitle_file.name}")
+
+        compose_video(
+            frames_dir=frames_dir,
+            fps=fps,
+            subtitle_path=str(subtitle_file),
+            music_dir=music_dir,
+            output_path=str(output_mp4),
+        )
+
+    return str(output_mp4)

tools/screencast/subtitle.py

@@ -0,0 +1,55 @@
+"""字幕引擎：根据 stdout 关键词打时间戳，生成 SRT 字幕文件。"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Tuple
+
+
+@dataclass
+class _Entry:
+    start: float   # 距录制开始的秒数
+    text: str
+    duration: float = 5.0  # 每条字幕默认显示 5 秒
+
+
+class SubtitleEngine:
+    def __init__(self, script: List[Tuple[str, str]], default_duration: float = 5.0):
+        """
+        script: [(关键词, 字幕文案), ...]
+        关键词大小写不敏感，每个关键词只触发一次。
+        """
+        self._script = script
+        self._default_duration = default_duration
+        self._entries: List[_Entry] = []
+        self._matched: set[str] = set()
+        self._t0: float = 0.0
+
+    def set_start_time(self, t: float) -> None:
+        self._t0 = t
+
+    def process_line(self, line: str) -> None:
+        lower = line.lower()
+        for keyword, text in self._script:
+            if keyword in self._matched:
+                continue
+            if keyword.lower() in lower:
+                elapsed = time.time() - self._t0
+                self._entries.append(_Entry(start=elapsed, text=text, duration=self._default_duration))
+                self._matched.add(keyword)
+
+    def generate_srt(self, output_path: str) -> None:
+        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(self._entries, 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")