4 Commits

Author SHA1 Message Date
c9398451d0 Release v1.0.11: shared runtime diagnostics and media_assets probe helpers
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 23s
2026-06-07 09:06:41 +08:00
2a542c0be6 ci: publish platform kit on main only
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 29s
2026-06-06 16:26:24 +08:00
e59661f2d8 fix(media-assets): download stable release bundle
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-06 16:16:18 +08:00
434a405e19 Release v1.0.10: shared media_assets and screencast media_assets_root override
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 17s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-06 10:32:35 +08:00
17 changed files with 1808 additions and 136 deletions

View File

@@ -4,9 +4,6 @@ on:
push: push:
branches: branches:
- main - main
tags:
- "v*"
workflow_dispatch:
jobs: jobs:
publish: publish:

View File

@@ -86,15 +86,59 @@ send_prompt_via_composer(page, "第一行\n第二行")
读取背景音乐、字体、水印和 ffmpeg 工具。 读取背景音乐、字体、水印和 ffmpeg 工具。
如果目录不存在,会优先通过 ZIP 下载公开仓库 如果目录不存在,会优先下载稳定 release bundle不依赖用户电脑安装 Git
`https://git.jc2009.com/client-commons/media-assets/archive/main.zip` `https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip`
这不依赖用户电脑安装 Git bundle 内含背景音乐、字体与水印;**不包含** ffmpeg 二进制。ffmpeg / ffprobe 仍由 bundle 内 `manifest.json``tools.ffmpeg` 下载源单独拉取
可通过环境变量覆盖: 可通过环境变量覆盖:
`MEDIA_ASSETS_ROOT=D:\OpenClaw\client-commons\media-assets` - `MEDIA_ASSETS_ROOT=/path/to/custom/media-assets` — 本地资源目录
- `MEDIA_ASSETS_BUNDLE_URL=https://example.com/media-assets.zip` — bundle 下载地址
### Runtime diagnostics统一 health 诊断)
`jiangchang_skill_core.runtime_diagnostics` 提供各 Skill 共用的 Runtime / media-assets / ffmpeg 探测避免每个技能各自复制检查逻辑。Skill 的 `health` 命令可调用:
```python
from jiangchang_skill_core import (
collect_runtime_diagnostics,
format_runtime_health_lines,
runtime_diagnostics_dict,
)
diag = collect_runtime_diagnostics(
skill_slug="my-skill",
platform_kit_min_version="1.0.10", # 可选;省略则只报告当前版本
skill_root="/path/to/my-skill", # 可选;用于检测 vendored jiangchang_skill_core
)
for line in format_runtime_health_lines(diag):
print(line)
print(runtime_diagnostics_dict(diag)) # 可 JSON 序列化
```
诊断只读探测本地状态,不会触发 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. 发版流程(维护者) ## 5. 发版流程(维护者)

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "jiangchang-platform-kit" name = "jiangchang-platform-kit"
version = "1.0.9" version = "1.0.11"
description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK"
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"

View File

@@ -23,13 +23,28 @@ from .unified_logging import (
) )
from .media_assets import ( from .media_assets import (
MediaAssetsStatus, MediaAssetsStatus,
background_music_issue,
ensure_media_assets, ensure_media_assets,
is_git_lfs_pointer,
is_usable_audio_file,
pick_background_music, pick_background_music,
probe_background_music,
probe_ffmpeg,
probe_media_assets,
resolve_ffmpeg, resolve_ffmpeg,
resolve_ffprobe, resolve_ffprobe,
resolve_media_assets_root, resolve_media_assets_root,
resolve_music_dir, resolve_music_dir,
) )
from .runtime_diagnostics import (
RuntimeDiagnostics,
RuntimeIssue,
collect_runtime_diagnostics,
format_runtime_health_lines,
is_jiangchang_skill_core_from_skill_tree,
runtime_diagnostics_dict,
version_ge,
)
try: try:
from . import rpa from . import rpa
@@ -43,12 +58,25 @@ __all__ = [
"EntitlementResult", "EntitlementResult",
"EntitlementServiceError", "EntitlementServiceError",
"MediaAssetsStatus", "MediaAssetsStatus",
"RuntimeDiagnostics",
"RuntimeIssue",
"background_music_issue",
"collect_runtime_diagnostics",
"ensure_media_assets", "ensure_media_assets",
"format_runtime_health_lines",
"is_git_lfs_pointer",
"is_jiangchang_skill_core_from_skill_tree",
"is_usable_audio_file",
"probe_background_music",
"probe_ffmpeg",
"probe_media_assets",
"pick_background_music", "pick_background_music",
"resolve_ffmpeg", "resolve_ffmpeg",
"resolve_ffprobe", "resolve_ffprobe",
"resolve_media_assets_root", "resolve_media_assets_root",
"resolve_music_dir", "resolve_music_dir",
"runtime_diagnostics_dict",
"version_ge",
"apply_cli_local_defaults", "apply_cli_local_defaults",
"attach_unified_file_handler", "attach_unified_file_handler",
"ensure_env_file", "ensure_env_file",

View File

@@ -69,6 +69,69 @@ def reset_cache() -> None:
_example_defaults_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: def ensure_env_file(skill_slug: str, example_path: str) -> str:
"""首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env已存在不覆盖返回路径。""" """首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env已存在不覆盖返回路径。"""
global _skill_slug, _example_path, _env_file_path global _skill_slug, _example_path, _env_file_path

View File

@@ -16,17 +16,86 @@ import zipfile
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Any, Mapping, Optional
MEDIA_ASSETS_ZIP_URL = ( MEDIA_ASSETS_BUNDLE_URL = (
"https://git.jc2009.com/client-commons/media-assets/archive/main.zip" "https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip"
) )
MEDIA_ASSETS_ZIP_URL = MEDIA_ASSETS_BUNDLE_URL
MEDIA_ASSETS_GIT_URL = "https://git.jc2009.com/client-commons/media-assets" MEDIA_ASSETS_GIT_URL = "https://git.jc2009.com/client-commons/media-assets"
_WIN_DEFAULT_DATA_ROOT = Path(r"D:\jiangchang-data") _WIN_DEFAULT_DATA_ROOT = Path(r"D:\jiangchang-data")
_NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang" _NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang"
_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"} _AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"}
_GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1"
_MIN_AUDIO_BYTES = 1024
def is_git_lfs_pointer(path: str | Path) -> bool:
try:
head = Path(path).read_bytes()[:256]
except OSError:
return False
return head.startswith(_GIT_LFS_POINTER_PREFIX)
_is_git_lfs_pointer = is_git_lfs_pointer
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: str | Path, *, min_bytes: int = _MIN_AUDIO_BYTES) -> bool:
p = Path(path)
if not p.is_file():
return False
if p.suffix.lower() not in _AUDIO_EXTENSIONS:
return False
if is_git_lfs_pointer(p):
return False
try:
if p.stat().st_size < min_bytes:
return False
except OSError:
return False
return True
_is_usable_audio_file = is_usable_audio_file
@dataclass @dataclass
@@ -39,6 +108,7 @@ class MediaAssetsStatus:
ffmpeg_path: Optional[Path] = None ffmpeg_path: Optional[Path] = None
ffprobe_path: Optional[Path] = None ffprobe_path: Optional[Path] = None
music_dir: Optional[Path] = None music_dir: Optional[Path] = None
music_ready: bool = False
fonts_dir: Optional[Path] = None fonts_dir: Optional[Path] = None
watermark_dir: Optional[Path] = None watermark_dir: Optional[Path] = None
@@ -47,6 +117,12 @@ def _env_dict(env: dict[str, str] | None) -> dict[str, str]:
return env if env is not None else dict(os.environ) return env if env is not None else dict(os.environ)
def _media_assets_bundle_url(env: dict[str, str] | None = None) -> str:
env_map = _env_dict(env)
override = (env_map.get("MEDIA_ASSETS_BUNDLE_URL") or "").strip()
return override or MEDIA_ASSETS_BUNDLE_URL
def _resolve_jiangchang_data_root( def _resolve_jiangchang_data_root(
data_root: str | Path | None = None, data_root: str | Path | None = None,
env: dict[str, str] | None = None, env: dict[str, str] | None = None,
@@ -275,6 +351,13 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
if not ffprobe_exists: if not ffprobe_exists:
warnings.append("ffprobe_missing") 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 = ( ready = (
exists exists
and music_exists and music_exists
@@ -282,6 +365,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
and watermark_exists and watermark_exists
and ffmpeg_exists and ffmpeg_exists
and ffprobe_exists and ffprobe_exists
and music_ready
) )
return MediaAssetsStatus( return MediaAssetsStatus(
@@ -293,6 +377,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
ffmpeg_path=ffmpeg_path if ffmpeg_exists else None, ffmpeg_path=ffmpeg_path if ffmpeg_exists else None,
ffprobe_path=ffprobe_path if ffprobe_exists else None, ffprobe_path=ffprobe_path if ffprobe_exists else None,
music_dir=music_dir if music_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, fonts_dir=fonts_dir if fonts_exists else None,
watermark_dir=watermark_dir if watermark_exists else None, watermark_dir=watermark_dir if watermark_exists else None,
) )
@@ -403,12 +488,16 @@ def _try_download_via_zip(
shared: Path, shared: Path,
target: Path, target: Path,
warnings: list[str], warnings: list[str],
*,
env: dict[str, str] | None = None,
url: str | None = None,
) -> bool: ) -> bool:
download_root = _download_dir(shared) download_root = _download_dir(shared)
download_root.mkdir(parents=True, exist_ok=True) download_root.mkdir(parents=True, exist_ok=True)
zip_path = download_root / "media-assets.zip" zip_path = download_root / "media-assets.zip"
extracted = download_root / "extracted" extracted = download_root / "extracted"
staging = download_root / "staging" staging = download_root / "staging"
bundle_url = url or _media_assets_bundle_url(env)
try: try:
if staging.exists(): if staging.exists():
@@ -416,7 +505,7 @@ def _try_download_via_zip(
if extracted.exists(): if extracted.exists():
shutil.rmtree(extracted, ignore_errors=True) shutil.rmtree(extracted, ignore_errors=True)
_download_zip(MEDIA_ASSETS_ZIP_URL, zip_path) _download_zip(bundle_url, zip_path)
_extract_zip(zip_path, extracted) _extract_zip(zip_path, extracted)
found = _find_media_assets_root_in_tree(extracted) found = _find_media_assets_root_in_tree(extracted)
@@ -487,10 +576,48 @@ def _needs_media_repo_content(status: MediaAssetsStatus) -> bool:
"music_dir_missing", "music_dir_missing",
"fonts_dir_missing", "fonts_dir_missing",
"watermark_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,
*,
env: dict[str, str] | None = None,
) -> MediaAssetsStatus:
if not _needs_music_content_repair(status):
return status
if _media_assets_root_overridden(env):
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, env=env):
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( def ensure_media_assets(
data_root: str | Path | None = None, data_root: str | Path | None = None,
env: dict[str, str] | None = None, env: dict[str, str] | None = None,
@@ -527,19 +654,27 @@ def ensure_media_assets(
shared.mkdir(parents=True, exist_ok=True) shared.mkdir(parents=True, exist_ok=True)
warnings: list[str] = [] warnings: list[str] = []
if _try_download_via_zip(shared, root, warnings): if _try_download_via_zip(shared, root, warnings, env=env_map):
return _maybe_ensure_ffmpeg_tools( return _maybe_ensure_ffmpeg_tools(
root, root,
_inspect_media_assets(root, source="zip"), _maybe_repair_music_content(
root,
_inspect_media_assets(root, source="zip"),
env=env_map,
),
) )
clone_needed = not root.exists() or not status.ready or update clone_needed = not root.exists() or not status.ready or update
if _try_download_via_git(root, warnings, clone=clone_needed): if _try_download_via_git(root, warnings, clone=clone_needed):
return _maybe_ensure_ffmpeg_tools( return _maybe_ensure_ffmpeg_tools(
root, root,
_inspect_media_assets( _maybe_repair_music_content(
root, root,
source="git" if _is_git_repo(root) else "local", _inspect_media_assets(
root,
source="git" if _is_git_repo(root) else "local",
),
env=env_map,
), ),
) )
@@ -547,7 +682,29 @@ def ensure_media_assets(
for warning in warnings: for warning in warnings:
if warning not in final.warnings: if warning not in final.warnings:
final.warnings.append(warning) final.warnings.append(warning)
return _maybe_ensure_ffmpeg_tools(root, final) repaired = _maybe_repair_music_content(root, final, env=env_map)
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( def resolve_ffmpeg(
@@ -574,6 +731,58 @@ def resolve_music_dir(
return status.music_dir return status.music_dir
def probe_background_music(
root: str | Path | None = None,
*,
env: Mapping[str, str] | None = None,
) -> dict[str, Any]:
"""只读探测背景音乐目录,不触发 media-assets 下载。"""
env_map = _env_dict(dict(env) if env is not None else None)
if root is not None:
music_dir = Path(root) / "music"
if not music_dir.is_dir():
music_dir = Path(root) if Path(root).is_dir() else None
else:
data_root = (env_map.get("JIANGCHANG_DATA_ROOT") or env_map.get("CLAW_DATA_ROOT") or "").strip()
assets_root = resolve_media_assets_root(data_root or None, env_map)
status = _inspect_media_assets(assets_root, source="local")
music_dir = status.music_dir
music_root = str(music_dir) if music_dir is not None else None
if music_dir is None or not music_dir.is_dir():
return {
"music_root": music_root,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": "music_dir_missing",
}
mp3_files = [path for path in music_dir.rglob("*.mp3") if path.is_file()]
usable = [path for path in music_dir.rglob("*") if is_usable_audio_file(path)]
issue = _music_content_issue(music_dir)
sample = str(sorted(usable, key=lambda p: str(p).lower())[0]) if usable else None
return {
"music_root": music_root,
"mp3_count": len(mp3_files),
"usable_count": len(usable),
"sample_path": sample,
"issue": issue,
}
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( def pick_background_music(
data_root: str | Path | None = None, data_root: str | Path | None = None,
env: dict[str, str] | None = None, env: dict[str, str] | None = None,
@@ -583,10 +792,11 @@ def pick_background_music(
if music_dir is None or not music_dir.is_dir(): if music_dir is None or not music_dir.is_dir():
return None return None
candidates: list[Path] = [] candidates = [
for path in music_dir.rglob("*"): path
if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS: for path in music_dir.rglob("*")
candidates.append(path) if _is_usable_audio_file(path)
]
if not candidates: if not candidates:
return None return None

View File

@@ -4,8 +4,6 @@ from __future__ import annotations
import asyncio import asyncio
import os import os
import random
import shutil
import subprocess import subprocess
import sys import sys
import time import time
@@ -15,6 +13,12 @@ from pathlib import Path
from typing import Any, Dict, IO, List, Optional from typing import Any, Dict, IO, List, Optional
from .. import config from .. import config
from ..media_assets import (
background_music_issue,
ensure_media_assets,
pick_background_music,
resolve_ffmpeg,
)
_JIANGCHANG_SUBTITLE_STYLE = ( _JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14," "FontName=Arial,FontSize=14,"
@@ -22,11 +26,6 @@ _JIANGCHANG_SUBTITLE_STYLE = (
"Outline=2,Shadow=1,Alignment=2" "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 _CAPTURE_FRAMERATE = 15
@@ -34,14 +33,23 @@ def _record_video_enabled() -> bool:
return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False) return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
def _resolve_music_file() -> Optional[Path]: def _resolve_ffmpeg_exe() -> Optional[Path]:
music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music" ffmpeg = resolve_ffmpeg()
if not music_dir.is_dir(): if ffmpeg is None:
return None return None
files = list(music_dir.rglob("*.mp3")) path = Path(ffmpeg)
if not files: return path if path.is_file() else None
return None
return random.choice(files)
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 @dataclass
@@ -87,7 +95,12 @@ def _escape_srt_path(path: str) -> str:
return path.replace("\\", "/").replace(":", "\\:") 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: try:
result = subprocess.run( result = subprocess.run(
cmd, cmd,
@@ -104,12 +117,12 @@ def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
return True, "" 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。""" """Windows gdigrab 全桌面录制;非 Windows 返回 None。"""
if sys.platform != "win32": if sys.platform != "win32":
return None return None
return [ return _ffmpeg_cmd(
"ffmpeg", ffmpeg_exe,
"-y", "-y",
"-f", "-f",
"gdigrab", "gdigrab",
@@ -124,10 +137,11 @@ def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]:
"-pix_fmt", "-pix_fmt",
"yuv420p", "yuv420p",
capture_path, capture_path,
] )
def _compose_capture_mp4( def _compose_capture_mp4(
ffmpeg_exe: Path,
capture_path: Path, capture_path: Path,
srt_path: Path, srt_path: Path,
output_path: Path, output_path: Path,
@@ -147,8 +161,7 @@ def _compose_capture_mp4(
f"[0:v]{subtitle_filter}[vout];" f"[0:v]{subtitle_filter}[vout];"
f"[1:a]volume={music_volume},apad[aout]" f"[1:a]volume={music_volume},apad[aout]"
) )
cmd = [ cmd_tail = [
"ffmpeg",
"-y", "-y",
"-i", "-i",
str(capture_path), str(capture_path),
@@ -176,8 +189,7 @@ def _compose_capture_mp4(
str(output_path), str(output_path),
] ]
else: else:
cmd = [ cmd_tail = [
"ffmpeg",
"-y", "-y",
"-i", "-i",
str(capture_path), str(capture_path),
@@ -197,7 +209,7 @@ def _compose_capture_mp4(
"yuv420p", "yuv420p",
str(output_path), str(output_path),
] ]
return _run_ffmpeg(cmd) return _run_ffmpeg(ffmpeg_exe, cmd_tail)
class RpaVideoSession: class RpaVideoSession:
@@ -287,11 +299,13 @@ class RpaVideoSession:
def _start_ffmpeg_capture(self) -> None: def _start_ffmpeg_capture(self) -> None:
if not self.enabled: if not self.enabled:
return return
if not shutil.which("ffmpeg"):
ffmpeg_exe = _resolve_ffmpeg_exe()
if ffmpeg_exe is None:
self.warnings.append("ffmpeg_not_found") self.warnings.append("ffmpeg_not_found")
return return
cmd = _build_desktop_record_cmd(self._capture_path) cmd = _build_desktop_record_cmd(ffmpeg_exe, self._capture_path)
if cmd is None: if cmd is None:
self._append_warning( self._append_warning(
"ffmpeg_record_start_failed", "ffmpeg_record_start_failed",
@@ -391,14 +405,21 @@ class RpaVideoSession:
except Exception as exc: except Exception as exc:
self.warnings.append(f"subtitle_write_failed: {exc}") self.warnings.append(f"subtitle_write_failed: {exc}")
music = _resolve_music_file() music = _resolve_music_file(self.warnings)
if music is None: 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") 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) srt_file = Path(self._srt_path)
use_srt = srt_file.is_file() and srt_file.stat().st_size > 0 use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
ok, err = _compose_capture_mp4( ok, err = _compose_capture_mp4(
ffmpeg_exe,
capture, capture,
srt_file if use_srt else Path(self._srt_path), srt_file if use_srt else Path(self._srt_path),
Path(self._planned_output), Path(self._planned_output),
@@ -407,6 +428,7 @@ class RpaVideoSession:
if not ok and music is not None: if not ok and music is not None:
self.warnings.append(f"ffmpeg_with_music_failed: {err}") self.warnings.append(f"ffmpeg_with_music_failed: {err}")
ok, err = _compose_capture_mp4( ok, err = _compose_capture_mp4(
ffmpeg_exe,
capture, capture,
srt_file if use_srt else Path(self._srt_path), srt_file if use_srt else Path(self._srt_path),
Path(self._planned_output), Path(self._planned_output),
@@ -414,8 +436,8 @@ class RpaVideoSession:
) )
if not ok and not use_srt: if not ok and not use_srt:
ok, err = _run_ffmpeg( ok, err = _run_ffmpeg(
ffmpeg_exe,
[ [
"ffmpeg",
"-y", "-y",
"-i", "-i",
str(capture), str(capture),
@@ -428,7 +450,7 @@ class RpaVideoSession:
"-pix_fmt", "-pix_fmt",
"yuv420p", "yuv420p",
str(self._planned_output), str(self._planned_output),
] ],
) )
if not ok: if not ok:
self._append_warning("ffmpeg_compose_failed", err) self._append_warning("ffmpeg_compose_failed", err)

View File

@@ -0,0 +1,321 @@
"""共享 Runtime 诊断health 输出与 media-assets / ffmpeg 探测。"""
from __future__ import annotations
import importlib.metadata
import logging
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping
import jiangchang_skill_core
from .media_assets import probe_background_music, probe_media_assets, resolve_media_assets_root
from .runtime_env import platform_default_data_root
_SHARED_CORE_HINT = (
"jiangchang_skill_core is loaded from the skill tree; expected the shared "
"jiangchang-platform-kit installed in the host Python runtime."
)
@dataclass(frozen=True)
class RuntimeIssue:
code: str
message: str
severity: str = "warning"
@dataclass(frozen=True)
class RuntimeDiagnostics:
skill_slug: str
python_executable: str
platform_kit_version: str | None
platform_kit_min_version: str | None
platform_kit_version_ok: bool | None
jiangchang_skill_core_file: str | None
claw_data_root: str | None
jiangchang_data_root: str | None
resolved_data_root: str | None
media_assets_root: str | None
ffmpeg_available: bool
ffmpeg_path: str | None
background_music_mp3_count: int
background_music_usable_count: int
background_music_issue: str | None
background_music_sample_path: str | None
record_video_enabled: bool
issues: tuple[RuntimeIssue, ...]
@property
def has_fatal_issues(self) -> bool:
return any(issue.severity == "error" for issue in self.issues)
def issue_codes(self) -> list[str]:
return [issue.code for issue in self.issues]
def _env_dict(env: Mapping[str, str] | None) -> dict[str, str]:
return dict(env) if env is not None else dict(os.environ)
def _resolve_data_root(env: Mapping[str, str]) -> str:
root = (env.get("CLAW_DATA_ROOT") or env.get("JIANGCHANG_DATA_ROOT") or "").strip()
if root:
return root
return platform_default_data_root()
def _parse_version(version: str) -> tuple[int, ...]:
parts: list[int] = []
for piece in version.strip().split("."):
digits = ""
for ch in piece:
if ch.isdigit():
digits += ch
else:
break
if digits:
parts.append(int(digits))
return tuple(parts)
def version_ge(installed: str, required: str) -> bool:
"""Compare PEP 440-like versions without external deps (post release suffix ignored)."""
return _parse_version(installed) >= _parse_version(required)
def _path_under(parent: str | Path, child: str | Path) -> bool:
try:
parent_path = Path(parent).resolve()
child_path = Path(child).resolve()
if sys.platform == "win32":
parent_norm = os.path.normcase(str(parent_path))
child_norm = os.path.normcase(str(child_path))
common = os.path.commonpath([parent_norm, child_norm])
return common == parent_norm
return os.path.commonpath([str(parent_path), str(child_path)]) == str(parent_path)
except (OSError, ValueError):
return False
def is_jiangchang_skill_core_from_skill_tree(
*,
skill_root: str | Path,
core_file: str | Path | None = None,
) -> bool:
"""Return True when jiangchang_skill_core is loaded from inside skill_root."""
try:
root = Path(skill_root)
if not str(root).strip():
return False
core = Path(core_file) if core_file is not None else Path(jiangchang_skill_core.__file__)
return _path_under(root, core)
except (OSError, TypeError, ValueError):
return False
def _platform_kit_version() -> str | None:
try:
return importlib.metadata.version("jiangchang-platform-kit")
except importlib.metadata.PackageNotFoundError:
return None
def _bool_from_env(env: Mapping[str, str], key: str, default: bool = False) -> bool:
val = env.get(key)
if val is None or val == "":
return default
return str(val).strip().lower() in ("1", "true", "yes", "on")
def collect_runtime_diagnostics(
*,
skill_slug: str,
platform_kit_min_version: str | None = None,
skill_root: str | Path | None = None,
record_video: bool | None = None,
env: Mapping[str, str] | None = None,
) -> RuntimeDiagnostics:
env_map = _env_dict(env)
issues: list[RuntimeIssue] = []
claw_data_root = env_map.get("CLAW_DATA_ROOT")
jiangchang_data_root = env_map.get("JIANGCHANG_DATA_ROOT")
resolved_data_root = _resolve_data_root(env_map)
media_root_path = resolve_media_assets_root(resolved_data_root, env_map)
media_status = probe_media_assets(resolved_data_root, env_map)
ffmpeg_path_obj = media_status.ffmpeg_path
ffmpeg_ok = ffmpeg_path_obj is not None and ffmpeg_path_obj.is_file()
music_probe = probe_background_music(media_root_path, env=env_map)
mp3_count = int(music_probe["mp3_count"])
usable_count = int(music_probe["usable_count"])
music_issue = music_probe.get("issue")
music_sample = music_probe.get("sample_path")
if music_issue is None and not media_status.music_ready and media_status.warnings:
for warning in media_status.warnings:
if warning.startswith("background_music") or warning == "music_dir_missing":
music_issue = warning
break
platform_version = _platform_kit_version()
platform_ok: bool | None
if platform_kit_min_version is None:
platform_ok = None
elif platform_version is None:
platform_ok = False
else:
platform_ok = version_ge(platform_version, platform_kit_min_version)
if platform_version is None:
issues.append(
RuntimeIssue(
code="platform_kit_not_installed",
message="jiangchang-platform-kit package metadata not found",
severity="error",
)
)
elif platform_kit_min_version is not None and not platform_ok:
issues.append(
RuntimeIssue(
code="platform_kit_version_low",
message=(
f"jiangchang-platform-kit version {platform_version} "
f"is below required {platform_kit_min_version}"
),
severity="error",
)
)
core_file: str | None
try:
core_file = os.path.abspath(jiangchang_skill_core.__file__)
except (OSError, TypeError, ValueError):
core_file = None
if skill_root is not None and core_file is not None:
if is_jiangchang_skill_core_from_skill_tree(skill_root=skill_root, core_file=core_file):
issues.append(
RuntimeIssue(
code="jiangchang_skill_core_loaded_from_skill_tree",
message=_SHARED_CORE_HINT,
severity="warning",
)
)
if record_video is None:
record_video_enabled = _bool_from_env(env_map, "OPENCLAW_RECORD_VIDEO", False)
else:
record_video_enabled = record_video
if not ffmpeg_ok:
severity = "warning"
message = "ffmpeg not available"
if record_video_enabled:
message = "OPENCLAW_RECORD_VIDEO enabled but ffmpeg is not available"
issues.append(
RuntimeIssue(
code="ffmpeg_unavailable",
message=message,
severity=severity,
)
)
if record_video_enabled and music_issue:
issues.append(
RuntimeIssue(
code="background_music_unavailable",
message=f"background music unavailable: {music_issue}",
severity="warning",
)
)
return RuntimeDiagnostics(
skill_slug=skill_slug,
python_executable=sys.executable,
platform_kit_version=platform_version,
platform_kit_min_version=platform_kit_min_version,
platform_kit_version_ok=platform_ok,
jiangchang_skill_core_file=core_file,
claw_data_root=claw_data_root,
jiangchang_data_root=jiangchang_data_root,
resolved_data_root=resolved_data_root,
media_assets_root=str(media_root_path),
ffmpeg_available=ffmpeg_ok,
ffmpeg_path=str(ffmpeg_path_obj) if ffmpeg_path_obj else None,
background_music_mp3_count=mp3_count,
background_music_usable_count=usable_count,
background_music_issue=str(music_issue) if music_issue else None,
background_music_sample_path=str(music_sample) if music_sample else None,
record_video_enabled=record_video_enabled,
issues=tuple(issues),
)
def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]:
lines = [
f"skill_slug: {diagnostics.skill_slug}",
f"python_executable: {diagnostics.python_executable}",
f"platform_kit_version: {diagnostics.platform_kit_version or ''}",
f"platform_kit_min_version: {diagnostics.platform_kit_min_version or ''}",
f"platform_kit_version_ok: {diagnostics.platform_kit_version_ok}",
f"jiangchang_skill_core_file: {diagnostics.jiangchang_skill_core_file or ''}",
f"CLAW_DATA_ROOT: {diagnostics.claw_data_root or ''}",
f"JIANGCHANG_DATA_ROOT: {diagnostics.jiangchang_data_root or ''}",
f"resolved_data_root: {diagnostics.resolved_data_root or ''}",
f"media_assets_root: {diagnostics.media_assets_root or ''}",
f"ffmpeg_available: {diagnostics.ffmpeg_available}",
f"ffmpeg_path: {diagnostics.ffmpeg_path or ''}",
f"background_music_mp3_count: {diagnostics.background_music_mp3_count}",
f"background_music_usable_count: {diagnostics.background_music_usable_count}",
f"background_music_issue: {diagnostics.background_music_issue or ''}",
f"background_music_sample_path: {diagnostics.background_music_sample_path or ''}",
f"record_video_enabled: {diagnostics.record_video_enabled}",
]
for issue in diagnostics.issues:
lines.append(f"runtime_issue[{issue.severity}]: {issue.code}{issue.message}")
return lines
def runtime_diagnostics_dict(diagnostics: RuntimeDiagnostics) -> dict[str, Any]:
return {
"skill_slug": diagnostics.skill_slug,
"python_executable": diagnostics.python_executable,
"platform_kit_version": diagnostics.platform_kit_version,
"platform_kit_min_version": diagnostics.platform_kit_min_version,
"platform_kit_version_ok": diagnostics.platform_kit_version_ok,
"jiangchang_skill_core_file": diagnostics.jiangchang_skill_core_file,
"CLAW_DATA_ROOT": diagnostics.claw_data_root,
"JIANGCHANG_DATA_ROOT": diagnostics.jiangchang_data_root,
"resolved_data_root": diagnostics.resolved_data_root,
"media_assets_root": diagnostics.media_assets_root,
"ffmpeg_available": diagnostics.ffmpeg_available,
"ffmpeg_path": diagnostics.ffmpeg_path,
"background_music_mp3_count": diagnostics.background_music_mp3_count,
"background_music_usable_count": diagnostics.background_music_usable_count,
"background_music_issue": diagnostics.background_music_issue,
"background_music_sample_path": diagnostics.background_music_sample_path,
"record_video_enabled": diagnostics.record_video_enabled,
"runtime_issues": [
{"code": issue.code, "message": issue.message, "severity": issue.severity}
for issue in diagnostics.issues
],
}
def log_runtime_diagnostics(
diagnostics: RuntimeDiagnostics,
logger: logging.Logger | None = None,
level: int = logging.INFO,
) -> None:
log = logger or logging.getLogger("jiangchang_skill_core.runtime_diagnostics")
for line in format_runtime_health_lines(diagnostics):
log.log(level, line)

View File

@@ -34,14 +34,22 @@ def platform_default_data_root() -> str:
def get_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: if env:
return env return env
return platform_default_data_root() return platform_default_data_root()
def get_user_id() -> str: 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: def _looks_like_skills_root(path: str) -> bool:
@@ -55,6 +63,7 @@ def _looks_like_skills_root(path: str) -> bool:
"toutiao-publisher", "toutiao-publisher",
"logistics-tracker", "logistics-tracker",
"api-key-vault", "api-key-vault",
"receive-order",
): ):
if os.path.isdir(os.path.join(path, marker)): if os.path.isdir(os.path.join(path, marker)):
return True return True
@@ -130,10 +139,21 @@ def apply_cli_local_defaults() -> None:
enabled = v in ("1", "true", "yes", "on") enabled = v in ("1", "true", "yes", "on")
if not enabled: if not enabled:
return return
if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip(): if not (
os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root() os.getenv("CLAW_DATA_ROOT")
if not (os.getenv("JIANGCHANG_USER_ID") or "").strip(): 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["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": def find_chrome_executable() -> "str | None":
"""查找系统已安装的 Chrome 或 Edge 可执行文件路径。 """查找系统已安装的 Chrome 或 Edge 可执行文件路径。

View File

@@ -1,16 +1,18 @@
"""FFmpeg 合成:帧序列 + SRT 字幕 + 背景音乐 → MP4。""" """FFmpeg 合成:帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
from __future__ import annotations from __future__ import annotations
import random
import subprocess import subprocess
import sys import sys
from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from jiangchang_skill_core.media_assets import (
background_music_issue,
pick_background_music,
resolve_ffmpeg,
)
# 匠厂统一字幕样式 —— 所有 skill 录屏共用,做品牌一致性 # 匠厂统一字幕样式 —— 所有 skill 录屏共用,做品牌一致性
# FontSize 141080p 屏幕约占 1.3% 高度,演示视频留足画面呼吸感
# Outline 214 号字配 Outline 3 会显胖2 已足够在浅色背景上保持可读
# 修改请评估对所有 skill 录屏的影响。
_JIANGCHANG_SUBTITLE_STYLE = ( _JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14," "FontName=Arial,FontSize=14,"
"PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000," "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( def compose_video(
frames_dir: str, frames_dir: str,
fps: int, fps: int,
subtitle_path: str, subtitle_path: str,
music_dir: str,
output_path: str, output_path: str,
*,
music_volume: float = 0.15, music_volume: float = 0.15,
) -> str: allow_no_music: bool = True,
frames_dir = Path(frames_dir) media_assets_root: str | Path | None = None,
subtitle_path = Path(subtitle_path) media_assets_env: dict[str, str] | None = None,
output_path = Path(output_path) ) -> ComposeVideoResult:
output_path.parent.mkdir(parents=True, exist_ok=True) """合成 MP4ffmpeg 与背景音乐均走 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] = []
# 随机选一首背景音乐 assets_env = _media_assets_env(media_assets_root, media_assets_env)
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 滤镜内部语法) ffmpeg_exe = resolve_ffmpeg(env=assets_env)
srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:") 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 = ( subtitle_filter = (
f"subtitles='{srt_path_str}'" f"subtitles='{srt_esc}'"
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'" f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
) )
# 视频和音频都放进 filter_complex避免 -vf 与 -filter_complex 混用报错 if music_file and music_file.is_file():
filter_complex = ( filter_complex = (
f"[0:v]{subtitle_filter}[vout];" f"[0:v]{subtitle_filter}[vout];"
f"[1:a]volume={music_volume},apad[aout]" 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 = [ cmd_tail = [
"ffmpeg", "-y", "-y",
"-framerate", str(fps), "-framerate",
"-i", str(frames_dir / "frame_%06d.png"), str(fps),
"-i", str(music_file), "-i",
"-filter_complex", filter_complex, str(frames_dir_path / "frame_%06d.png"),
"-map", "[vout]", "-vf",
"-map", "[aout]", subtitle_filter,
"-c:v", "libx264", "-c:v",
"-preset", "fast", "libx264",
"-crf", "23", "-preset",
"-c:a", "aac", "fast",
"-b:a", "128k", "-crf",
"-shortest", "23",
"-pix_fmt", "yuv420p", "-pix_fmt",
str(output_path), "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 合成...") print(f"[screencast] 输出: {output_path_obj}")
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace") return ComposeVideoResult(str(output_path_obj.resolve()), warnings)
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

@@ -30,6 +30,8 @@ def _ensure_sdk_on_path() -> None:
_ensure_sdk_on_path() _ensure_sdk_on_path()
from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window # noqa: E402 from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window # noqa: E402
import warnings
from .composer import compose_video from .composer import compose_video
from .recorder import ScreenRecorder from .recorder import ScreenRecorder
from .subtitle import SubtitleEngine from .subtitle import SubtitleEngine
@@ -54,10 +56,11 @@ def run_screencast(
subtitle_script: [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout subtitle_script: [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout
pytest_args: 传递给 pytest 的参数列表 pytest_args: 传递给 pytest 的参数列表
output_dir: 最终 MP4 输出目录 output_dir: 最终 MP4 输出目录
media_assets_root: media-assets 仓库本地路径; media_assets_root: media-assets 根目录;不传时使用
不传时读 MEDIA_ASSETS_ROOT 环境变量, ``resolve_media_assets_root()`` 默认路径
再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets JIANGCHANG_DATA_ROOT/shared/media-assets 或 MEDIA_ASSETS_ROOT
music_subdir: music_assets_root 下音乐子目录,默认 "music" music_subdir: **已废弃**,保留仅为兼容旧调用;背景音乐由
``media_assets.pick_background_music()`` 统一解析
fps: 截帧帧率,默认 10 fps: 截帧帧率,默认 10
activate_window: 录屏开始前是否自动激活+最大化匠厂窗口(默认 True activate_window: 录屏开始前是否自动激活+最大化匠厂窗口(默认 True
Windows 走 ctypes非 Windows 走 jiangchang:// 协议) Windows 走 ctypes非 Windows 走 jiangchang:// 协议)
@@ -67,13 +70,14 @@ def run_screencast(
Returns: Returns:
输出 MP4 的绝对路径 输出 MP4 的绝对路径
""" """
if media_assets_root is None: if music_subdir != "music":
media_assets_root = os.environ.get( warnings.warn(
"MEDIA_ASSETS_ROOT", "music_subdir is deprecated and ignored; "
r"D:\OpenClaw\client-commons\media-assets", "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 = Path(output_dir)
output_dir_path.mkdir(parents=True, exist_ok=True) output_dir_path.mkdir(parents=True, exist_ok=True)
@@ -135,8 +139,8 @@ def run_screencast(
frames_dir=frames_dir, frames_dir=frames_dir,
fps=fps, fps=fps,
subtitle_path=str(subtitle_file), subtitle_path=str(subtitle_file),
music_dir=music_dir,
output_path=str(output_mp4), output_path=str(output_mp4),
media_assets_root=media_assets_root,
) )
return str(output_mp4) return str(output_mp4)

View File

@@ -81,6 +81,27 @@ class TestConfig(unittest.TestCase):
self.assertEqual(config.get_float("BAD", 9.9), 9.9) self.assertEqual(config.get_float("BAD", 9.9), 9.9)
self.assertEqual(config.get_int("BAD", 7), 7) 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__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@@ -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: def _make_complete_assets(root: Path, *, with_ffmpeg: bool = True, with_ffprobe: bool = True) -> None:
(root / "music" / "calm").mkdir(parents=True) (root / "music" / "calm").mkdir(parents=True)
(root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 1024)
(root / "fonts").mkdir(parents=True) (root / "fonts").mkdir(parents=True)
(root / "watermark").mkdir(parents=True) (root / "watermark").mkdir(parents=True)
bin_key = ma._platform_bin_key() bin_key = ma._platform_bin_key()
@@ -57,6 +58,56 @@ def _install_tools_from_manifest(root: Path, warnings: list[str]) -> bool:
return True return True
def _build_archive_zip(dest: Path, inner_name: str) -> None:
manifest = _sample_manifest()
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/calm/track.mp3", b"x" * 1024)
zf.writestr(f"{inner_name}/fonts/.keep", "")
zf.writestr(f"{inner_name}/watermark/.keep", "")
def test_default_bundle_url_is_release() -> None:
url = ma._media_assets_bundle_url()
assert url == ma.MEDIA_ASSETS_BUNDLE_URL
assert url.endswith("/releases/download/vlatest/media-assets.zip")
assert "/archive/main.zip" not in url
def test_bundle_url_env_override() -> None:
override = "https://example.com/media-assets.zip"
assert ma._media_assets_bundle_url(env={"MEDIA_ASSETS_BUNDLE_URL": override}) == override
def test_zip_download_uses_bundle_url_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
data_root = tmp_path / "data"
shared = data_root / "shared"
target = shared / "media-assets"
archive = tmp_path / "media-assets.zip"
override = "https://example.com/media-assets.zip"
_build_archive_zip(archive, "media-assets")
captured_urls: list[str] = []
def _fake_download(url: str, dest: Path) -> None:
captured_urls.append(url)
dest.write_bytes(archive.read_bytes())
monkeypatch.setattr(ma, "_download_zip", _fake_download)
monkeypatch.setattr(ma, "_find_git_executable", lambda: None)
monkeypatch.setattr(ma, "_try_download_ffmpeg_tools", _install_tools_from_manifest)
env = {
"JIANGCHANG_DATA_ROOT": str(data_root),
"MEDIA_ASSETS_BUNDLE_URL": override,
}
status = ma.ensure_media_assets(env=env)
assert captured_urls == [override]
assert target.is_dir()
assert status.ready is True
def test_media_assets_root_env_priority(tmp_path: Path) -> None: def test_media_assets_root_env_priority(tmp_path: Path) -> None:
custom = tmp_path / "custom-media" custom = tmp_path / "custom-media"
custom.mkdir() custom.mkdir()
@@ -117,9 +168,9 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.
root = tmp_path / "data" / "shared" / "media-assets" root = tmp_path / "data" / "shared" / "media-assets"
_make_complete_assets(root) _make_complete_assets(root)
music = root / "music" music = root / "music"
(music / "b-upbeat.mp3").write_bytes(b"mp3") (music / "b-upbeat.mp3").write_bytes(b"x" * 1024)
(music / "calm" / "a-calm.mp3").write_bytes(b"mp3") (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 1024)
(music / "z-last.wav").write_bytes(b"wav") (music / "z-last.wav").write_bytes(b"x" * 1024)
monkeypatch.setattr( monkeypatch.setattr(
ma, ma,
@@ -131,23 +182,18 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.
assert picked == music / "b-upbeat.mp3" assert picked == music / "b-upbeat.mp3"
def _build_archive_zip(dest: Path, inner_name: str) -> None: @pytest.mark.parametrize("inner_name", ["media-assets", "media-assets-main"])
manifest = _sample_manifest() def test_zip_wrapper_layout_detected(
with zipfile.ZipFile(dest, "w") as zf: tmp_path: Path,
zf.writestr(f"{inner_name}/README.md", "# media-assets\n") monkeypatch: pytest.MonkeyPatch,
zf.writestr(f"{inner_name}/manifest.json", json.dumps(manifest, ensure_ascii=False)) inner_name: str,
zf.writestr(f"{inner_name}/music/.keep", "") ) -> None:
zf.writestr(f"{inner_name}/fonts/.keep", "")
zf.writestr(f"{inner_name}/watermark/.keep", "")
def test_zip_wrapper_layout_detected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
data_root = tmp_path / "data" data_root = tmp_path / "data"
shared = data_root / "shared" shared = data_root / "shared"
target = shared / "media-assets" target = shared / "media-assets"
archive = tmp_path / "media-assets.zip" archive = tmp_path / "media-assets.zip"
_build_archive_zip(archive, "media-assets-main") _build_archive_zip(archive, inner_name)
def _fake_download(url: str, dest: Path) -> None: def _fake_download(url: str, dest: Path) -> None:
dest.write_bytes(archive.read_bytes()) dest.write_bytes(archive.read_bytes())
@@ -213,9 +259,138 @@ def test_windows_default_data_root() -> None:
assert root == Path(r"D:\jiangchang-data") / "shared" / "media-assets" 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_lfs_pointer_warning_in_probe_and_ensure(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 / "lfs-only.mp3").write_text(
"version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
)
monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
probed = ma.probe_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
assert "background_music_lfs_pointer_only" in probed.warnings
assert probed.music_ready is False
ensured = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
assert "background_music_lfs_pointer_only" in ensured.warnings
assert ensured.ready is False
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" * 1024)
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: def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
root = tmp_path / "data" / "shared" / "media-assets" 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" * 1024)
(root / "fonts").mkdir(parents=True) (root / "fonts").mkdir(parents=True)
(root / "watermark").mkdir(parents=True) (root / "watermark").mkdir(parents=True)
(root / "manifest.json").write_text( (root / "manifest.json").write_text(
@@ -231,3 +406,32 @@ def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatc
assert status.ready is True assert status.ready is True
assert status.ffmpeg_path is not None assert status.ffmpeg_path is not None
assert status.ffprobe_path is not None assert status.ffprobe_path is not None
def test_is_usable_audio_file_rejects_small_and_lfs(tmp_path: Path) -> None:
small = tmp_path / "small.mp3"
small.write_bytes(b"x" * 512)
assert ma.is_usable_audio_file(small) is False
ok = tmp_path / "ok.mp3"
ok.write_bytes(b"x" * 1024)
assert ma.is_usable_audio_file(ok) is True
lfs = tmp_path / "lfs.mp3"
lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n")
assert ma.is_usable_audio_file(lfs) is False
assert ma.is_git_lfs_pointer(lfs) is True
def test_probe_background_music_structure(tmp_path: Path) -> None:
root = tmp_path / "media-assets"
music = root / "music" / "calm"
music.mkdir(parents=True)
(music / "track.mp3").write_bytes(b"x" * 1024)
probe = ma.probe_background_music(root)
assert probe["music_root"] == str(root / "music")
assert probe["mp3_count"] == 1
assert probe["usable_count"] == 1
assert probe["issue"] is None
assert probe["sample_path"] == str(music / "track.mp3")

View File

@@ -0,0 +1,313 @@
# -*- coding: utf-8 -*-
"""runtime_diagnostics 公共 Runtime / media-assets 诊断测试。"""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
import pytest
import jiangchang_skill_core
from jiangchang_skill_core import media_assets as ma
from jiangchang_skill_core.runtime_diagnostics import (
RuntimeDiagnostics,
RuntimeIssue,
collect_runtime_diagnostics,
format_runtime_health_lines,
is_jiangchang_skill_core_from_skill_tree,
runtime_diagnostics_dict,
version_ge,
)
def test_version_ge_post_release() -> None:
assert version_ge("1.0.10.post23", "1.0.10") is True
assert version_ge("1.0.10", "1.0.10") is True
assert version_ge("1.0.9.post19", "1.0.10") is False
def test_is_jiangchang_skill_core_from_skill_tree(tmp_path: Path) -> None:
skill_root = tmp_path / "my-skill"
inside = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
inside.parent.mkdir(parents=True)
inside.write_text("# vendored\n", encoding="utf-8")
outside = tmp_path / "site-packages" / "jiangchang_skill_core" / "__init__.py"
outside.parent.mkdir(parents=True)
outside.write_text("# shared\n", encoding="utf-8")
assert is_jiangchang_skill_core_from_skill_tree(
skill_root=skill_root,
core_file=inside,
) is True
assert is_jiangchang_skill_core_from_skill_tree(
skill_root=skill_root,
core_file=outside,
) is False
def test_collect_runtime_diagnostics_json_serializable(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
data_root = tmp_path / "data"
env = {
"CLAW_DATA_ROOT": str(data_root),
"OPENCLAW_RECORD_VIDEO": "0",
}
media_root = data_root / "shared" / "media-assets"
(media_root / "music").mkdir(parents=True)
def _fake_probe_media_assets(data_root_arg=None, env_arg=None):
return ma.MediaAssetsStatus(
root=media_root,
exists=True,
ready=False,
source="local",
warnings=["ffmpeg_missing"],
ffmpeg_path=None,
music_dir=media_root / "music",
music_ready=False,
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
_fake_probe_media_assets,
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
lambda *a, **k: {
"music_root": str(media_root / "music"),
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": "background_music_dir_empty",
},
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
lambda: "1.0.10",
)
diag = collect_runtime_diagnostics(
skill_slug="my-skill",
platform_kit_min_version="1.0.9",
env=env,
)
payload = runtime_diagnostics_dict(diag)
json.dumps(payload)
assert payload["skill_slug"] == "my-skill"
assert payload["resolved_data_root"] == str(data_root)
assert payload["platform_kit_version_ok"] is True
def test_platform_kit_min_version_none_skips_version_issue(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
lambda *a, **k: ma.MediaAssetsStatus(
root=tmp_path,
exists=False,
ready=False,
source="local",
),
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
lambda *a, **k: {
"music_root": None,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": "music_dir_missing",
},
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
lambda: "1.0.8",
)
diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
assert diag.platform_kit_version_ok is None
assert "platform_kit_version_low" not in diag.issue_codes()
def test_platform_kit_missing_records_issue(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
lambda *a, **k: ma.MediaAssetsStatus(
root=tmp_path,
exists=False,
ready=False,
source="local",
),
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
lambda *a, **k: {
"music_root": None,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": None,
},
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
lambda: None,
)
diag = collect_runtime_diagnostics(
skill_slug="my-skill",
platform_kit_min_version="1.0.10",
env=env,
)
assert "platform_kit_not_installed" in diag.issue_codes()
def test_record_video_ffmpeg_and_music_warnings_not_fatal(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
media_root = tmp_path / "shared" / "media-assets"
env = {"CLAW_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"}
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
lambda *a, **k: ma.MediaAssetsStatus(
root=media_root,
exists=True,
ready=False,
source="local",
warnings=["ffmpeg_missing", "music_dir_missing"],
ffmpeg_path=None,
music_dir=None,
),
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
lambda *a, **k: {
"music_root": None,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": "music_dir_missing",
},
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
lambda: "1.0.10",
)
diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
codes = diag.issue_codes()
assert "ffmpeg_unavailable" in codes
assert "background_music_unavailable" in codes
assert diag.has_fatal_issues is False
ffmpeg_issue = next(i for i in diag.issues if i.code == "ffmpeg_unavailable")
assert ffmpeg_issue.severity == "warning"
def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None:
skill_root = tmp_path / "my-skill"
fake_core = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
fake_core.parent.mkdir(parents=True)
fake_core.write_text("# vendored\n", encoding="utf-8")
env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
with patch.object(jiangchang_skill_core, "__file__", str(fake_core)):
diag = collect_runtime_diagnostics(
skill_slug="my-skill",
skill_root=skill_root,
record_video=False,
env=env,
)
assert "jiangchang_skill_core_loaded_from_skill_tree" in diag.issue_codes()
def test_format_runtime_health_lines_core_fields(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
lambda *a, **k: ma.MediaAssetsStatus(
root=tmp_path,
exists=False,
ready=False,
source="local",
),
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
lambda *a, **k: {
"music_root": None,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": None,
},
)
monkeypatch.setattr(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
lambda: "1.0.10",
)
diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
text = "\n".join(format_runtime_health_lines(diag))
for marker in (
"python_executable:",
"platform_kit_version:",
"resolved_data_root:",
"media_assets_root:",
"ffmpeg_available:",
"background_music_mp3_count:",
):
assert marker in text
def test_background_music_lfs_and_size_rules(tmp_path: Path) -> None:
music = tmp_path / "music"
music.mkdir()
(music / "tiny.mp3").write_bytes(b"x" * 512)
(music / "lfs.mp3").write_text(
"version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
)
(music / "ok.mp3").write_bytes(b"x" * 1024)
probe = ma.probe_background_music(tmp_path)
assert probe["mp3_count"] == 3
assert probe["usable_count"] == 1
assert probe["sample_path"] == str(music / "ok.mp3")
def test_runtime_diagnostics_frozen_dataclass() -> None:
diag = RuntimeDiagnostics(
skill_slug="x",
python_executable="python",
platform_kit_version="1.0.10",
platform_kit_min_version=None,
platform_kit_version_ok=None,
jiangchang_skill_core_file=None,
claw_data_root=None,
jiangchang_data_root=None,
resolved_data_root="/tmp",
media_assets_root="/tmp/media",
ffmpeg_available=False,
ffmpeg_path=None,
background_music_mp3_count=0,
background_music_usable_count=0,
background_music_issue=None,
background_music_sample_path=None,
record_video_enabled=False,
issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
)
assert diag.has_fatal_issues is False
assert diag.issue_codes() == ["ffmpeg_unavailable"]

View File

@@ -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"

View File

@@ -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",
)

View File

@@ -7,8 +7,8 @@ import os
import sys import sys
import tempfile import tempfile
import unittest import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from jiangchang_skill_core import config from jiangchang_skill_core import config
from jiangchang_skill_core.rpa.video_session import RpaVideoSession from jiangchang_skill_core.rpa.video_session import RpaVideoSession
@@ -70,9 +70,10 @@ class TestRpaVideoSession(unittest.TestCase):
config.reset_cache() config.reset_cache()
@unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture") @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") @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 = MagicMock()
proc.poll.return_value = None proc.poll.return_value = None
proc.stdin = MagicMock() proc.stdin = MagicMock()
@@ -96,7 +97,7 @@ class TestRpaVideoSession(unittest.TestCase):
self.assertTrue(mock_popen.called) self.assertTrue(mock_popen.called)
cmd = mock_popen.call_args[0][0] 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("gdigrab", cmd)
self.assertIn("desktop", cmd) self.assertIn("desktop", cmd)
kwargs = mock_popen.call_args[1] kwargs = mock_popen.call_args[1]
@@ -105,8 +106,8 @@ class TestRpaVideoSession(unittest.TestCase):
os.environ.pop("OPENCLAW_RECORD_VIDEO", None) os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
config.reset_cache() config.reset_cache()
@patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value=None) @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe", return_value=None)
def test_ffmpeg_not_found_warning(self, _which: MagicMock) -> None: def test_ffmpeg_not_found_warning(self, _resolve: MagicMock) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1" os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
try: try:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp: