diff --git a/tools/screencast/recorder.py b/tools/screencast/recorder.py index cfd213c..331049c 100644 --- a/tools/screencast/recorder.py +++ b/tools/screencast/recorder.py @@ -4,15 +4,27 @@ from __future__ import annotations import threading import time from pathlib import Path +from typing import Optional import mss import mss.tools class ScreenRecorder: - def __init__(self, frames_dir: str, fps: int = 10): + def __init__( + self, + frames_dir: str, + fps: int = 10, + monitor_index: Optional[int] = None, + ): + """ + Args: + monitor_index: mss 显示器索引;None=monitors[0](所有屏合并区域, + 旧行为),1=主屏,2+=第 N 屏。多显示器场景推荐显式传 1。 + """ self._frames_dir = Path(frames_dir) self._fps = fps + self._monitor_index = monitor_index self._stop_event = threading.Event() self._thread: threading.Thread | None = None self._frame_count = 0 @@ -36,7 +48,10 @@ class ScreenRecorder: def _capture_loop(self) -> None: interval = 1.0 / self._fps with mss.mss() as sct: - monitor = sct.monitors[0] # 全屏(monitor[0] 是所有显示器合并区域) + idx = self._monitor_index if self._monitor_index is not None else 0 + if idx < 0 or idx >= len(sct.monitors): + idx = 0 # 越界兜底 + monitor = sct.monitors[idx] while not self._stop_event.is_set(): t0 = time.perf_counter() img = sct.grab(monitor) diff --git a/tools/screencast/runner.py b/tools/screencast/runner.py index 13ef29f..e9aa61e 100644 --- a/tools/screencast/runner.py +++ b/tools/screencast/runner.py @@ -10,6 +10,24 @@ from datetime import datetime from pathlib import Path from typing import List, Optional, Tuple + +def _ensure_sdk_on_path() -> None: + """record_screencast.py 直接 python 跑时不走 conftest,SDK 可能不在 sys.path。 + 自带兜底:从 tools/screencast/runner.py 上溯到 platform-kit/src。""" + import importlib.util + if importlib.util.find_spec("jiangchang_desktop_sdk") is not None: + return + # tools/screencast/runner.py → tools/screencast → tools → platform-kit + here = Path(__file__).resolve() + src = here.parent.parent.parent / "src" + if (src / "jiangchang_desktop_sdk").is_dir(): + import sys as _sys + _sys.path.insert(0, str(src)) + + +_ensure_sdk_on_path() +from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window # noqa: E402 + from .composer import compose_video from .recorder import ScreenRecorder from .subtitle import SubtitleEngine @@ -23,6 +41,8 @@ def run_screencast( media_assets_root: Optional[str] = None, music_subdir: str = "music", fps: int = 10, + activate_window: bool = True, + monitor_index: Optional[int] = None, ) -> str: """ 录制一次技能桌面 E2E 演示视频。 @@ -36,7 +56,11 @@ def run_screencast( 不传时读 MEDIA_ASSETS_ROOT 环境变量, 再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets music_subdir: music_assets_root 下音乐子目录,默认 "music" - fps: 截帧帧率,默认 10 + fps: 截帧帧率,默认 10 + activate_window: 录屏开始前是否自动激活+最大化匠厂窗口(默认 True; + Windows 走 ctypes,非 Windows 走 jiangchang:// 协议) + monitor_index: 指定录哪个显示器;None=所有显示器合并(旧行为), + 1=主屏,2+=第 N 屏(推荐多显示器场景显式传 1) Returns: 输出 MP4 的绝对路径 @@ -59,7 +83,15 @@ def run_screencast( engine = SubtitleEngine(subtitle_script) with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir: - recorder = ScreenRecorder(frames_dir, fps=fps) + recorder = ScreenRecorder(frames_dir, fps=fps, monitor_index=monitor_index) + + if activate_window: + ok = activate_and_maximize_jiangchang_window() + if ok: + print("[screencast] 已激活+最大化匠厂窗口") + time.sleep(2) # 等窗口浮起稳定 + else: + print("[screencast] 未能激活匠厂窗口,继续录制(可能录到桌面边角)") print(f"[screencast] 启动录制 → {output_mp4.name}") recorder.start() diff --git a/tools/screencast/subtitle.py b/tools/screencast/subtitle.py index a8b7ae8..e52c5ff 100644 --- a/tools/screencast/subtitle.py +++ b/tools/screencast/subtitle.py @@ -15,9 +15,15 @@ class _Entry: class SubtitleEngine: - def __init__(self, script: List[Tuple[str, str]], default_duration: float = 5.0): + def __init__( + self, + script: List[Tuple], + default_duration: float = 5.0, + ): """ - script: [(关键词, 字幕文案), ...] + script: [(关键词, 字幕文案) | (关键词, 字幕文案, 显示时长秒), ...] + - 二元组:用 default_duration(向后兼容) + - 三元组:本条字幕独立指定显示秒数(用于长等待场景延长字幕停留) 关键词大小写不敏感,每个关键词只触发一次。 """ self._script = script @@ -31,12 +37,19 @@ class SubtitleEngine: def process_line(self, line: str) -> None: lower = line.lower() - for keyword, text in self._script: + for entry_tuple in self._script: + if len(entry_tuple) == 2: + keyword, text = entry_tuple + duration = self._default_duration + elif len(entry_tuple) == 3: + keyword, text, duration = entry_tuple + else: + continue 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._entries.append(_Entry(start=elapsed, text=text, duration=float(duration))) self._matched.add(keyword) def generate_srt(self, output_path: str) -> None: