chendelian
ea25fc2638
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 1m2s
143 lines
5.2 KiB
Python
143 lines
5.2 KiB
Python
"""主编排器：录屏 + 运行 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
|
||
|
||
|
||
def _ensure_sdk_on_path() -> None:
|
||
"""pip 安装后应能直接 import；仅源码开发态才把 src 加入 sys.path。"""
|
||
import importlib.util
|
||
|
||
if importlib.util.find_spec("jiangchang_desktop_sdk") is not None:
|
||
return
|
||
# src/screencast/runner.py → src/screencast → src
|
||
here = Path(__file__).resolve()
|
||
src = here.parent.parent
|
||
sdk_dir = src / "jiangchang_desktop_sdk"
|
||
if sdk_dir.is_dir():
|
||
src_str = str(src)
|
||
if src_str not in sys.path:
|
||
sys.path.insert(0, src_str)
|
||
|
||
|
||
_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
|
||
|
||
|
||
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,
|
||
activate_window: bool = True,
|
||
monitor_index: Optional[int] = None,
|
||
) -> 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
|
||
activate_window: 录屏开始前是否自动激活+最大化匠厂窗口（默认 True；
|
||
Windows 走 ctypes，非 Windows 走 jiangchang:// 协议）
|
||
monitor_index: 指定录哪个显示器；None=所有显示器合并（旧行为），
|
||
1=主屏，2+=第 N 屏（推荐多显示器场景显式传 1）
|
||
|
||
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, 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()
|
||
engine.set_start_time(time.time())
|
||
|
||
try:
|
||
# Windows 子进程 stdout 默认走系统 ANSI code page（中文 Windows 是 cp936/GBK），
|
||
# 这边用 encoding="utf-8" 解码会乱码。注入 PYTHONIOENCODING + PYTHONUTF8 让
|
||
# 子进程 Python 强制以 UTF-8 输出，两端编码对齐。非 Windows 默认就 UTF-8，无副作用。
|
||
env = os.environ.copy()
|
||
env["PYTHONIOENCODING"] = "utf-8"
|
||
env["PYTHONUTF8"] = "1"
|
||
|
||
proc = subprocess.Popen(
|
||
[sys.executable, "-m", "pytest", *pytest_args],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
env=env,
|
||
)
|
||
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)
|