feat: resolve shared media assets

Co-authored-by: Cursor <cursoragent@cursor.com>

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

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]