修改打包方式
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 1m2s

This commit is contained in:
2026-05-28 16:49:30 +08:00
parent 43ec2d66a3
commit ea25fc2638
17 changed files with 678 additions and 376 deletions

View File

@@ -2,8 +2,10 @@ name: Publish Python Package to Gitea
on: on:
push: push:
branches:
- main
tags: tags:
- 'v*' - "v*"
workflow_dispatch: workflow_dispatch:
jobs: jobs:
@@ -23,6 +25,13 @@ jobs:
steps: steps:
- uses: https://git.jc2009.com/admin/actions-checkout@v4 - uses: https://git.jc2009.com/admin/actions-checkout@v4
- name: Resolve and patch package version
env:
GITHUB_REF: ${{ github.ref }}
GITHUB_RUN_NUMBER: ${{ github.run_number }}
GITHUB_RUN_ID: ${{ github.run_id }}
run: python3.12 tools/ci_set_package_version.py
- name: Install build tools - name: Install build tools
run: python3.12 -m pip install --upgrade build twine --break-system-packages run: python3.12 -m pip install --upgrade build twine --break-system-packages

View File

@@ -56,13 +56,40 @@ finally:
## 5. 发版流程(维护者) ## 5. 发版流程(维护者)
### main 分支自动预发布
每次推送到 `main` 会触发 `.github/workflows/publish.yml`,自动构建并发布 **唯一** 的 post 版本(基于 `pyproject.toml` 中的基础版本,例如 `0.1.0.post42`)。版本号不会写回仓库,仅在 CI 构建前临时改写 `pyproject.toml``jiangchang_desktop_sdk.__version__`
```bash ```bash
git tag v0.1.0 git push origin main
git push origin v0.1.0
# CI 自动打包并发布到 Gitea Package Registry见 .github/workflows/publish.yml
``` ```
仓库内还保留 `examples/``tools/``python-runtime/` 目录及既有 Skill / 前端相关的 GitHub Actions 工作流,与本 PyPI 包的发布彼此独立。 测试人员升级安装:
```bash
python -m pip install --upgrade \
--index-url https://git.jc2009.com/api/packages/client-jiangchang/pypi/simple/ \
--extra-index-url https://pypi.org/simple \
jiangchang-platform-kit
```
### 正式版本tag
打 tag 并推送后,发布 **去掉 `v` 前缀** 的正式版本号(例如 `v0.1.1` → PyPI 包 `0.1.1`
```bash
git tag v0.1.1
git push origin v0.1.1
```
### 安装后验证
```bash
python -c "from jiangchang_desktop_sdk.e2e_helpers import send_prompt_via_composer; print('e2e_helpers OK')"
python -c "from screencast import run_screencast; print('screencast OK')"
```
仓库内还保留 `examples/``tools/``python-runtime/` 目录及既有 Skill / 前端相关的 GitHub Actions 工作流,与本 PyPI 包的发布彼此独立。`tools/screencast/` 为兼容薄壳pip 包内实现位于 `screencast` 包(`src/screencast/`)。
## v0.2.0 — data-jcid migration (internal refactor) ## v0.2.0 — data-jcid migration (internal refactor)

View File

@@ -13,6 +13,7 @@ authors = [{ name = "client-jiangchang" }]
dependencies = [ dependencies = [
"requests>=2.31.0", "requests>=2.31.0",
"playwright>=1.42.0", "playwright>=1.42.0",
"mss>=9.0.1",
] ]
[project.urls] [project.urls]

68
sanity_check.py Normal file
View File

@@ -0,0 +1,68 @@
"""
Stage 2 集成验证脚本——一次性 sanity check。
跑前准备:
1. 启动匠厂客户端(确认能正常打开聊天页 + 至少配好一个可用模型)
2. 在本仓库激活 Python venv,确认 playwright 已装
3. 直接 python sanity_check.py 即可
"""
import sys
import time
import traceback
print("=" * 60)
print("Stage 2 SDK Sanity Check")
print("=" * 60)
try:
from jiangchang_desktop_sdk import JiangchangDesktopClient
print("[1/6] import OK")
except Exception as e:
print(f"[FAIL] import failed: {e}")
sys.exit(1)
try:
with JiangchangDesktopClient() as c:
print("[2/6] context entered")
c.ensure_app_running()
print("[3/6] ensure_app_running OK")
c.wait_gateway_ready()
print("[4/6] wait_gateway_ready OK (note: returns even if gw not actually ready)")
c.new_task()
print("[5/6] new_task OK")
t0 = time.time()
answer = c.ask("你好,请用一句话回复,确认 SDK 可用")
elapsed = time.time() - t0
print(f"[6/6] ask() returned in {elapsed:.1f}s")
print(f" answer length: {len(answer)} chars")
print(f" answer preview: {answer[:200]!r}")
assert answer.strip(), "Empty answer!"
print()
print("[BONUS] verifying read() works")
msgs = c.read()
print(f" read() returned {len(msgs)} messages")
for i, m in enumerate(msgs[-4:]):
print(f" [-{len(msgs)-i}] role={m.role!r} id={m.id!r} "
f"content_len={len(m.content) if m.content else 0}")
print()
print("[BONUS] verifying send_file raises NotImplementedError")
try:
c.send_file("/tmp/fake.txt")
print(" [FAIL] send_file did NOT raise!")
except NotImplementedError as e:
print(f" [OK] send_file raised: {str(e)[:80]}...")
print()
print("=" * 60)
print("ALL CHECKS PASSED ✓")
print("=" * 60)
except Exception as e:
print(f"\n[FAIL] sanity check crashed:")
traceback.print_exc()
sys.exit(2)

View File

@@ -134,3 +134,74 @@ def apply_cli_local_defaults() -> None:
os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
if not (os.getenv("JIANGCHANG_USER_ID") or "").strip(): if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip() os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()
def find_chrome_executable() -> "str | None":
"""查找系统已安装的 Chrome 或 Edge 可执行文件路径。
查找顺序:
1. 常见文件路径PROGRAMFILES、LOCALAPPDATA、macOS、Linux
2. Windows 注册表 App PathsHKLM 优先HKCU 其次)
3. PATH shutil.which 兜底
返回找到的路径字符串,找不到返回 None。
宿主/技能调用 Playwright launch 时应传 executable_path=find_chrome_executable()
而非 channel="chrome",以避免 Playwright 内部检测在系统级安装时失效。
"""
import shutil
if sys.platform == "win32":
local_app_data = os.environ.get("LOCALAPPDATA", "")
prog_files = os.environ.get("PROGRAMFILES", r"C:\Program Files")
prog_files_x86 = os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)")
candidates = [
os.path.join(prog_files, r"Google\Chrome\Application\chrome.exe"),
os.path.join(prog_files_x86, r"Google\Chrome\Application\chrome.exe"),
os.path.join(local_app_data, r"Google\Chrome\Application\chrome.exe"),
os.path.join(prog_files, r"Microsoft\Edge\Application\msedge.exe"),
os.path.join(prog_files_x86, r"Microsoft\Edge\Application\msedge.exe"),
os.path.join(local_app_data, r"Microsoft\Edge\Application\msedge.exe"),
]
elif sys.platform == "darwin":
candidates = [
"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
"/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
"/Applications/Chromium.app/Contents/MacOS/Chromium",
]
else:
candidates = [
"/usr/bin/google-chrome-stable",
"/usr/bin/google-chrome",
"/usr/bin/chromium-browser",
"/usr/bin/chromium",
"/usr/bin/microsoft-edge",
]
for path in candidates:
if path and os.path.isfile(path):
return path
# Windows 注册表兜底(覆盖系统级 HKLM 安装)
if sys.platform == "win32":
try:
import winreg
for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
for reg_path in (
r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\chrome.exe",
r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\msedge.exe",
):
try:
with winreg.OpenKey(hive, reg_path) as key:
val, _ = winreg.QueryValueEx(key, "")
if val and os.path.isfile(val):
return val
except OSError:
continue
except ImportError:
pass
# PATH 兜底
for name in ("google-chrome-stable", "google-chrome", "chrome", "msedge", "chromium"):
found = shutil.which(name)
if found:
return found
return None

