From ea25fc2638d5b72132320835b5ed225291f110de Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Thu, 28 May 2026 16:49:30 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E6=89=93=E5=8C=85=E6=96=B9?= =?UTF-8?q?=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/publish.yml | 13 ++- README.md | 35 +++++- pyproject.toml | 1 + sanity_check.py | 68 +++++++++++ src/jiangchang_skill_core/runtime_env.py | 71 ++++++++++++ src/screencast/__init__.py | 3 + src/screencast/composer.py | 80 +++++++++++++ src/screencast/recorder.py | 63 ++++++++++ src/screencast/runner.py | 142 +++++++++++++++++++++++ src/screencast/subtitle.py | 94 +++++++++++++++ tools/ci_set_package_version.py | 79 +++++++++++++ tools/screencast/__init__.py | 5 +- tools/screencast/_bootstrap.py | 15 +++ tools/screencast/composer.py | 82 +------------ tools/screencast/recorder.py | 65 +---------- tools/screencast/runner.py | 142 +---------------------- tools/screencast/subtitle.py | 96 +-------------- 17 files changed, 678 insertions(+), 376 deletions(-) create mode 100644 sanity_check.py create mode 100644 src/screencast/__init__.py create mode 100644 src/screencast/composer.py create mode 100644 src/screencast/recorder.py create mode 100644 src/screencast/runner.py create mode 100644 src/screencast/subtitle.py create mode 100644 tools/ci_set_package_version.py create mode 100644 tools/screencast/_bootstrap.py diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index f02a56c..c6afe4c 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -2,8 +2,10 @@ name: Publish Python Package to Gitea on: push: + branches: + - main tags: - - 'v*' + - "v*" workflow_dispatch: jobs: @@ -23,6 +25,13 @@ jobs: steps: - 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 run: python3.12 -m pip install --upgrade build twine --break-system-packages @@ -36,4 +45,4 @@ jobs: run: | python3.12 -m twine upload \ --repository-url https://git.jc2009.com/api/packages/client-jiangchang/pypi \ - dist/* \ No newline at end of file + dist/* diff --git a/README.md b/README.md index 3d312a2..a949bea 100644 --- a/README.md +++ b/README.md @@ -56,13 +56,40 @@ finally: ## 5. 发版流程（维护者） +### main 分支自动预发布 + +每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **唯一** 的 post 版本（基于 `pyproject.toml` 中的基础版本，例如 `0.1.0.post42`）。版本号不会写回仓库，仅在 CI 构建前临时改写 `pyproject.toml` 与 `jiangchang_desktop_sdk.__version__`。 + ```bash -git tag v0.1.0 -git push origin v0.1.0 -# CI 自动打包并发布到 Gitea Package Registry（见 .github/workflows/publish.yml） +git push origin main ``` -仓库内还保留 `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) diff --git a/pyproject.toml b/pyproject.toml index cd3f2af..59ac1ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ authors = [{ name = "client-jiangchang" }] dependencies = [ "requests>=2.31.0", "playwright>=1.42.0", + "mss>=9.0.1", ] [project.urls] diff --git a/sanity_check.py b/sanity_check.py new file mode 100644 index 0000000..dabb811 --- /dev/null +++ b/sanity_check.py @@ -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) \ No newline at end of file diff --git a/src/jiangchang_skill_core/runtime_env.py b/src/jiangchang_skill_core/runtime_env.py index 3ac554b..2092467 100644 --- a/src/jiangchang_skill_core/runtime_env.py +++ b/src/jiangchang_skill_core/runtime_env.py @@ -134,3 +134,74 @@ def apply_cli_local_defaults() -> None: os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root() if not (os.getenv("JIANGCHANG_USER_ID") or "").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 Paths（HKLM 优先，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 diff --git a/src/screencast/__init__.py b/src/screencast/__init__.py new file mode 100644 index 0000000..92a0979 --- /dev/null +++ b/src/screencast/__init__.py @@ -0,0 +1,3 @@ +from .runner import run_screencast + +__all__ = ["run_screencast"] diff --git a/src/screencast/composer.py b/src/screencast/composer.py new file mode 100644 index 0000000..4ea95b9 --- /dev/null +++ b/src/screencast/composer.py @@ -0,0 +1,80 @@ +"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。""" +from __future__ import annotations + +import random +import subprocess +import sys +from pathlib import Path + + +# 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性 +# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感 +# Outline 2：14 号字配 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) diff --git a/src/screencast/recorder.py b/src/screencast/recorder.py new file mode 100644 index 0000000..331049c --- /dev/null +++ b/src/screencast/recorder.py @@ -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) diff --git a/src/screencast/runner.py b/src/screencast/runner.py new file mode 100644 index 0000000..856a90e --- /dev/null +++ b/src/screencast/runner.py @@ -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) diff --git a/src/screencast/subtitle.py b/src/screencast/subtitle.py new file mode 100644 index 0000000..e52c5ff --- /dev/null +++ b/src/screencast/subtitle.py @@ -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") diff --git a/tools/ci_set_package_version.py b/tools/ci_set_package_version.py new file mode 100644 index 0000000..5227547 --- /dev/null +++ b/tools/ci_set_package_version.py @@ -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 diff --git a/tools/screencast/__init__.py b/tools/screencast/__init__.py index 92a0979..e58035d 100644 --- a/tools/screencast/__init__.py +++ b/tools/screencast/__init__.py @@ -1,3 +1,6 @@ -from .runner import run_screencast +from ._bootstrap import bootstrap_src + +bootstrap_src() +from screencast import run_screencast __all__ = ["run_screencast"] diff --git a/tools/screencast/_bootstrap.py b/tools/screencast/_bootstrap.py new file mode 100644 index 0000000..1f1dd98 --- /dev/null +++ b/tools/screencast/_bootstrap.py @@ -0,0 +1,15 @@ +"""源码仓库内运行 tools/screencast 时，确保 src 在 sys.path（pip 安装后无需）。""" +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) diff --git a/tools/screencast/composer.py b/tools/screencast/composer.py index 4ea95b9..f680362 100644 --- a/tools/screencast/composer.py +++ b/tools/screencast/composer.py @@ -1,80 +1,6 @@ -"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。""" -from __future__ import annotations +from ._bootstrap import bootstrap_src -import random -import subprocess -import sys -from pathlib import Path +bootstrap_src() +from screencast.composer import compose_video - -# 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性 -# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感 -# Outline 2：14 号字配 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) +__all__ = ["compose_video"] diff --git a/tools/screencast/recorder.py b/tools/screencast/recorder.py index 331049c..9ae55d4 100644 --- a/tools/screencast/recorder.py +++ b/tools/screencast/recorder.py @@ -1,63 +1,6 @@ -"""全屏截帧引擎，后台线程持续截图存为 PNG 帧序列。""" -from __future__ import annotations +from ._bootstrap import bootstrap_src -import threading -import time -from pathlib import Path -from typing import Optional +bootstrap_src() +from screencast.recorder import ScreenRecorder -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) +__all__ = ["ScreenRecorder"] diff --git a/tools/screencast/runner.py b/tools/screencast/runner.py index e9aa61e..fa55dcc 100644 --- a/tools/screencast/runner.py +++ b/tools/screencast/runner.py @@ -1,140 +1,6 @@ -"""主编排器：录屏 + 运行 pytest + 生成字幕 + FFmpeg 合成。""" -from __future__ import annotations +from ._bootstrap import bootstrap_src -import os -import subprocess -import sys -import tempfile -import time -from datetime import datetime -from pathlib import Path -from typing import List, Optional, Tuple +bootstrap_src() +from screencast.runner import run_screencast - -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 - - -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) +__all__ = ["run_screencast"] diff --git a/tools/screencast/subtitle.py b/tools/screencast/subtitle.py index e52c5ff..9bd8bf0 100644 --- a/tools/screencast/subtitle.py +++ b/tools/screencast/subtitle.py @@ -1,94 +1,6 @@ -"""字幕引擎：根据 stdout 关键词打时间戳，生成 SRT 字幕文件。""" -from __future__ import annotations +from ._bootstrap import bootstrap_src -import time -from dataclasses import dataclass -from pathlib import Path -from typing import List, Tuple +bootstrap_src() +from screencast.subtitle import 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") +__all__ = ["SubtitleEngine"]