From c9398451d0ccc45df13efb5f8f743ace276cbfa3 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 7 Jun 2026 09:06:41 +0800 Subject: [PATCH] Release v1.0.11: shared runtime diagnostics and media_assets probe helpers --- README.md | 23 ++ pyproject.toml | 2 +- src/jiangchang_skill_core/__init__.py | 22 ++ src/jiangchang_skill_core/media_assets.py | 65 +++- .../runtime_diagnostics.py | 321 ++++++++++++++++++ tests/test_media_assets.py | 43 ++- tests/test_runtime_diagnostics.py | 313 +++++++++++++++++ 7 files changed, 772 insertions(+), 17 deletions(-) create mode 100644 src/jiangchang_skill_core/runtime_diagnostics.py create mode 100644 tests/test_runtime_diagnostics.py diff --git a/README.md b/README.md index 5e547fe..7dcb6f6 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,29 @@ bundle 内含背景音乐、字体与水印；**不包含** ffmpeg 二进制。f - `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`**。 diff --git a/pyproject.toml b/pyproject.toml index 0f86230..ae0e2e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.0.10" +version = "1.0.11" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index 38143e8..52e6161 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -25,7 +25,10 @@ from .media_assets import ( MediaAssetsStatus, background_music_issue, ensure_media_assets, + is_git_lfs_pointer, + is_usable_audio_file, pick_background_music, + probe_background_music, probe_ffmpeg, probe_media_assets, resolve_ffmpeg, @@ -33,6 +36,15 @@ from .media_assets import ( resolve_media_assets_root, 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: from . import rpa @@ -46,8 +58,16 @@ __all__ = [ "EntitlementResult", "EntitlementServiceError", "MediaAssetsStatus", + "RuntimeDiagnostics", + "RuntimeIssue", "background_music_issue", + "collect_runtime_diagnostics", "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", @@ -55,6 +75,8 @@ __all__ = [ "resolve_ffprobe", "resolve_media_assets_root", "resolve_music_dir", + "runtime_diagnostics_dict", + "version_ge", "apply_cli_local_defaults", "attach_unified_file_handler", "ensure_env_file", diff --git a/src/jiangchang_skill_core/media_assets.py b/src/jiangchang_skill_core/media_assets.py index acaa1e4..4714774 100644 --- a/src/jiangchang_skill_core/media_assets.py +++ b/src/jiangchang_skill_core/media_assets.py @@ -16,7 +16,7 @@ import zipfile from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from typing import Optional +from typing import Any, Mapping, Optional MEDIA_ASSETS_BUNDLE_URL = ( "https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip" @@ -29,17 +29,20 @@ _NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang" _AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"} _GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1" -_MIN_AUDIO_BYTES = 128 +_MIN_AUDIO_BYTES = 1024 -def _is_git_lfs_pointer(path: Path) -> bool: +def is_git_lfs_pointer(path: str | Path) -> bool: try: - head = path.read_bytes()[:256] + 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 @@ -76,21 +79,25 @@ def _needs_music_content_repair(status: MediaAssetsStatus) -> bool: ) -def _is_usable_audio_file(path: Path) -> bool: - if not path.is_file(): +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 path.suffix.lower() not in _AUDIO_EXTENSIONS: + if p.suffix.lower() not in _AUDIO_EXTENSIONS: return False - if _is_git_lfs_pointer(path): + if is_git_lfs_pointer(p): return False try: - if path.stat().st_size < _MIN_AUDIO_BYTES: + if p.stat().st_size < min_bytes: return False except OSError: return False return True +_is_usable_audio_file = is_usable_audio_file + + @dataclass class MediaAssetsStatus: root: Path @@ -724,6 +731,46 @@ def resolve_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, diff --git a/src/jiangchang_skill_core/runtime_diagnostics.py b/src/jiangchang_skill_core/runtime_diagnostics.py new file mode 100644 index 0000000..620c3fe --- /dev/null +++ b/src/jiangchang_skill_core/runtime_diagnostics.py @@ -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) diff --git a/tests/test_media_assets.py b/tests/test_media_assets.py index c4d1842..875be8c 100644 --- a/tests/test_media_assets.py +++ b/tests/test_media_assets.py @@ -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: (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 / "watermark").mkdir(parents=True) bin_key = ma._platform_bin_key() @@ -63,7 +63,7 @@ def _build_archive_zip(dest: Path, inner_name: str) -> None: with zipfile.ZipFile(dest, "w") as zf: zf.writestr(f"{inner_name}/README.md", "# media-assets\n") zf.writestr(f"{inner_name}/manifest.json", json.dumps(manifest, ensure_ascii=False)) - zf.writestr(f"{inner_name}/music/calm/track.mp3", b"x" * 256) + 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", "") @@ -168,9 +168,9 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest. root = tmp_path / "data" / "shared" / "media-assets" _make_complete_assets(root) music = root / "music" - (music / "b-upbeat.mp3").write_bytes(b"x" * 256) - (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 256) - (music / "z-last.wav").write_bytes(b"x" * 256) + (music / "b-upbeat.mp3").write_bytes(b"x" * 1024) + (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 1024) + (music / "z-last.wav").write_bytes(b"x" * 1024) monkeypatch.setattr( ma, @@ -297,7 +297,7 @@ def test_pick_background_music_skips_lfs_pointer(tmp_path: Path, monkeypatch: py _make_complete_assets(root, with_ffmpeg=True, with_ffprobe=True) music = root / "music" (music / "calm" / "track.mp3").unlink() - (music / "real.mp3").write_bytes(b"x" * 256) + (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") @@ -390,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: root = tmp_path / "data" / "shared" / "media-assets" (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 / "watermark").mkdir(parents=True) (root / "manifest.json").write_text( @@ -406,3 +406,32 @@ def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatc assert status.ready is True assert status.ffmpeg_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") diff --git a/tests/test_runtime_diagnostics.py b/tests/test_runtime_diagnostics.py new file mode 100644 index 0000000..b5d35b9 --- /dev/null +++ b/tests/test_runtime_diagnostics.py @@ -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"]