View File

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

View File

@@ -0,0 +1,80 @@
"""FFmpeg 合成:帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
from __future__ import annotations
import random
import subprocess
import sys
from pathlib import Path
# 匠厂统一字幕样式 —— 所有 skill 录屏共用,做品牌一致性
# FontSize 141080p 屏幕约占 1.3% 高度,演示视频留足画面呼吸感
# Outline 214 号字配 Outline 3 会显胖2 已足够在浅色背景上保持可读
# 修改请评估对所有 skill 录屏的影响。
_JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14,"
"PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
"Outline=2,Shadow=1,Alignment=2"
)
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}'"
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
)
# 视频和音频都放进 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)

View File

@@ -0,0 +1,63 @@
"""全屏截帧引擎,后台线程持续截图存为 PNG 帧序列。"""
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,
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
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:
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)
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)

142
src/screencast/runner.py Normal file
View File

@@ -0,0 +1,142 @@
"""主编排器:录屏 + 运行 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)

View File

@@ -0,0 +1,94 @@
"""字幕引擎:根据 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],
default_duration: float = 5.0,
):
"""
script: [(关键词, 字幕文案) | (关键词, 字幕文案, 显示时长秒), ...]
- 二元组:用 default_duration向后兼容
- 三元组:本条字幕独立指定显示秒数(用于长等待场景延长字幕停留)
关键词大小写不敏感,每个关键词只触发一次。
"""
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 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=float(duration)))
self._matched.add(keyword)
def generate_srt(self, output_path: str) -> None:
"""生成 SRT 文件,自动处理字幕重叠。
字幕重叠修复策略(避免屏幕上同时显示 2+ 条叠成两行):
- 字幕按 start 时间升序排序后;
- 相邻字幕保留 GAP=0.1s 间隔;
- 每条字幕保证至少 MIN_DURATION=1.0s 显示时间(短到这个值仍读不完是字幕脚本设计问题);
- 若当前条让位给下一条后剩余时间 < MIN_DURATION则把下一条 start 往后推。
这样任意时刻屏幕上最多 1 条字幕。
"""
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}"
# 防重叠 post-process
MIN_DURATION = 1.0
GAP = 0.1
entries = sorted(self._entries, key=lambda e: e.start)
for i in range(len(entries) - 1):
cur, nxt = entries[i], entries[i + 1]
max_end = nxt.start - GAP
new_dur = max_end - cur.start
if new_dur < MIN_DURATION:
# 太挤cur 至少撑 MIN_DURATION 秒,把 nxt 推迟
cur.duration = MIN_DURATION
nxt.start = cur.start + MIN_DURATION + GAP
else:
# 正常让位cur 显示到 (nxt.start - GAP)
cur.duration = min(cur.duration, new_dur)
# 最后一条不需要让位,保持原 duration
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(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")

View File

@@ -0,0 +1,79 @@
"""Patch pyproject.toml and __init__.py version for CI builds (not committed)."""
from __future__ import annotations
import os
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
def _read_base_version() -> str:
text = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
match = re.search(r'^version\s*=\s*"([^"]+)"', text, re.MULTILINE)
if not match:
raise SystemExit("ERROR: version not found in pyproject.toml")
base = match.group(1)
return re.sub(r"\.post\d+$", "", base)
def resolve_version() -> str:
ref = os.environ.get("GITHUB_REF", "")
if ref.startswith("refs/tags/v"):
return ref.removeprefix("refs/tags/v")
if ref.startswith("refs/tags/"):
return ref.removeprefix("refs/tags/")
run_number = (
os.environ.get("GITHUB_RUN_NUMBER")
or os.environ.get("GITHUB_RUN_ID")
or "0"
)
base = _read_base_version()
return f"{base}.post{run_number}"
def patch_version(version: str) -> None:
pyproject = ROOT / "pyproject.toml"
py_text = pyproject.read_text(encoding="utf-8")
py_text, n = re.subn(
r'^version\s*=\s*"[^"]*"',
f'version = "{version}"',
py_text,
count=1,
flags=re.MULTILINE,
)
if n != 1:
raise SystemExit("ERROR: failed to patch version in pyproject.toml")
pyproject.write_text(py_text, encoding="utf-8")
init_py = ROOT / "src" / "jiangchang_desktop_sdk" / "__init__.py"
init_text = init_py.read_text(encoding="utf-8")
init_text, n = re.subn(
r'^__version__\s*=\s*"[^"]*"',
f'__version__ = "{version}"',
init_text,
count=1,
flags=re.MULTILINE,
)
if n != 1:
raise SystemExit("ERROR: failed to patch __version__ in __init__.py")
init_py.write_text(init_text, encoding="utf-8")
def main() -> None:
version = resolve_version()
print(f"Resolved package version: {version}")
patch_version(version)
if os.environ.get("GITHUB_OUTPUT"):
with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as fh:
fh.write(f"package_version={version}\n")
if __name__ == "__main__":
try:
main()
except SystemExit as exc:
print(exc, file=sys.stderr)
raise

View File

@@ -1,3 +1,6 @@
from .runner import run_screencast from ._bootstrap import bootstrap_src
bootstrap_src()
from screencast import run_screencast
__all__ = ["run_screencast"] __all__ = ["run_screencast"]

View File

@@ -0,0 +1,15 @@
"""源码仓库内运行 tools/screencast 时,确保 src 在 sys.pathpip 安装后无需)。"""
from __future__ import annotations
import sys
from pathlib import Path
def bootstrap_src() -> None:
try:
import screencast # noqa: F401
except ImportError:
src = Path(__file__).resolve().parent.parent.parent / "src"
src_str = str(src)
if src.is_dir() and src_str not in sys.path:
sys.path.insert(0, src_str)

View File

@@ -1,80 +1,6 @@
"""FFmpeg 合成:帧序列 + SRT 字幕 + 背景音乐 → MP4。""" from ._bootstrap import bootstrap_src
from __future__ import annotations
import random bootstrap_src()
import subprocess from screencast.composer import compose_video
import sys
from pathlib import Path
__all__ = ["compose_video"]
# 匠厂统一字幕样式 —— 所有 skill 录屏共用,做品牌一致性
# FontSize 141080p 屏幕约占 1.3% 高度,演示视频留足画面呼吸感
# Outline 214 号字配 Outline 3 会显胖2 已足够在浅色背景上保持可读
# 修改请评估对所有 skill 录屏的影响。
_JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14,"
"PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
"Outline=2,Shadow=1,Alignment=2"
)
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}'"
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
)
# 视频和音频都放进 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)

View File

@@ -1,63 +1,6 @@
"""全屏截帧引擎,后台线程持续截图存为 PNG 帧序列。""" from ._bootstrap import bootstrap_src
from __future__ import annotations
import threading bootstrap_src()
import time from screencast.recorder import ScreenRecorder
from pathlib import Path
from typing import Optional
import mss __all__ = ["ScreenRecorder"]
import mss.tools
class ScreenRecorder:
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
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:
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)
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)

View File

@@ -1,140 +1,6 @@
"""主编排器:录屏 + 运行 pytest + 生成字幕 + FFmpeg 合成。""" from ._bootstrap import bootstrap_src
from __future__ import annotations
import os bootstrap_src()
import subprocess from screencast.runner import run_screencast
import sys
import tempfile
import time
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Tuple
__all__ = ["run_screencast"]
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 .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)

View File

@@ -1,94 +1,6 @@
"""字幕引擎:根据 stdout 关键词打时间戳,生成 SRT 字幕文件。""" from ._bootstrap import bootstrap_src
from __future__ import annotations
import time bootstrap_src()
from dataclasses import dataclass from screencast.subtitle import SubtitleEngine
from pathlib import Path
from typing import List, Tuple
__all__ = ["SubtitleEngine"]
@dataclass
class _Entry:
start: float # 距录制开始的秒数
text: str
duration: float = 5.0 # 每条字幕默认显示 5 秒
class SubtitleEngine:
def __init__(
self,
script: List[Tuple],
default_duration: float = 5.0,
):
"""
script: [(关键词, 字幕文案) | (关键词, 字幕文案, 显示时长秒), ...]
- 二元组:用 default_duration向后兼容
- 三元组:本条字幕独立指定显示秒数(用于长等待场景延长字幕停留)
关键词大小写不敏感,每个关键词只触发一次。
"""
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 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=float(duration)))
self._matched.add(keyword)
def generate_srt(self, output_path: str) -> None:
"""生成 SRT 文件,自动处理字幕重叠。
字幕重叠修复策略(避免屏幕上同时显示 2+ 条叠成两行):
- 字幕按 start 时间升序排序后;
- 相邻字幕保留 GAP=0.1s 间隔;
- 每条字幕保证至少 MIN_DURATION=1.0s 显示时间(短到这个值仍读不完是字幕脚本设计问题);
- 若当前条让位给下一条后剩余时间 < MIN_DURATION则把下一条 start 往后推。
这样任意时刻屏幕上最多 1 条字幕。
"""
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}"
# 防重叠 post-process
MIN_DURATION = 1.0
GAP = 0.1
entries = sorted(self._entries, key=lambda e: e.start)
for i in range(len(entries) - 1):
cur, nxt = entries[i], entries[i + 1]
max_end = nxt.start - GAP
new_dur = max_end - cur.start
if new_dur < MIN_DURATION:
# 太挤cur 至少撑 MIN_DURATION 秒,把 nxt 推迟
cur.duration = MIN_DURATION
nxt.start = cur.start + MIN_DURATION + GAP
else:
# 正常让位cur 显示到 (nxt.start - GAP)
cur.duration = min(cur.duration, new_dur)
# 最后一条不需要让位,保持原 duration
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(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")