feat(screencast): 自动激活窗口 / 多显示器 / 字幕独立时长

基于 SDK 新模块(前两个 commit)的 screencast 工具能力升级:

- runner.py:
  · 新增 activate_window: bool = True 参数,录屏前自动调
    activate_and_maximize_jiangchang_window,避免开头几秒录到 PowerShell/桌面
  · 新增 monitor_index: Optional[int] = None 参数,透传给 ScreenRecorder
  · 自带 SDK 路径兜底 _ensure_sdk_on_path,让 record_screencast.py 直接 python
    跑时(不走 pytest conftest)也能 import jiangchang_desktop_sdk.window_win32

- recorder.py:
  · ScreenRecorder.__init__ 加 monitor_index 参数
  · _capture_loop 用 idx 选 monitor,越界兜底到 0
  · None = 旧行为(monitors[0] 所有屏合并),1 = 主屏,2+ = 第 N 屏
  · 多显示器场景推荐显式传 monitor_index=1 避免录出半边空白

- subtitle.py:
  · SubtitleEngine.process_line 支持二元组(兼容旧脚本)和三元组(每条独立 duration)
  · 长等待场景可以让某条字幕停留更久(如 30s)覆盖中间无日志的"字幕真空"

所有改动向后兼容,全部新增参数有默认值。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-05-25 11:14:11 +08:00
parent 311441dab0
commit 43ec2d66a3
3 changed files with 68 additions and 8 deletions

View File

@@ -4,15 +4,27 @@ from __future__ import annotations
import threading import threading
import time import time
from pathlib import Path from pathlib import Path
from typing import Optional
import mss import mss
import mss.tools import mss.tools
class ScreenRecorder: 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._frames_dir = Path(frames_dir)
self._fps = fps self._fps = fps
self._monitor_index = monitor_index
self._stop_event = threading.Event() self._stop_event = threading.Event()
self._thread: threading.Thread | None = None self._thread: threading.Thread | None = None
self._frame_count = 0 self._frame_count = 0
@@ -36,7 +48,10 @@ class ScreenRecorder:
def _capture_loop(self) -> None: def _capture_loop(self) -> None:
interval = 1.0 / self._fps interval = 1.0 / self._fps
with mss.mss() as sct: 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(): while not self._stop_event.is_set():
t0 = time.perf_counter() t0 = time.perf_counter()
img = sct.grab(monitor) img = sct.grab(monitor)

View File

@@ -10,6 +10,24 @@ from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import List, Optional, Tuple from typing import List, Optional, Tuple
def _ensure_sdk_on_path() -> None:
"""record_screencast.py 直接 python 跑时不走 conftestSDK 可能不在 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 .composer import compose_video
from .recorder import ScreenRecorder from .recorder import ScreenRecorder
from .subtitle import SubtitleEngine from .subtitle import SubtitleEngine
@@ -23,6 +41,8 @@ def run_screencast(
media_assets_root: Optional[str] = None, media_assets_root: Optional[str] = None,
music_subdir: str = "music", music_subdir: str = "music",
fps: int = 10, fps: int = 10,
activate_window: bool = True,
monitor_index: Optional[int] = None,
) -> str: ) -> str:
""" """
录制一次技能桌面 E2E 演示视频。 录制一次技能桌面 E2E 演示视频。
@@ -36,7 +56,11 @@ def run_screencast(
不传时读 MEDIA_ASSETS_ROOT 环境变量, 不传时读 MEDIA_ASSETS_ROOT 环境变量,
再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets 再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
music_subdir: music_assets_root 下音乐子目录,默认 "music" 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: Returns:
输出 MP4 的绝对路径 输出 MP4 的绝对路径
@@ -59,7 +83,15 @@ def run_screencast(
engine = SubtitleEngine(subtitle_script) engine = SubtitleEngine(subtitle_script)
with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir: 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}") print(f"[screencast] 启动录制 → {output_mp4.name}")
recorder.start() recorder.start()

View File

@@ -15,9 +15,15 @@ class _Entry:
class SubtitleEngine: 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 self._script = script
@@ -31,12 +37,19 @@ class SubtitleEngine:
def process_line(self, line: str) -> None: def process_line(self, line: str) -> None:
lower = line.lower() 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: if keyword in self._matched:
continue continue
if keyword.lower() in lower: if keyword.lower() in lower:
elapsed = time.time() - self._t0 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) self._matched.add(keyword)
def generate_srt(self, output_path: str) -> None: def generate_srt(self, output_path: str) -> None: