3 Commits

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

View File

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

View File

@@ -86,15 +86,39 @@ 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=/path/to/custom/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` ### 录屏合成(`screencast`

View File

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

View File

@@ -25,7 +25,10 @@ from .media_assets import (
MediaAssetsStatus, MediaAssetsStatus,
background_music_issue, 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_ffmpeg,
probe_media_assets, probe_media_assets,
resolve_ffmpeg, resolve_ffmpeg,
@@ -33,6 +36,15 @@ from .media_assets import (
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
@@ -46,8 +58,16 @@ __all__ = [
"EntitlementResult", "EntitlementResult",
"EntitlementServiceError", "EntitlementServiceError",
"MediaAssetsStatus", "MediaAssetsStatus",
"RuntimeDiagnostics",
"RuntimeIssue",
"background_music_issue", "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_ffmpeg",
"probe_media_assets", "probe_media_assets",
"pick_background_music", "pick_background_music",
@@ -55,6 +75,8 @@ __all__ = [
"resolve_ffprobe", "resolve_ffprobe",
"resolve_media_assets_root", "resolve_media_assets_root",
"resolve_music_dir", "resolve_music_dir",
"runtime_diagnostics_dict",
"version_ge",
"apply_cli_local_defaults", "apply_cli_local_defaults",
"attach_unified_file_handler", "attach_unified_file_handler",
"ensure_env_file", "ensure_env_file",

View File

@@ -16,11 +16,12 @@ 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")
@@ -28,17 +29,20 @@ _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" _GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1"
_MIN_AUDIO_BYTES = 128 _MIN_AUDIO_BYTES = 1024
def _is_git_lfs_pointer(path: Path) -> bool: def is_git_lfs_pointer(path: str | Path) -> bool:
try: try:
head = path.read_bytes()[:256] head = Path(path).read_bytes()[:256]
except OSError: except OSError:
return False return False
return head.startswith(_GIT_LFS_POINTER_PREFIX) return head.startswith(_GIT_LFS_POINTER_PREFIX)
_is_git_lfs_pointer = is_git_lfs_pointer
def _enumerate_audio_like(music_dir: Path) -> list[Path]: def _enumerate_audio_like(music_dir: Path) -> list[Path]:
return [ return [
path path
@@ -75,21 +79,25 @@ def _needs_music_content_repair(status: MediaAssetsStatus) -> bool:
) )
def _is_usable_audio_file(path: Path) -> bool: def is_usable_audio_file(path: str | Path, *, min_bytes: int = _MIN_AUDIO_BYTES) -> bool:
if not path.is_file(): p = Path(path)
if not p.is_file():
return False return False
if path.suffix.lower() not in _AUDIO_EXTENSIONS: if p.suffix.lower() not in _AUDIO_EXTENSIONS:
return False return False
if _is_git_lfs_pointer(path): if is_git_lfs_pointer(p):
return False return False
try: try:
if path.stat().st_size < _MIN_AUDIO_BYTES: if p.stat().st_size < min_bytes:
return False return False
except OSError: except OSError:
return False return False
return True return True
_is_usable_audio_file = is_usable_audio_file
@dataclass @dataclass
class MediaAssetsStatus: class MediaAssetsStatus:
root: Path root: Path
@@ -109,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,
@@ -474,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():
@@ -487,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)
@@ -565,10 +583,15 @@ def _needs_media_repo_content(status: MediaAssetsStatus) -> bool:
) )
def _maybe_repair_music_content(root: Path, status: MediaAssetsStatus) -> MediaAssetsStatus: def _maybe_repair_music_content(
root: Path,
status: MediaAssetsStatus,
*,
env: dict[str, str] | None = None,
) -> MediaAssetsStatus:
if not _needs_music_content_repair(status): if not _needs_music_content_repair(status):
return status return status
if _media_assets_root_overridden(None): if _media_assets_root_overridden(env):
return status return status
warnings: list[str] = [] warnings: list[str] = []
@@ -581,7 +604,7 @@ def _maybe_repair_music_content(root: Path, status: MediaAssetsStatus) -> MediaA
return _maybe_ensure_ffmpeg_tools(root, refreshed) return _maybe_ensure_ffmpeg_tools(root, refreshed)
shared = _shared_parent(root) shared = _shared_parent(root)
if _try_download_via_zip(shared, root, warnings): if _try_download_via_zip(shared, root, warnings, env=env):
refreshed = _inspect_media_assets(root, source="zip") refreshed = _inspect_media_assets(root, source="zip")
for warning in warnings: for warning in warnings:
if warning not in refreshed.warnings: if warning not in refreshed.warnings:
@@ -631,12 +654,13 @@ 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,
_maybe_repair_music_content( _maybe_repair_music_content(
root, root,
_inspect_media_assets(root, source="zip"), _inspect_media_assets(root, source="zip"),
env=env_map,
), ),
) )
@@ -650,6 +674,7 @@ def ensure_media_assets(
root, root,
source="git" if _is_git_repo(root) else "local", source="git" if _is_git_repo(root) else "local",
), ),
env=env_map,
), ),
) )
@@ -657,7 +682,7 @@ 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)
repaired = _maybe_repair_music_content(root, final) repaired = _maybe_repair_music_content(root, final, env=env_map)
for warning in warnings: for warning in warnings:
if warning not in repaired.warnings: if warning not in repaired.warnings:
repaired.warnings.append(warning) repaired.warnings.append(warning)
@@ -706,6 +731,46 @@ 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( def background_music_issue(
data_root: str | Path | None = None, data_root: str | Path | None = None,
env: dict[str, str] | None = None, env: dict[str, str] | None = None,

View File

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

View File

@@ -15,7 +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" * 256) (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()
@@ -58,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()
@@ -118,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"x" * 256) (music / "b-upbeat.mp3").write_bytes(b"x" * 1024)
(music / "calm" / "a-calm.mp3").write_bytes(b"x" * 256) (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 1024)
(music / "z-last.wav").write_bytes(b"x" * 256) (music / "z-last.wav").write_bytes(b"x" * 1024)
monkeypatch.setattr( monkeypatch.setattr(
ma, ma,
@@ -132,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/calm/track.mp3", b"x" * 256) ) -> 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())
@@ -226,12 +271,33 @@ def test_probe_media_assets_does_not_download(tmp_path: Path, monkeypatch: pytes
assert status.ffmpeg_path is None 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: def test_pick_background_music_skips_lfs_pointer(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
root = tmp_path / "data" / "shared" / "media-assets" root = tmp_path / "data" / "shared" / "media-assets"
_make_complete_assets(root, with_ffmpeg=True, with_ffprobe=True) _make_complete_assets(root, with_ffmpeg=True, with_ffprobe=True)
music = root / "music" music = root / "music"
(music / "calm" / "track.mp3").unlink() (music / "calm" / "track.mp3").unlink()
(music / "real.mp3").write_bytes(b"x" * 256) (music / "real.mp3").write_bytes(b"x" * 1024)
lfs = music / "lfs-only.mp3" lfs = music / "lfs-only.mp3"
lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n") lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n")
@@ -324,7 +390,7 @@ def test_video_finalize_without_music_fallback(tmp_path: Path, monkeypatch: pyte
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" / "calm").mkdir(parents=True) (root / "music" / "calm").mkdir(parents=True)
(root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 256) (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(
@@ -340,3 +406,32 @@ def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatc
assert status.ready is True assert status.ready is True
assert status.ffmpeg_path is not None assert status.ffmpeg_path is not None
assert status.ffprobe_path is not None assert status.ffprobe_path is not None
def test_is_usable_audio_file_rejects_small_and_lfs(tmp_path: Path) -> None:
small = tmp_path / "small.mp3"
small.write_bytes(b"x" * 512)
assert ma.is_usable_audio_file(small) is False
ok = tmp_path / "ok.mp3"
ok.write_bytes(b"x" * 1024)
assert ma.is_usable_audio_file(ok) is True
lfs = tmp_path / "lfs.mp3"
lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n")
assert ma.is_usable_audio_file(lfs) is False
assert ma.is_git_lfs_pointer(lfs) is True
def test_probe_background_music_structure(tmp_path: Path) -> None:
root = tmp_path / "media-assets"
music = root / "music" / "calm"
music.mkdir(parents=True)
(music / "track.mp3").write_bytes(b"x" * 1024)
probe = ma.probe_background_music(root)
assert probe["music_root"] == str(root / "music")
assert probe["mp3_count"] == 1
assert probe["usable_count"] == 1
assert probe["issue"] is None
assert probe["sample_path"] == str(music / "track.mp3")

View File

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