diff --git a/README.md b/README.md index e5cfdac..492e45c 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,27 @@ send_prompt_via_composer(page, "第一行\n第二行") 可通过环境变量覆盖： -`MEDIA_ASSETS_ROOT=D:\OpenClaw\client-commons\media-assets` +`MEDIA_ASSETS_ROOT=/path/to/custom/media-assets` + +### 录屏合成（`screencast`） + +`run_screencast()` / `compose_video()` 通过 `jiangchang_skill_core.media_assets` 解析 ffmpeg 与背景音乐，**不使用系统 PATH 中的 `ffmpeg`**。 + +- **默认**：不传 `media_assets_root` 时，使用 `{JIANGCHANG_DATA_ROOT}/shared/media-assets`（或进程环境中已有的 `MEDIA_ASSETS_ROOT`）。 +- **覆盖**：传入 `media_assets_root="/path/to/media-assets"`，或在 `compose_video(..., media_assets_env={"MEDIA_ASSETS_ROOT": "..."})` 中设置；参数 `media_assets_root` 优先于 `media_assets_env` 里已有的 `MEDIA_ASSETS_ROOT`。 +- `run_screencast(media_assets_root=...)` 会把该路径原样传给 `compose_video()`；`music_subdir` 已废弃（会触发 `DeprecationWarning`），音乐选择统一由 `pick_background_music()` 完成。 + +```python +from screencast import run_screencast + +run_screencast( + skill_slug="demo-skill", + subtitle_script=[("PASSED", "测试通过")], + pytest_args=["-q", "tests/test_demo.py"], + output_dir="./out", + media_assets_root="/path/to/media-assets", # 可选 +) +``` ## 5. 发版流程（维护者） diff --git a/pyproject.toml b/pyproject.toml index f721d69..0f86230 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.0.9" +version = "1.0.10" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index 30f19b7..38143e8 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -23,8 +23,11 @@ from .unified_logging import ( ) from .media_assets import ( MediaAssetsStatus, + background_music_issue, ensure_media_assets, pick_background_music, + probe_ffmpeg, + probe_media_assets, resolve_ffmpeg, resolve_ffprobe, resolve_media_assets_root, @@ -43,7 +46,10 @@ __all__ = [ "EntitlementResult", "EntitlementServiceError", "MediaAssetsStatus", + "background_music_issue", "ensure_media_assets", + "probe_ffmpeg", + "probe_media_assets", "pick_background_music", "resolve_ffmpeg", "resolve_ffprobe", diff --git a/src/jiangchang_skill_core/config.py b/src/jiangchang_skill_core/config.py index 1a0a45a..0f02986 100644 --- a/src/jiangchang_skill_core/config.py +++ b/src/jiangchang_skill_core/config.py @@ -69,6 +69,69 @@ def reset_cache() -> None: _example_defaults_cache = None +def get_env_file_path() -> str | None: + return _env_file_path + + +def get_example_path() -> str | None: + return _example_path + + +def _iter_env_assignments(path: str) -> list[tuple[str, str]]: + """按文件顺序返回 (key, 完整赋值行)。""" + result: list[tuple[str, str]] = [] + if not path or not os.path.isfile(path): + return result + with open(path, encoding="utf-8") as f: + for raw_line in f: + line = raw_line.rstrip("\n\r") + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + if "=" not in stripped: + continue + key, _, _ = stripped.partition("=") + key = key.strip() + if key: + result.append((key, stripped)) + return result + + +def merge_missing_env_keys( + example_path: str, + dest_path: str, + *, + comment_skill: str | None = None, +) -> list[str]: + """将 example 中有而用户 .env 没有的 key 追加到 dest 末尾，不修改已有项。""" + if not example_path or not os.path.isfile(example_path): + return [] + if not dest_path or not os.path.isfile(dest_path): + return [] + + dest_keys = set(_parse_env_file(dest_path).keys()) + to_append: list[tuple[str, str]] = [] + for key, assignment in _iter_env_assignments(example_path): + if key not in dest_keys: + to_append.append((key, assignment)) + + if not to_append: + return [] + + header = ( + f"# Added by {comment_skill} from .env.example" + if comment_skill + else "# Added from .env.example" + ) + with open(dest_path, "a", encoding="utf-8") as f: + f.write("\n\n" + header + "\n") + for _, assignment in to_append: + f.write(assignment + "\n") + + reset_cache() + return [key for key, _ in to_append] + + def ensure_env_file(skill_slug: str, example_path: str) -> str: """首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env（已存在不覆盖），返回路径。""" global _skill_slug, _example_path, _env_file_path diff --git a/src/jiangchang_skill_core/media_assets.py b/src/jiangchang_skill_core/media_assets.py index 7ddba0f..8b11e3c 100644 --- a/src/jiangchang_skill_core/media_assets.py +++ b/src/jiangchang_skill_core/media_assets.py @@ -27,6 +27,67 @@ _WIN_DEFAULT_DATA_ROOT = Path(r"D:\jiangchang-data") _NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang" _AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"} +_GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1" +_MIN_AUDIO_BYTES = 128 + + +def _is_git_lfs_pointer(path: Path) -> bool: + try: + head = path.read_bytes()[:256] + except OSError: + return False + return head.startswith(_GIT_LFS_POINTER_PREFIX) + + +def _enumerate_audio_like(music_dir: Path) -> list[Path]: + return [ + path + for path in music_dir.rglob("*") + if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS + ] + + +def _music_content_issue(music_dir: Optional[Path]) -> Optional[str]: + if music_dir is None or not music_dir.is_dir(): + return "music_dir_missing" + files = _enumerate_audio_like(music_dir) + if not files: + return "background_music_dir_empty" + if all(_is_git_lfs_pointer(path) for path in files): + return "background_music_lfs_pointer_only" + if not any(_is_usable_audio_file(path) for path in files): + return "background_music_invalid" + return None + + +def _music_is_ready(music_dir: Optional[Path]) -> bool: + return _music_content_issue(music_dir) is None + + +def _needs_music_content_repair(status: MediaAssetsStatus) -> bool: + if status.music_dir is None: + return "music_dir_missing" in status.warnings + issue = _music_content_issue(status.music_dir) + return issue in ( + "background_music_dir_empty", + "background_music_lfs_pointer_only", + "background_music_invalid", + ) + + +def _is_usable_audio_file(path: Path) -> bool: + if not path.is_file(): + return False + if path.suffix.lower() not in _AUDIO_EXTENSIONS: + return False + if _is_git_lfs_pointer(path): + return False + try: + if path.stat().st_size < _MIN_AUDIO_BYTES: + return False + except OSError: + return False + return True @dataclass @@ -39,6 +100,7 @@ class MediaAssetsStatus: ffmpeg_path: Optional[Path] = None ffprobe_path: Optional[Path] = None music_dir: Optional[Path] = None + music_ready: bool = False fonts_dir: Optional[Path] = None watermark_dir: Optional[Path] = None @@ -275,6 +337,13 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus: if not ffprobe_exists: warnings.append("ffprobe_missing") + music_ready = False + if music_exists: + music_issue = _music_content_issue(music_dir) + if music_issue and music_issue != "music_dir_missing": + warnings.append(music_issue) + music_ready = music_issue is None + ready = ( exists and music_exists @@ -282,6 +351,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus: and watermark_exists and ffmpeg_exists and ffprobe_exists + and music_ready ) return MediaAssetsStatus( @@ -293,6 +363,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus: ffmpeg_path=ffmpeg_path if ffmpeg_exists else None, ffprobe_path=ffprobe_path if ffprobe_exists else None, music_dir=music_dir if music_exists else None, + music_ready=music_ready, fonts_dir=fonts_dir if fonts_exists else None, watermark_dir=watermark_dir if watermark_exists else None, ) @@ -487,10 +558,43 @@ def _needs_media_repo_content(status: MediaAssetsStatus) -> bool: "music_dir_missing", "fonts_dir_missing", "watermark_dir_missing", + "background_music_dir_empty", + "background_music_lfs_pointer_only", + "background_music_invalid", ) ) +def _maybe_repair_music_content(root: Path, status: MediaAssetsStatus) -> MediaAssetsStatus: + if not _needs_music_content_repair(status): + return status + if _media_assets_root_overridden(None): + return status + + warnings: list[str] = [] + if _is_git_repo(root) and root.exists(): + if _try_download_via_git(root, warnings, clone=False): + refreshed = _inspect_media_assets(root, source="git") + for warning in warnings: + if warning not in refreshed.warnings: + refreshed.warnings.append(warning) + return _maybe_ensure_ffmpeg_tools(root, refreshed) + + shared = _shared_parent(root) + if _try_download_via_zip(shared, root, warnings): + refreshed = _inspect_media_assets(root, source="zip") + for warning in warnings: + if warning not in refreshed.warnings: + refreshed.warnings.append(warning) + return _maybe_ensure_ffmpeg_tools(root, refreshed) + + refreshed = _inspect_media_assets(root, source=status.source) + for warning in warnings: + if warning not in refreshed.warnings: + refreshed.warnings.append(warning) + return refreshed + + def ensure_media_assets( data_root: str | Path | None = None, env: dict[str, str] | None = None, @@ -530,16 +634,22 @@ def ensure_media_assets( if _try_download_via_zip(shared, root, warnings): return _maybe_ensure_ffmpeg_tools( root, - _inspect_media_assets(root, source="zip"), + _maybe_repair_music_content( + root, + _inspect_media_assets(root, source="zip"), + ), ) clone_needed = not root.exists() or not status.ready or update if _try_download_via_git(root, warnings, clone=clone_needed): return _maybe_ensure_ffmpeg_tools( root, - _inspect_media_assets( + _maybe_repair_music_content( root, - source="git" if _is_git_repo(root) else "local", + _inspect_media_assets( + root, + source="git" if _is_git_repo(root) else "local", + ), ), ) @@ -547,7 +657,29 @@ def ensure_media_assets( for warning in warnings: if warning not in final.warnings: final.warnings.append(warning) - return _maybe_ensure_ffmpeg_tools(root, final) + repaired = _maybe_repair_music_content(root, final) + for warning in warnings: + if warning not in repaired.warnings: + repaired.warnings.append(warning) + return _maybe_ensure_ffmpeg_tools(root, repaired) + + +def probe_media_assets( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> MediaAssetsStatus: + """只检查本地 media-assets 目录，不触发下载。""" + env_map = _env_dict(env) + root = resolve_media_assets_root(data_root, env_map) + return _inspect_media_assets(root, source="local") + + +def probe_ffmpeg( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[Path]: + status = probe_media_assets(data_root, env) + return status.ffmpeg_path def resolve_ffmpeg( @@ -574,6 +706,18 @@ def resolve_music_dir( return status.music_dir +def background_music_issue( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[str]: + """若 music 目录存在但无可用音频，返回标准 warning code。""" + if pick_background_music(data_root, env) is not None: + return None + + status = ensure_media_assets(data_root, env) + return _music_content_issue(status.music_dir) + + def pick_background_music( data_root: str | Path | None = None, env: dict[str, str] | None = None, @@ -583,10 +727,11 @@ def pick_background_music( if music_dir is None or not music_dir.is_dir(): return None - candidates: list[Path] = [] - for path in music_dir.rglob("*"): - if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS: - candidates.append(path) + candidates = [ + path + for path in music_dir.rglob("*") + if _is_usable_audio_file(path) + ] if not candidates: return None diff --git a/src/jiangchang_skill_core/rpa/video_session.py b/src/jiangchang_skill_core/rpa/video_session.py index 072d258..45d2d71 100644 --- a/src/jiangchang_skill_core/rpa/video_session.py +++ b/src/jiangchang_skill_core/rpa/video_session.py @@ -4,8 +4,6 @@ from __future__ import annotations import asyncio import os -import random -import shutil import subprocess import sys import time @@ -15,6 +13,12 @@ from pathlib import Path from typing import Any, Dict, IO, List, Optional from .. import config +from ..media_assets import ( + background_music_issue, + ensure_media_assets, + pick_background_music, + resolve_ffmpeg, +) _JIANGCHANG_SUBTITLE_STYLE = ( "FontName=Arial,FontSize=14," @@ -22,11 +26,6 @@ _JIANGCHANG_SUBTITLE_STYLE = ( "Outline=2,Shadow=1,Alignment=2" ) -_DEFAULT_MUSIC_ROOT = os.environ.get( - "MEDIA_ASSETS_ROOT", - r"D:\OpenClaw\client-commons\media-assets", -) - _CAPTURE_FRAMERATE = 15 @@ -34,14 +33,23 @@ def _record_video_enabled() -> bool: return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False) -def _resolve_music_file() -> Optional[Path]: - music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music" - if not music_dir.is_dir(): +def _resolve_ffmpeg_exe() -> Optional[Path]: + ffmpeg = resolve_ffmpeg() + if ffmpeg is None: return None - files = list(music_dir.rglob("*.mp3")) - if not files: - return None - return random.choice(files) + path = Path(ffmpeg) + return path if path.is_file() else None + + +def _resolve_music_file(warnings: List[str]) -> Optional[Path]: + music = pick_background_music() + if music is not None: + return music + + issue = background_music_issue() + if issue: + warnings.append(issue) + return None @dataclass @@ -87,7 +95,12 @@ def _escape_srt_path(path: str) -> str: return path.replace("\\", "/").replace(":", "\\:") -def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]: +def _ffmpeg_cmd(ffmpeg_exe: Path, *args: str) -> List[str]: + return [str(ffmpeg_exe), *args] + + +def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: List[str]) -> tuple[bool, str]: + cmd = _ffmpeg_cmd(ffmpeg_exe, *cmd_tail) try: result = subprocess.run( cmd, @@ -104,12 +117,12 @@ def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]: return True, "" -def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]: +def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]: """Windows gdigrab 全桌面录制；非 Windows 返回 None。""" if sys.platform != "win32": return None - return [ - "ffmpeg", + return _ffmpeg_cmd( + ffmpeg_exe, "-y", "-f", "gdigrab", @@ -124,10 +137,11 @@ def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]: "-pix_fmt", "yuv420p", capture_path, - ] + ) def _compose_capture_mp4( + ffmpeg_exe: Path, capture_path: Path, srt_path: Path, output_path: Path, @@ -147,8 +161,7 @@ def _compose_capture_mp4( f"[0:v]{subtitle_filter}[vout];" f"[1:a]volume={music_volume},apad[aout]" ) - cmd = [ - "ffmpeg", + cmd_tail = [ "-y", "-i", str(capture_path), @@ -176,8 +189,7 @@ def _compose_capture_mp4( str(output_path), ] else: - cmd = [ - "ffmpeg", + cmd_tail = [ "-y", "-i", str(capture_path), @@ -197,7 +209,7 @@ def _compose_capture_mp4( "yuv420p", str(output_path), ] - return _run_ffmpeg(cmd) + return _run_ffmpeg(ffmpeg_exe, cmd_tail) class RpaVideoSession: @@ -287,11 +299,13 @@ class RpaVideoSession: def _start_ffmpeg_capture(self) -> None: if not self.enabled: return - if not shutil.which("ffmpeg"): + + ffmpeg_exe = _resolve_ffmpeg_exe() + if ffmpeg_exe is None: self.warnings.append("ffmpeg_not_found") return - cmd = _build_desktop_record_cmd(self._capture_path) + cmd = _build_desktop_record_cmd(ffmpeg_exe, self._capture_path) if cmd is None: self._append_warning( "ffmpeg_record_start_failed", @@ -391,14 +405,21 @@ class RpaVideoSession: except Exception as exc: self.warnings.append(f"subtitle_write_failed: {exc}") - music = _resolve_music_file() - if music is None: + music = _resolve_music_file(self.warnings) + if music is None and "background_music_lfs_pointer_only" not in self.warnings and "background_music_invalid" not in self.warnings: self.warnings.append("background_music_missing") + ffmpeg_exe = _resolve_ffmpeg_exe() + if ffmpeg_exe is None: + self.warnings.append("ffmpeg_not_found") + self.artifacts = self.summary() + return + srt_file = Path(self._srt_path) use_srt = srt_file.is_file() and srt_file.stat().st_size > 0 ok, err = _compose_capture_mp4( + ffmpeg_exe, capture, srt_file if use_srt else Path(self._srt_path), Path(self._planned_output), @@ -407,6 +428,7 @@ class RpaVideoSession: if not ok and music is not None: self.warnings.append(f"ffmpeg_with_music_failed: {err}") ok, err = _compose_capture_mp4( + ffmpeg_exe, capture, srt_file if use_srt else Path(self._srt_path), Path(self._planned_output), @@ -414,8 +436,8 @@ class RpaVideoSession: ) if not ok and not use_srt: ok, err = _run_ffmpeg( + ffmpeg_exe, [ - "ffmpeg", "-y", "-i", str(capture), @@ -428,7 +450,7 @@ class RpaVideoSession: "-pix_fmt", "yuv420p", str(self._planned_output), - ] + ], ) if not ok: self._append_warning("ffmpeg_compose_failed", err) diff --git a/src/jiangchang_skill_core/runtime_env.py b/src/jiangchang_skill_core/runtime_env.py index 2092467..c201549 100644 --- a/src/jiangchang_skill_core/runtime_env.py +++ b/src/jiangchang_skill_core/runtime_env.py @@ -34,14 +34,22 @@ def platform_default_data_root() -> str: def get_data_root() -> str: - env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() + env = ( + os.getenv("CLAW_DATA_ROOT") + or os.getenv("JIANGCHANG_DATA_ROOT") + or "" + ).strip() if env: return env return platform_default_data_root() def get_user_id() -> str: - return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon" + return ( + os.getenv("CLAW_USER_ID") + or os.getenv("JIANGCHANG_USER_ID") + or "" + ).strip() or "_anon" def _looks_like_skills_root(path: str) -> bool: @@ -55,6 +63,7 @@ def _looks_like_skills_root(path: str) -> bool: "toutiao-publisher", "logistics-tracker", "api-key-vault", + "receive-order", ): if os.path.isdir(os.path.join(path, marker)): return True @@ -130,10 +139,21 @@ def apply_cli_local_defaults() -> None: enabled = v in ("1", "true", "yes", "on") if not enabled: return - if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip(): - os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root() - if not (os.getenv("JIANGCHANG_USER_ID") or "").strip(): + if not ( + os.getenv("CLAW_DATA_ROOT") + or os.getenv("JIANGCHANG_DATA_ROOT") + or "" + ).strip(): + default_root = platform_default_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = default_root + os.environ.setdefault("CLAW_DATA_ROOT", default_root) + if not ( + os.getenv("CLAW_USER_ID") + or os.getenv("JIANGCHANG_USER_ID") + or "" + ).strip(): os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip() + os.environ.setdefault("CLAW_USER_ID", DEFAULT_LOCAL_USER_ID.strip()) def find_chrome_executable() -> "str | None": """查找系统已安装的 Chrome 或 Edge 可执行文件路径。 diff --git a/src/screencast/composer.py b/src/screencast/composer.py index 4ea95b9..70c4fd9 100644 --- a/src/screencast/composer.py +++ b/src/screencast/composer.py @@ -1,16 +1,18 @@ """FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。""" from __future__ import annotations -import random import subprocess import sys +from dataclasses import dataclass, field from pathlib import Path +from jiangchang_skill_core.media_assets import ( + background_music_issue, + pick_background_music, + resolve_ffmpeg, +) # 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性 -# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感 -# Outline 2：14 号字配 Outline 3 会显胖，2 已足够在浅色背景上保持可读 -# 修改请评估对所有 skill 录屏的影响。 _JIANGCHANG_SUBTITLE_STYLE = ( "FontName=Arial,FontSize=14," "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000," @@ -18,63 +20,155 @@ _JIANGCHANG_SUBTITLE_STYLE = ( ) +@dataclass +class ComposeVideoResult: + output_path: str + warnings: list[str] = field(default_factory=list) + + +def _escape_srt_path(path: str) -> str: + return path.replace("\\", "/").replace(":", "\\:") + + +def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]: + cmd = [str(ffmpeg_exe), *cmd_tail] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + except FileNotFoundError: + return False, "ffmpeg not found" + if result.returncode != 0: + err = (result.stderr or result.stdout or "").strip() + return False, err[:500] if err else f"exit {result.returncode}" + return True, "" + + +def _media_assets_env( + media_assets_root: str | Path | None, + media_assets_env: dict[str, str] | None, +) -> dict[str, str] | None: + if media_assets_root is None and media_assets_env is None: + return None + env = dict(media_assets_env or {}) + if media_assets_root is not None: + env["MEDIA_ASSETS_ROOT"] = str(media_assets_root) + return env + + 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) + allow_no_music: bool = True, + media_assets_root: str | Path | None = None, + media_assets_env: dict[str, str] | None = None, +) -> ComposeVideoResult: + """合成 MP4；ffmpeg 与背景音乐均走 jiangchang_skill_core.media_assets。""" + frames_dir_path = Path(frames_dir) + subtitle_path_obj = Path(subtitle_path) + output_path_obj = Path(output_path) + output_path_obj.parent.mkdir(parents=True, exist_ok=True) + warnings: list[str] = [] - # 随机选一首背景音乐 - 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}") + assets_env = _media_assets_env(media_assets_root, media_assets_env) - # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法） - srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:") + ffmpeg_exe = resolve_ffmpeg(env=assets_env) + if ffmpeg_exe is None or not Path(ffmpeg_exe).is_file(): + raise RuntimeError("ffmpeg_not_found") + ffmpeg_path = Path(ffmpeg_exe) + music_file = pick_background_music(env=assets_env) + if music_file is None: + issue = background_music_issue(env=assets_env) + if issue: + warnings.append(issue) + else: + warnings.append("background_music_missing") + if not allow_no_music: + raise FileNotFoundError(warnings[-1]) + + srt_esc = _escape_srt_path(str(subtitle_path_obj.resolve())) subtitle_filter = ( - f"subtitles='{srt_path_str}'" + f"subtitles='{srt_esc}'" 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]" - ) + if music_file and music_file.is_file(): + filter_complex = ( + f"[0:v]{subtitle_filter}[vout];" + f"[1:a]volume={music_volume},apad[aout]" + ) + cmd_tail = [ + "-y", + "-framerate", + str(fps), + "-i", + str(frames_dir_path / "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_obj), + ] + ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail) + if not ok: + warnings.append(f"ffmpeg_with_music_failed: {err}") + if not allow_no_music: + raise RuntimeError(f"ffmpeg_compose_failed: {err}") + music_file = None + else: + print(f"[screencast] 背景音乐: {music_file.name}") + print(f"[screencast] 输出: {output_path_obj}") + return ComposeVideoResult(str(output_path_obj.resolve()), warnings) - 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), + cmd_tail = [ + "-y", + "-framerate", + str(fps), + "-i", + str(frames_dir_path / "frame_%06d.png"), + "-vf", + subtitle_filter, + "-c:v", + "libx264", + "-preset", + "fast", + "-crf", + "23", + "-pix_fmt", + "yuv420p", + str(output_path_obj), ] + print("[screencast] 运行 FFmpeg 合成（无背景音乐）...") + ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail) + if not ok: + print(err, file=sys.stderr) + raise RuntimeError(f"ffmpeg_compose_failed: {err}") - 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) + print(f"[screencast] 输出: {output_path_obj}") + return ComposeVideoResult(str(output_path_obj.resolve()), warnings) diff --git a/src/screencast/runner.py b/src/screencast/runner.py index 856a90e..bb7c974 100644 --- a/src/screencast/runner.py +++ b/src/screencast/runner.py @@ -30,6 +30,8 @@ def _ensure_sdk_on_path() -> None: _ensure_sdk_on_path() from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window # noqa: E402 +import warnings + from .composer import compose_video from .recorder import ScreenRecorder from .subtitle import SubtitleEngine @@ -54,10 +56,11 @@ def run_screencast( 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" + media_assets_root: media-assets 根目录；不传时使用 + ``resolve_media_assets_root()`` 默认路径 + （JIANGCHANG_DATA_ROOT/shared/media-assets 或 MEDIA_ASSETS_ROOT） + music_subdir: **已废弃**，保留仅为兼容旧调用；背景音乐由 + ``media_assets.pick_background_music()`` 统一解析 fps: 截帧帧率，默认 10 activate_window: 录屏开始前是否自动激活+最大化匠厂窗口（默认 True； Windows 走 ctypes，非 Windows 走 jiangchang:// 协议） @@ -67,13 +70,14 @@ def run_screencast( Returns: 输出 MP4 的绝对路径 """ - if media_assets_root is None: - media_assets_root = os.environ.get( - "MEDIA_ASSETS_ROOT", - r"D:\OpenClaw\client-commons\media-assets", + if music_subdir != "music": + warnings.warn( + "music_subdir is deprecated and ignored; " + "background music is resolved via media_assets.pick_background_music()", + DeprecationWarning, + stacklevel=2, ) - music_dir = str(Path(media_assets_root) / music_subdir) output_dir_path = Path(output_dir) output_dir_path.mkdir(parents=True, exist_ok=True) @@ -135,8 +139,8 @@ def run_screencast( frames_dir=frames_dir, fps=fps, subtitle_path=str(subtitle_file), - music_dir=music_dir, output_path=str(output_mp4), + media_assets_root=media_assets_root, ) return str(output_mp4) diff --git a/tests/test_config.py b/tests/test_config.py index 4abc86b..85ba6ac 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -81,6 +81,27 @@ class TestConfig(unittest.TestCase): self.assertEqual(config.get_float("BAD", 9.9), 9.9) self.assertEqual(config.get_int("BAD", 7), 7) + def test_merge_missing_env_keys_appends_only_new(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + example = os.path.join(tmp, ".env.example") + dest = os.path.join(tmp, ".env") + with open(example, "w", encoding="utf-8") as f: + f.write("EXISTING=1\nNEW_KEY=abc\n") + with open(dest, "w", encoding="utf-8") as f: + f.write("EXISTING=keep\n") + + added = config.merge_missing_env_keys( + example, + dest, + comment_skill="demo-skill", + ) + self.assertEqual(added, ["NEW_KEY"]) + with open(dest, encoding="utf-8") as f: + content = f.read() + self.assertIn("EXISTING=keep", content) + self.assertIn("NEW_KEY=abc", content) + self.assertIn("Added by demo-skill", content) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_media_assets.py b/tests/test_media_assets.py index 9768e42..b8e1fb7 100644 --- a/tests/test_media_assets.py +++ b/tests/test_media_assets.py @@ -15,6 +15,7 @@ from jiangchang_skill_core import media_assets as ma def _make_complete_assets(root: Path, *, with_ffmpeg: bool = True, with_ffprobe: bool = True) -> None: (root / "music" / "calm").mkdir(parents=True) + (root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 256) (root / "fonts").mkdir(parents=True) (root / "watermark").mkdir(parents=True) bin_key = ma._platform_bin_key() @@ -117,9 +118,9 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest. root = tmp_path / "data" / "shared" / "media-assets" _make_complete_assets(root) music = root / "music" - (music / "b-upbeat.mp3").write_bytes(b"mp3") - (music / "calm" / "a-calm.mp3").write_bytes(b"mp3") - (music / "z-last.wav").write_bytes(b"wav") + (music / "b-upbeat.mp3").write_bytes(b"x" * 256) + (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 256) + (music / "z-last.wav").write_bytes(b"x" * 256) monkeypatch.setattr( ma, @@ -136,7 +137,7 @@ def _build_archive_zip(dest: Path, inner_name: str) -> None: with zipfile.ZipFile(dest, "w") as zf: zf.writestr(f"{inner_name}/README.md", "# media-assets\n") zf.writestr(f"{inner_name}/manifest.json", json.dumps(manifest, ensure_ascii=False)) - zf.writestr(f"{inner_name}/music/.keep", "") + zf.writestr(f"{inner_name}/music/calm/track.mp3", b"x" * 256) zf.writestr(f"{inner_name}/fonts/.keep", "") zf.writestr(f"{inner_name}/watermark/.keep", "") @@ -213,9 +214,117 @@ def test_windows_default_data_root() -> None: assert root == Path(r"D:\jiangchang-data") / "shared" / "media-assets" +def test_probe_media_assets_does_not_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + def _fail_download(*args, **kwargs): + raise AssertionError("download should not be called") + + monkeypatch.setattr(ma, "_try_download_via_zip", _fail_download) + monkeypatch.setattr(ma, "_try_download_via_git", _fail_download) + + status = ma.probe_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert status.exists is False + assert status.ffmpeg_path is None + + +def test_pick_background_music_skips_lfs_pointer(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root, with_ffmpeg=True, with_ffprobe=True) + music = root / "music" + (music / "calm" / "track.mp3").unlink() + (music / "real.mp3").write_bytes(b"x" * 256) + lfs = music / "lfs-only.mp3" + lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n") + + monkeypatch.setattr( + ma, + "ensure_media_assets", + lambda *a, **k: ma._inspect_media_assets(root, source="local"), + ) + + picked = ma.pick_background_music(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert picked == music / "real.mp3" + + +def test_pick_background_music_no_valid_files(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root) + music = root / "music" + (music / "calm" / "track.mp3").unlink() + (music / "pointer.mp3").write_text( + "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n" + ) + + monkeypatch.setattr( + ma, + "ensure_media_assets", + lambda *a, **k: ma._inspect_media_assets(root, source="local"), + ) + + picked = ma.pick_background_music(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert picked is None + + +def test_video_finalize_without_music_fallback(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import asyncio + import os + import tempfile + from pathlib import Path + + from jiangchang_skill_core import config + from jiangchang_skill_core.rpa.video_session import RpaVideoSession + + ffmpeg_exe = tmp_path / "ffmpeg.exe" + ffmpeg_exe.write_bytes(b"ffmpeg") + + def _fake_resolve_ffmpeg(*args, **kwargs): + return ffmpeg_exe + + def _fake_pick_music(*args, **kwargs): + return None + + monkeypatch.setattr( + "jiangchang_skill_core.rpa.video_session.resolve_ffmpeg", + _fake_resolve_ffmpeg, + ) + monkeypatch.setattr( + "jiangchang_skill_core.rpa.video_session.pick_background_music", + _fake_pick_music, + ) + monkeypatch.setattr( + "jiangchang_skill_core.rpa.video_session.background_music_issue", + lambda *a, **k: None, + ) + + os.environ["OPENCLAW_RECORD_VIDEO"] = "1" + try: + config.reset_cache() + with tempfile.TemporaryDirectory() as skill_data: + session = RpaVideoSession( + skill_slug="receive-order", + skill_data_dir=skill_data, + batch_id="batch_no_music", + ) + capture = Path(session._capture_path) + capture.parent.mkdir(parents=True, exist_ok=True) + capture.write_bytes(b"\x00" * 64) + + monkeypatch.setattr( + "jiangchang_skill_core.rpa.video_session._compose_capture_mp4", + lambda *a, **k: (True, ""), + ) + + asyncio.run(session.finalize()) + assert "background_music_missing" in session.warnings + assert session.output_video_path is not None + finally: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + config.reset_cache() + + def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: root = tmp_path / "data" / "shared" / "media-assets" - (root / "music").mkdir(parents=True) + (root / "music" / "calm").mkdir(parents=True) + (root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 256) (root / "fonts").mkdir(parents=True) (root / "watermark").mkdir(parents=True) (root / "manifest.json").write_text( diff --git a/tests/test_screencast_composer.py b/tests/test_screencast_composer.py new file mode 100644 index 0000000..cbb845d --- /dev/null +++ b/tests/test_screencast_composer.py @@ -0,0 +1,252 @@ +# -*- coding: utf-8 -*- +"""screencast composer 媒体资源集成测试。""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +from screencast.composer import ComposeVideoResult, compose_video + + +def _write_frame(frames_dir: Path) -> None: + frames_dir.mkdir(parents=True, exist_ok=True) + (frames_dir / "frame_000001.png").write_bytes(b"png") + + +def _write_srt(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("1\n00:00:00,000 --> 00:00:02,000\nhello\n", encoding="utf-8") + + +def test_compose_video_uses_absolute_ffmpeg(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + ffmpeg = tmp_path / "ffmpeg.exe" + ffmpeg.write_bytes(b"ffmpeg") + _write_frame(frames) + _write_srt(srt) + + captured: list[list[str]] = [] + + def _fake_run(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]: + captured.append([str(ffmpeg_exe), *cmd_tail]) + output.write_bytes(b"mp4") + return True, "" + + with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch( + "screencast.composer.pick_background_music", + return_value=None, + ), patch("screencast.composer.background_music_issue", return_value="background_music_missing"), patch( + "screencast.composer._run_ffmpeg", + side_effect=_fake_run, + ): + result = compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + ) + + assert isinstance(result, ComposeVideoResult) + assert captured + assert captured[0][0] == str(ffmpeg) + assert "background_music_missing" in result.warnings + + +def test_compose_video_with_music(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + ffmpeg = tmp_path / "ffmpeg.exe" + music = tmp_path / "music.mp3" + ffmpeg.write_bytes(b"ffmpeg") + music.write_bytes(b"x" * 256) + _write_frame(frames) + _write_srt(srt) + + with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch( + "screencast.composer.pick_background_music", + return_value=music, + ), patch( + "screencast.composer._run_ffmpeg", + return_value=(True, ""), + ): + result = compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + ) + output.write_bytes(b"mp4") + assert result.warnings == [] + + +def test_compose_video_music_failure_fallback(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + ffmpeg = tmp_path / "ffmpeg.exe" + music = tmp_path / "music.mp3" + ffmpeg.write_bytes(b"ffmpeg") + music.write_bytes(b"x" * 256) + _write_frame(frames) + _write_srt(srt) + + calls = {"count": 0} + + def _fake_run(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]: + calls["count"] += 1 + if calls["count"] == 1: + return False, "music mux failed" + output.write_bytes(b"mp4") + return True, "" + + with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch( + "screencast.composer.pick_background_music", + return_value=music, + ), patch( + "screencast.composer._run_ffmpeg", + side_effect=_fake_run, + ): + result = compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + ) + + assert any("ffmpeg_with_music_failed" in w for w in result.warnings) + assert calls["count"] == 2 + + +def test_compose_video_passes_media_assets_root_to_media_apis(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + custom_root = tmp_path / "custom-media-assets" + custom_root.mkdir() + ffmpeg = custom_root / "ffmpeg.exe" + ffmpeg.write_bytes(b"ffmpeg") + _write_frame(frames) + _write_srt(srt) + + captured_env: list[dict[str, str] | None] = [] + + def _capture_env(*args, **kwargs): + captured_env.append(kwargs.get("env")) + if len(captured_env) == 1: + return ffmpeg + if len(captured_env) == 2: + return None + return "background_music_missing" + + with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch( + "screencast.composer.pick_background_music", + side_effect=_capture_env, + ), patch( + "screencast.composer.background_music_issue", + side_effect=_capture_env, + ), patch( + "screencast.composer._run_ffmpeg", + return_value=(True, ""), + ): + compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + media_assets_root=str(custom_root), + ) + output.write_bytes(b"mp4") + + assert len(captured_env) == 3 + for env in captured_env: + assert env is not None + assert env["MEDIA_ASSETS_ROOT"] == str(custom_root) + + +def test_compose_video_default_does_not_set_media_assets_root(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + ffmpeg = tmp_path / "ffmpeg.exe" + ffmpeg.write_bytes(b"ffmpeg") + _write_frame(frames) + _write_srt(srt) + + captured_env: list[dict[str, str] | None] = [] + + def _capture_env(*args, **kwargs): + captured_env.append(kwargs.get("env")) + if len(captured_env) == 1: + return ffmpeg + if len(captured_env) == 2: + return None + return "background_music_missing" + + with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch( + "screencast.composer.pick_background_music", + side_effect=_capture_env, + ), patch( + "screencast.composer.background_music_issue", + side_effect=_capture_env, + ), patch( + "screencast.composer._run_ffmpeg", + return_value=(True, ""), + ): + compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + ) + + assert all(env is None for env in captured_env) + + +def test_compose_video_media_assets_root_overrides_env(tmp_path: Path) -> None: + frames = tmp_path / "frames" + srt = tmp_path / "out.srt" + output = tmp_path / "final.mp4" + custom_root = tmp_path / "override-root" + custom_root.mkdir() + ffmpeg = custom_root / "ffmpeg.exe" + ffmpeg.write_bytes(b"ffmpeg") + _write_frame(frames) + _write_srt(srt) + + original_env = {"MEDIA_ASSETS_ROOT": "/old/path", "OTHER": "keep"} + captured_env: list[dict[str, str] | None] = [] + + def _capture_env(*args, **kwargs): + captured_env.append(kwargs.get("env")) + if len(captured_env) == 1: + return ffmpeg + if len(captured_env) == 2: + return None + return "background_music_missing" + + with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch( + "screencast.composer.pick_background_music", + side_effect=_capture_env, + ), patch( + "screencast.composer.background_music_issue", + side_effect=_capture_env, + ), patch( + "screencast.composer._run_ffmpeg", + return_value=(True, ""), + ): + compose_video( + frames_dir=str(frames), + fps=10, + subtitle_path=str(srt), + output_path=str(output), + media_assets_root=str(custom_root), + media_assets_env=original_env, + ) + output.write_bytes(b"mp4") + + assert original_env["MEDIA_ASSETS_ROOT"] == "/old/path" + assert captured_env[0]["MEDIA_ASSETS_ROOT"] == str(custom_root) + assert captured_env[0]["OTHER"] == "keep" diff --git a/tests/test_screencast_runner.py b/tests/test_screencast_runner.py new file mode 100644 index 0000000..ddb2337 --- /dev/null +++ b/tests/test_screencast_runner.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""screencast runner 媒体资源参数传递测试。""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from screencast.runner import run_screencast + + +def test_run_screencast_passes_media_assets_root_to_compose_video(tmp_path: Path) -> None: + custom_root = tmp_path / "media-assets" + custom_root.mkdir() + captured: dict = {} + + mock_recorder = MagicMock() + mock_recorder.frame_count = 1 + + def _fake_compose_video(**kwargs): + captured.update(kwargs) + Path(kwargs["output_path"]).write_bytes(b"mp4") + from screencast.composer import ComposeVideoResult + + return ComposeVideoResult(kwargs["output_path"]) + + with patch("screencast.runner.activate_and_maximize_jiangchang_window", return_value=True), patch( + "screencast.runner.ScreenRecorder", + return_value=mock_recorder, + ), patch("screencast.runner.SubtitleEngine") as subtitle_cls, patch( + "screencast.runner.subprocess.Popen", + ) as popen_cls, patch( + "screencast.runner.compose_video", + side_effect=_fake_compose_video, + ): + subtitle = subtitle_cls.return_value + proc = popen_cls.return_value + proc.stdout = iter([]) + proc.wait.return_value = 0 + + out = run_screencast( + skill_slug="demo-skill", + subtitle_script=[("ok", "done")], + pytest_args=["-q", "tests/test_demo.py"], + output_dir=str(tmp_path / "out"), + media_assets_root=str(custom_root), + ) + + assert Path(out).is_file() + assert captured.get("media_assets_root") == str(custom_root) + + +def test_run_screencast_music_subdir_deprecated_warning() -> None: + with pytest.warns(DeprecationWarning, match="music_subdir is deprecated"): + with patch("screencast.runner.activate_and_maximize_jiangchang_window", return_value=False), patch( + "screencast.runner.ScreenRecorder", + ) as rec_cls, patch("screencast.runner.SubtitleEngine"), patch( + "screencast.runner.subprocess.Popen", + ) as popen_cls, patch( + "screencast.runner.compose_video", + ) as compose_mock: + rec = rec_cls.return_value + rec.frame_count = 0 + proc = popen_cls.return_value + proc.stdout = iter([]) + proc.wait.return_value = 0 + from screencast.composer import ComposeVideoResult + + compose_mock.return_value = ComposeVideoResult("out.mp4") + + run_screencast( + skill_slug="demo", + subtitle_script=[], + pytest_args=["-q"], + output_dir=".", + music_subdir="calm", + ) diff --git a/tests/test_video_session.py b/tests/test_video_session.py index 8e0f537..be41f28 100644 --- a/tests/test_video_session.py +++ b/tests/test_video_session.py @@ -7,8 +7,8 @@ import os import sys import tempfile import unittest +from pathlib import Path from unittest.mock import MagicMock, patch - from jiangchang_skill_core import config from jiangchang_skill_core.rpa.video_session import RpaVideoSession @@ -70,9 +70,10 @@ class TestRpaVideoSession(unittest.TestCase): config.reset_cache() @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture") - @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value="ffmpeg") + @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe") @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen") - def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, _which: MagicMock) -> None: + def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, mock_resolve: MagicMock) -> None: + mock_resolve.return_value = Path(r"C:\media\bin\win32-x64\ffmpeg.exe") proc = MagicMock() proc.poll.return_value = None proc.stdin = MagicMock() @@ -96,7 +97,7 @@ class TestRpaVideoSession(unittest.TestCase): self.assertTrue(mock_popen.called) cmd = mock_popen.call_args[0][0] - self.assertIn("ffmpeg", cmd) + self.assertTrue(str(cmd[0]).endswith("ffmpeg.exe")) self.assertIn("gdigrab", cmd) self.assertIn("desktop", cmd) kwargs = mock_popen.call_args[1] @@ -105,8 +106,8 @@ class TestRpaVideoSession(unittest.TestCase): os.environ.pop("OPENCLAW_RECORD_VIDEO", None) config.reset_cache() - @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value=None) - def test_ffmpeg_not_found_warning(self, _which: MagicMock) -> None: + @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe", return_value=None) + def test_ffmpeg_not_found_warning(self, _resolve: MagicMock) -> None: os.environ["OPENCLAW_RECORD_VIDEO"] = "1" try: with tempfile.TemporaryDirectory() as tmp: