Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c9398451d0 | |||
| 2a542c0be6 | |||
| e59661f2d8 | |||
| 434a405e19 |
3
.github/workflows/publish.yml
vendored
3
.github/workflows/publish.yml
vendored
@@ -4,9 +4,6 @@ on:
|
|||||||
push:
|
push:
|
||||||
branches:
|
branches:
|
||||||
- main
|
- main
|
||||||
tags:
|
|
||||||
- "v*"
|
|
||||||
workflow_dispatch:
|
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
publish:
|
publish:
|
||||||
|
|||||||
52
README.md
52
README.md
@@ -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. 发版流程(维护者)
|
||||||
|
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
321
src/jiangchang_skill_core/runtime_diagnostics.py
Normal file
321
src/jiangchang_skill_core/runtime_diagnostics.py
Normal 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)
|
||||||
@@ -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 可执行文件路径。
|
||||||
|
|
||||||
|
|||||||
@@ -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 14:1080p 屏幕约占 1.3% 高度,演示视频留足画面呼吸感
|
|
||||||
# Outline 2:14 号字配 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)
|
"""合成 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] = []
|
||||||
|
|
||||||
# 随机选一首背景音乐
|
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)
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
313
tests/test_runtime_diagnostics.py
Normal file
313
tests/test_runtime_diagnostics.py
Normal 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"]
|
||||||
252
tests/test_screencast_composer.py
Normal file
252
tests/test_screencast_composer.py
Normal 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"
|
||||||
78
tests/test_screencast_runner.py
Normal file
78
tests/test_screencast_runner.py
Normal 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",
|
||||||
|
)
|
||||||
@@ -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:
|
||||||
|
|||||||
Reference in New Issue
Block a user