diff --git a/README.md b/README.md index 0a19ea3..7e3df04 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ pip install jiangchang-platform-kit --index-url https://git.jc2009.com/api/packa ## 3. 包含的模块 -- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析等，供 Skill CLI / 服务进程使用。 +- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。 - **jiangchang_desktop_sdk**：匠厂桌面应用自动化客户端（基于 CDP + Playwright 同步 API）。**本包不声明 Playwright 依赖**；请在运行环境中自行安装 Playwright，以便 `import playwright` 可用。 ## 4. 快速开始 @@ -78,6 +78,24 @@ assert_skill_env_file(ctx, ["API_KEY"]) send_prompt_via_composer(page, "第一行\n第二行") ``` +## 共享媒体资源 + +`jiangchang_skill_core.media_assets` 提供共享媒体资源解析能力。默认会从： + +`{JIANGCHANG_DATA_ROOT}/shared/media-assets` + +读取背景音乐、字体、水印和 ffmpeg 工具。 + +如果目录不存在，会优先通过 ZIP 下载公开仓库： + +`https://git.jc2009.com/client-commons/media-assets/archive/main.zip` + +这不依赖用户电脑安装 Git。 + +可通过环境变量覆盖： + +`MEDIA_ASSETS_ROOT=D:\OpenClaw\client-commons\media-assets` + ## 5. 发版流程（维护者） ### main 分支自动预发布 diff --git a/pyproject.toml b/pyproject.toml index 59ac1ba..9fe77ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "0.1.0" +version = "0.1.1" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_desktop_sdk/__init__.py b/src/jiangchang_desktop_sdk/__init__.py index 4670a61..2a452e1 100644 --- a/src/jiangchang_desktop_sdk/__init__.py +++ b/src/jiangchang_desktop_sdk/__init__.py @@ -23,7 +23,7 @@ __all__ = [ "LaunchError", "GatewayDownError", ] -__version__ = "0.1.0" +__version__ = "0.1.1" def __getattr__(name: str): diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index df9c549..30f19b7 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -21,6 +21,15 @@ from .unified_logging import ( setup_skill_logging, subprocess_env_with_trace, ) +from .media_assets import ( + MediaAssetsStatus, + ensure_media_assets, + pick_background_music, + resolve_ffmpeg, + resolve_ffprobe, + resolve_media_assets_root, + resolve_music_dir, +) try: from . import rpa @@ -33,6 +42,13 @@ __all__ = [ "EntitlementError", "EntitlementResult", "EntitlementServiceError", + "MediaAssetsStatus", + "ensure_media_assets", + "pick_background_music", + "resolve_ffmpeg", + "resolve_ffprobe", + "resolve_media_assets_root", + "resolve_music_dir", "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 new file mode 100644 index 0000000..43fe403 --- /dev/null +++ b/src/jiangchang_skill_core/media_assets.py @@ -0,0 +1,432 @@ +"""共享媒体资源解析：背景音乐、字体、水印与 ffmpeg 工具。""" + +from __future__ import annotations + +import os +import platform +import shutil +import subprocess +import sys +import tempfile +import urllib.error +import urllib.request +import zipfile +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Optional + +MEDIA_ASSETS_ZIP_URL = ( + "https://git.jc2009.com/client-commons/media-assets/archive/main.zip" +) +MEDIA_ASSETS_GIT_URL = "https://git.jc2009.com/client-commons/media-assets" + +_WIN_DEFAULT_DATA_ROOT = Path(r"D:\jiangchang-data") +_NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang" + +_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"} + + +@dataclass +class MediaAssetsStatus: + root: Path + exists: bool + ready: bool + source: str + warnings: list[str] = field(default_factory=list) + ffmpeg_path: Optional[Path] = None + ffprobe_path: Optional[Path] = None + music_dir: Optional[Path] = None + fonts_dir: Optional[Path] = None + watermark_dir: Optional[Path] = None + + +def _env_dict(env: dict[str, str] | None) -> dict[str, str]: + return env if env is not None else dict(os.environ) + + +def _resolve_jiangchang_data_root( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Path: + env_map = _env_dict(env) + from_env = (env_map.get("JIANGCHANG_DATA_ROOT") or "").strip() + if from_env: + return Path(from_env) + if data_root is not None: + return Path(data_root) + if sys.platform == "win32": + return _WIN_DEFAULT_DATA_ROOT + return _NON_WIN_DEFAULT_DATA_ROOT + + +def resolve_media_assets_root( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Path: + env_map = _env_dict(env) + media_root = (env_map.get("MEDIA_ASSETS_ROOT") or "").strip() + if media_root: + return Path(media_root) + data = _resolve_jiangchang_data_root(data_root, env) + return data / "shared" / "media-assets" + + +def _platform_bin_key() -> str: + system = platform.system().lower() + machine = platform.machine().lower() + + if system == "windows": + if machine in {"amd64", "x86_64"}: + return "win32-x64" + if machine in {"arm64", "aarch64"}: + return "win32-arm64" + elif system == "darwin": + if machine in {"arm64", "aarch64"}: + return "darwin-arm64" + if machine in {"x86_64", "amd64"}: + return "darwin-x64" + elif system == "linux": + if machine in {"x86_64", "amd64"}: + return "linux-x64" + if machine in {"arm64", "aarch64"}: + return "linux-arm64" + + return f"{system}-{machine}" + + +def _tool_filename(name: str) -> str: + if platform.system().lower() == "windows": + return f"{name}.exe" + return name + + +def _platform_tool_paths(root: Path) -> tuple[Path, Path]: + bin_key = _platform_bin_key() + bin_dir = root / "bin" / bin_key + return ( + bin_dir / _tool_filename("ffmpeg"), + bin_dir / _tool_filename("ffprobe"), + ) + + +def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus: + warnings: list[str] = [] + exists = root.exists() + music_dir = root / "music" + fonts_dir = root / "fonts" + watermark_dir = root / "watermark" + ffmpeg_path, ffprobe_path = _platform_tool_paths(root) + + music_exists = music_dir.is_dir() + fonts_exists = fonts_dir.is_dir() + watermark_exists = watermark_dir.is_dir() + ffmpeg_exists = ffmpeg_path.is_file() + ffprobe_exists = ffprobe_path.is_file() + + if not music_exists: + warnings.append("music_dir_missing") + if not fonts_exists: + warnings.append("fonts_dir_missing") + if not watermark_exists: + warnings.append("watermark_dir_missing") + if not ffmpeg_exists: + warnings.append("ffmpeg_missing") + if not ffprobe_exists: + warnings.append("ffprobe_missing") + + ready = ( + exists + and music_exists + and fonts_exists + and watermark_exists + and ffmpeg_exists + and ffprobe_exists + ) + + return MediaAssetsStatus( + root=root, + exists=exists, + ready=ready, + source=source, + warnings=warnings, + ffmpeg_path=ffmpeg_path if ffmpeg_exists else None, + ffprobe_path=ffprobe_path if ffprobe_exists else None, + music_dir=music_dir if music_exists else None, + fonts_dir=fonts_dir if fonts_exists else None, + watermark_dir=watermark_dir if watermark_exists else None, + ) + + +def _shared_parent(root: Path) -> Path: + return root.parent + + +def _download_dir(shared: Path) -> Path: + return shared / ".media-assets-download" + + +def _find_media_assets_root_in_tree(extracted: Path) -> Path | None: + if not extracted.is_dir(): + return None + + def _looks_like_assets(path: Path) -> bool: + if (path / "music").is_dir(): + return True + if (path / "fonts").is_dir(): + return True + if (path / "watermark").is_dir(): + return True + if (path / "README.md").is_file(): + return True + return False + + if _looks_like_assets(extracted): + return extracted + + children = [p for p in extracted.iterdir() if p.is_dir()] + for child in children: + if _looks_like_assets(child): + return child + + for path in extracted.rglob("*"): + if path.is_dir() and _looks_like_assets(path): + return path + return None + + +def _download_zip(url: str, dest: Path) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + with urllib.request.urlopen(url, timeout=120) as response: + dest.write_bytes(response.read()) + + +def _extract_zip(zip_path: Path, extract_to: Path) -> None: + extract_to.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(zip_path, "r") as zf: + zf.extractall(extract_to) + + +def _find_git_executable() -> Path | None: + git = shutil.which("git") + return Path(git) if git else None + + +def _git_clone(url: str, target: Path) -> None: + target.parent.mkdir(parents=True, exist_ok=True) + subprocess.run( + ["git", "clone", url, str(target)], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def _git_pull(target: Path) -> None: + subprocess.run( + ["git", "pull"], + cwd=str(target), + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def _is_git_repo(path: Path) -> bool: + return (path / ".git").exists() + + +def _install_media_assets(source: Path, target: Path) -> None: + target.parent.mkdir(parents=True, exist_ok=True) + if not target.exists(): + shutil.move(str(source), str(target)) + return + + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + backup = target.parent / f"media-assets.backup-{timestamp}" + shutil.move(str(target), str(backup)) + try: + shutil.move(str(source), str(target)) + except Exception: + if backup.exists() and not target.exists(): + shutil.move(str(backup), str(target)) + raise + else: + shutil.rmtree(backup, ignore_errors=True) + + +def _cleanup_download_artifacts(download_root: Path) -> None: + shutil.rmtree(download_root, ignore_errors=True) + + +def _try_download_via_zip( + shared: Path, + target: Path, + warnings: list[str], +) -> bool: + download_root = _download_dir(shared) + download_root.mkdir(parents=True, exist_ok=True) + zip_path = download_root / "media-assets.zip" + extracted = download_root / "extracted" + staging = download_root / "staging" + + try: + if staging.exists(): + shutil.rmtree(staging, ignore_errors=True) + if extracted.exists(): + shutil.rmtree(extracted, ignore_errors=True) + + _download_zip(MEDIA_ASSETS_ZIP_URL, zip_path) + _extract_zip(zip_path, extracted) + + found = _find_media_assets_root_in_tree(extracted) + if found is None: + warnings.append("media_assets_zip_invalid_layout") + return False + + if staging.exists(): + shutil.rmtree(staging, ignore_errors=True) + shutil.copytree(found, staging) + _install_media_assets(staging, target) + _cleanup_download_artifacts(download_root) + return True + except urllib.error.URLError: + warnings.append("media_assets_zip_download_failed") + return False + except zipfile.BadZipFile: + warnings.append("media_assets_zip_extract_failed") + return False + except OSError: + warnings.append("media_assets_zip_extract_failed") + return False + except Exception: + warnings.append("media_assets_zip_extract_failed") + return False + + +def _try_download_via_git( + target: Path, + warnings: list[str], + *, + clone: bool, +) -> bool: + git_exe = _find_git_executable() + if git_exe is None: + warnings.append("git_not_available") + return False + + try: + if clone: + if target.exists(): + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + partial = target.parent / f"media-assets.git-staging-{timestamp}" + if partial.exists(): + shutil.rmtree(partial, ignore_errors=True) + _git_clone(MEDIA_ASSETS_GIT_URL, partial) + _install_media_assets(partial, target) + else: + _git_clone(MEDIA_ASSETS_GIT_URL, target) + else: + _git_pull(target) + return True + except (subprocess.CalledProcessError, OSError): + if clone: + warnings.append("media_assets_git_clone_failed") + return False + + +def _media_assets_root_overridden(env: dict[str, str] | None) -> bool: + env_map = _env_dict(env) + return bool((env_map.get("MEDIA_ASSETS_ROOT") or "").strip()) + + +def ensure_media_assets( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, + *, + update: bool = False, +) -> MediaAssetsStatus: + env_map = _env_dict(env) + root = resolve_media_assets_root(data_root, env_map) + overridden = _media_assets_root_overridden(env_map) + + status = _inspect_media_assets(root, source="local") + if overridden: + return status + + if status.ready and not update: + return status + + if status.ready and update: + if _is_git_repo(root): + warnings: list[str] = [] + if _try_download_via_git(root, warnings, clone=False): + return _inspect_media_assets(root, source="git") + status = _inspect_media_assets(root, source="local") + status.warnings.extend(warnings) + return status + + shared = _shared_parent(root) + shared.mkdir(parents=True, exist_ok=True) + warnings: list[str] = [] + + if _try_download_via_zip(shared, root, warnings): + return _inspect_media_assets(root, source="zip") + + clone_needed = not root.exists() or not status.ready or update + if _try_download_via_git(root, warnings, clone=clone_needed): + return _inspect_media_assets( + root, + source="git" if _is_git_repo(root) else "local", + ) + + final = _inspect_media_assets(root, source="local") + for warning in warnings: + if warning not in final.warnings: + final.warnings.append(warning) + return final + + +def resolve_ffmpeg( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[Path]: + status = ensure_media_assets(data_root, env) + return status.ffmpeg_path + + +def resolve_ffprobe( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[Path]: + status = ensure_media_assets(data_root, env) + return status.ffprobe_path + + +def resolve_music_dir( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[Path]: + status = ensure_media_assets(data_root, env) + return status.music_dir + + +def pick_background_music( + data_root: str | Path | None = None, + env: dict[str, str] | None = None, +) -> Optional[Path]: + status = ensure_media_assets(data_root, env) + music_dir = status.music_dir + if music_dir is None or not music_dir.is_dir(): + return None + + candidates: list[Path] = [] + for path in music_dir.rglob("*"): + if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS: + candidates.append(path) + + if not candidates: + return None + + candidates.sort(key=lambda p: str(p).lower()) + return candidates[0] diff --git a/tests/test_media_assets.py b/tests/test_media_assets.py new file mode 100644 index 0000000..e8d3f0b --- /dev/null +++ b/tests/test_media_assets.py @@ -0,0 +1,187 @@ +# -*- coding: utf-8 -*- +"""media_assets 共享媒体资源解析测试。""" +from __future__ import annotations + +import platform +import zipfile +from pathlib import Path +from unittest.mock import patch + +import pytest + +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 / "fonts").mkdir(parents=True) + (root / "watermark").mkdir(parents=True) + bin_key = ma._platform_bin_key() + bin_dir = root / "bin" / bin_key + bin_dir.mkdir(parents=True) + if with_ffmpeg: + (bin_dir / ma._tool_filename("ffmpeg")).write_bytes(b"ffmpeg") + if with_ffprobe: + (bin_dir / ma._tool_filename("ffprobe")).write_bytes(b"ffprobe") + + +def test_media_assets_root_env_priority(tmp_path: Path) -> None: + custom = tmp_path / "custom-media" + custom.mkdir() + env = { + "MEDIA_ASSETS_ROOT": str(custom), + "JIANGCHANG_DATA_ROOT": str(tmp_path / "data"), + } + assert ma.resolve_media_assets_root(env=env) == custom + + +def test_default_media_assets_path(tmp_path: Path) -> None: + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")} + expected = tmp_path / "data" / "shared" / "media-assets" + assert ma.resolve_media_assets_root(env=env) == expected + + +def test_complete_assets_skips_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root) + + 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.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert status.ready is True + assert status.warnings == [] + assert status.source == "local" + + +def test_missing_ffmpeg_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root, with_ffmpeg=False) + + monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False) + monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False) + + status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert status.ready is False + assert "ffmpeg_missing" in status.warnings + + +def test_missing_ffprobe_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root, with_ffprobe=False) + + monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False) + monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False) + + status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert status.ready is False + assert "ffprobe_missing" in status.warnings + + +def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "data" / "shared" / "media-assets" + _make_complete_assets(root) + music = root / "music" + (music / "b-upbeat.mp3").write_bytes(b"mp3") + (music / "calm" / "a-calm.mp3").write_bytes(b"mp3") + (music / "z-last.wav").write_bytes(b"wav") + + 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 / "b-upbeat.mp3" + + +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}/music/.keep", "") + zf.writestr(f"{inner_name}/fonts/.keep", "") + zf.writestr(f"{inner_name}/watermark/.keep", "") + bin_key = ma._platform_bin_key() + zf.writestr( + f"{inner_name}/bin/{bin_key}/{ma._tool_filename('ffmpeg')}", + b"ffmpeg", + ) + zf.writestr( + f"{inner_name}/bin/{bin_key}/{ma._tool_filename('ffprobe')}", + b"ffprobe", + ) + + +def test_zip_wrapper_layout_detected(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" + + _build_archive_zip(archive, "media-assets-main") + + def _fake_download(url: str, dest: Path) -> None: + dest.write_bytes(archive.read_bytes()) + + monkeypatch.setattr(ma, "_download_zip", _fake_download) + monkeypatch.setattr(ma, "_find_git_executable", lambda: None) + + status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(data_root)}) + assert target.is_dir() + assert status.ready is True + assert status.source == "zip" + assert "media_assets_zip_invalid_layout" not in status.warnings + + +def test_zip_download_failure_returns_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import urllib.error + + def _raise_url_error(url: str, dest: Path) -> None: + raise urllib.error.URLError("network down") + + monkeypatch.setattr(ma, "_download_zip", _raise_url_error) + monkeypatch.setattr(ma, "_find_git_executable", lambda: None) + + status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert status.ready is False + assert "media_assets_zip_download_failed" in status.warnings + assert "git_not_available" in status.warnings + + +def test_git_not_available_when_zip_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import urllib.error + + monkeypatch.setattr( + ma, + "_download_zip", + lambda url, dest: (_ for _ in ()).throw(urllib.error.URLError("fail")), + ) + monkeypatch.setattr(ma, "_find_git_executable", lambda: None) + + status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}) + assert "git_not_available" in status.warnings + + +def test_media_assets_root_override_no_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + custom = tmp_path / "override" + custom.mkdir() + + def _fail_download(*args, **kwargs): + raise AssertionError("download should not be called when MEDIA_ASSETS_ROOT is set") + + monkeypatch.setattr(ma, "_try_download_via_zip", _fail_download) + + status = ma.ensure_media_assets(env={"MEDIA_ASSETS_ROOT": str(custom)}) + assert status.root == custom + assert status.exists is True + + +@pytest.mark.skipif(platform.system().lower() != "windows", reason="Windows default path check") +def test_windows_default_data_root() -> None: + with patch.dict("os.environ", {}, clear=True): + root = ma.resolve_media_assets_root() + assert root == Path(r"D:\jiangchang-data") / "shared" / "media-assets"