fix: restore platform kit version to 1.0.9

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

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,17 +78,37 @@ 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 分支自动预发布
 
-每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **唯一** 的 post 版本（基于 `pyproject.toml` 中的基础版本，例如 `0.1.0.post42`）。版本号不会写回仓库，仅在 CI 构建前临时改写 `pyproject.toml` 与 `jiangchang_desktop_sdk.__version__`。
+每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **post 预发布版本**（仅供 CI 流水线内部使用，**不是**正式版本线）。
+
+正式版本线使用 **1.0.x** semver，通过 git tag 发布。
 
 ```bash
 git push origin main
 ```
 
-测试人员升级安装：
+测试人员升级安装（始终安装当前正式 tag 对应的版本，例如 `1.0.9`）：
 
 ```bash
 python -m pip install --upgrade \
@@ -99,11 +119,11 @@ python -m pip install --upgrade \
 
 ### 正式版本（tag）
 
-打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v0.1.1` → PyPI 包 `0.1.1`）：
+打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v1.0.9` → PyPI 包 `1.0.9`）：
 
 ```bash
-git tag v0.1.1
+git tag v1.0.9
-git push origin v0.1.1
+git push origin v1.0.9
 ```
 
 ### 安装后验证

pyproject.toml

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

src/jiangchang_desktop_sdk/__init__.py

@@ -23,7 +23,7 @@ __all__ = [
     "LaunchError",
     "GatewayDownError",
 ]
-__version__ = "0.1.0"
+__version__ = "1.0.9"
 
 
 def __getattr__(name: str):

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,595 @@
+"""共享媒体资源解析：背景音乐、字体、水印与 ffmpeg 工具。"""
+
+from __future__ import annotations
+
+import fnmatch
+import json
+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 _read_manifest(root: Path) -> dict | None:
+    manifest_path = root / "manifest.json"
+    if not manifest_path.is_file():
+        return None
+    try:
+        return json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def _ffmpeg_platform_spec(manifest: dict) -> dict | None:
+    tools = manifest.get("tools")
+    if not isinstance(tools, dict):
+        return None
+    ffmpeg = tools.get("ffmpeg")
+    if not isinstance(ffmpeg, dict):
+        return None
+    spec = ffmpeg.get(_platform_bin_key())
+    if isinstance(spec, dict):
+        return spec
+    return None
+
+
+def _manifest_tool_urls(spec: dict) -> list[tuple[str, str]]:
+    urls: list[tuple[str, str]] = []
+    raw_urls = spec.get("urls")
+    if isinstance(raw_urls, list):
+        for item in raw_urls:
+            if isinstance(item, dict):
+                name = str(item.get("name") or "source")
+                url = str(item.get("url") or "").strip()
+                if url:
+                    urls.append((name, url))
+    legacy = str(spec.get("url") or "").strip()
+    if legacy:
+        urls.append(("legacy", legacy))
+    return urls
+
+
+def _match_archive_member(member_name: str, pattern: str) -> bool:
+    normalized = member_name.replace("\\", "/").lstrip("/")
+    return fnmatch.fnmatch(normalized, pattern)
+
+
+def _extract_tool_members(
+    zip_path: Path,
+    archive_paths: dict[str, str],
+) -> dict[str, bytes]:
+    found: dict[str, bytes] = {}
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        for tool_name, pattern in archive_paths.items():
+            if not isinstance(pattern, str) or not pattern:
+                continue
+            for info in zf.infolist():
+                if info.is_dir():
+                    continue
+                if _match_archive_member(info.filename, pattern):
+                    found[tool_name] = zf.read(info)
+                    break
+    return found
+
+
+def _try_download_ffmpeg_tools(root: Path, warnings: list[str]) -> bool:
+    manifest = _read_manifest(root)
+    if manifest is None:
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    spec = _ffmpeg_platform_spec(manifest)
+    if spec is None:
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    archive_paths = spec.get("archive_paths")
+    if not isinstance(archive_paths, dict):
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    urls = _manifest_tool_urls(spec)
+    if not urls:
+        warnings.append("ffmpeg_tool_download_failed")
+        return False
+
+    ffmpeg_path, ffprobe_path = _platform_tool_paths(root)
+    bin_dir = ffmpeg_path.parent
+    download_root = _download_dir(_shared_parent(root)) / "ffmpeg-tools"
+    download_root.mkdir(parents=True, exist_ok=True)
+
+    for source_name, url in urls:
+        zip_path = download_root / f"{source_name}.zip"
+        try:
+            if zip_path.exists():
+                zip_path.unlink()
+            _download_zip(url, zip_path)
+            members = _extract_tool_members(zip_path, archive_paths)
+            ffmpeg_bytes = members.get("ffmpeg")
+            ffprobe_bytes = members.get("ffprobe")
+            if not ffmpeg_bytes or not ffprobe_bytes:
+                continue
+            bin_dir.mkdir(parents=True, exist_ok=True)
+            ffmpeg_path.write_bytes(ffmpeg_bytes)
+            ffprobe_path.write_bytes(ffprobe_bytes)
+            shutil.rmtree(download_root, ignore_errors=True)
+            return True
+        except urllib.error.URLError:
+            continue
+        except (zipfile.BadZipFile, OSError, KeyError):
+            continue
+        except Exception:
+            continue
+
+    warnings.append("ffmpeg_tool_download_failed")
+    return False
+
+
+def _maybe_ensure_ffmpeg_tools(root: Path, status: MediaAssetsStatus) -> MediaAssetsStatus:
+    if status.ready:
+        return status
+    if not status.exists:
+        return status
+    if "ffmpeg_missing" not in status.warnings and "ffprobe_missing" not in status.warnings:
+        return status
+
+    warnings = list(status.warnings)
+    if _try_download_ffmpeg_tools(root, warnings):
+        refreshed = _inspect_media_assets(root, source=status.source)
+        for warning in warnings:
+            if warning not in refreshed.warnings:
+                refreshed.warnings.append(warning)
+        return refreshed
+
+    refreshed = _inspect_media_assets(root, source=status.source)
+    for warning in warnings:
+        if warning not in refreshed.warnings:
+            refreshed.warnings.append(warning)
+    return refreshed
+
+
+def _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 _needs_media_repo_content(status: MediaAssetsStatus) -> bool:
+    return any(
+        warning in status.warnings
+        for warning in (
+            "music_dir_missing",
+            "fonts_dir_missing",
+            "watermark_dir_missing",
+        )
+    )
+
+
+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 _maybe_ensure_ffmpeg_tools(root, status)
+
+    if status.exists and not status.ready and not update and not _needs_media_repo_content(status):
+        return _maybe_ensure_ffmpeg_tools(root, 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 _maybe_ensure_ffmpeg_tools(
+                    root,
+                    _inspect_media_assets(root, source="git"),
+                )
+            status = _inspect_media_assets(root, source="local")
+            status.warnings.extend(warnings)
+            return _maybe_ensure_ffmpeg_tools(root, status)
+
+    shared = _shared_parent(root)
+    shared.mkdir(parents=True, exist_ok=True)
+    warnings: list[str] = []
+
+    if _try_download_via_zip(shared, root, warnings):
+        return _maybe_ensure_ffmpeg_tools(
+            root,
+            _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 _maybe_ensure_ffmpeg_tools(
+            root,
+            _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 _maybe_ensure_ffmpeg_tools(root, 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]

src/jiangchang_skill_core/rpa/video_session.py

@@ -1,7 +1,4 @@
-"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。
+"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
-
-Playwright 仅用于浏览器自动化，不负责录屏。
-"""
 
 from __future__ import annotations
 
@@ -15,7 +12,7 @@ import time
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, IO, List, Optional
 
 from .. import config
 
@@ -222,8 +219,6 @@ class RpaVideoSession:
         self.warnings: List[str] = []
         self.output_video_path: Optional[str] = None
         self.capture_path: Optional[str] = None
-        # 已废弃：勿用于 Playwright；始终为 None。
-        self.record_video_dir: Optional[str] = None
         self.artifacts: Dict[str, Any] = {}
 
         ts = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -236,6 +231,7 @@ class RpaVideoSession:
         )
         self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
         self._logs_dir = os.path.join(self._artifact_root, "logs")
+        self._ffmpeg_record_log_path = os.path.join(self._logs_dir, "ffmpeg-record.log")
         self._capture_path = os.path.join(self._artifact_root, "capture.mp4")
         self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt")
         self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
@@ -243,6 +239,7 @@ class RpaVideoSession:
         self._steps: List[_StepEntry] = []
         self._t0: Optional[float] = None
         self._ffmpeg_proc: Optional[subprocess.Popen] = None
+        self._ffmpeg_record_log_file: Optional[IO[str]] = None
 
         if self.enabled:
             os.makedirs(self._videos_dir, exist_ok=True)
@@ -250,6 +247,21 @@ class RpaVideoSession:
             os.makedirs(self._logs_dir, exist_ok=True)
             os.makedirs(self._artifact_root, exist_ok=True)
 
+    def _record_log_hint(self) -> str:
+        if self.enabled and self._ffmpeg_record_log_path:
+            return f"; record_log={self._ffmpeg_record_log_path}"
+        return ""
+
+    def _append_warning(self, code: str, detail: str = "") -> None:
+        msg = code if not detail else f"{code}: {detail}"
+        if code in (
+            "ffmpeg_record_start_failed",
+            "ffmpeg_capture_missing",
+            "ffmpeg_compose_failed",
+        ):
+            msg += self._record_log_hint()
+        self.warnings.append(msg)
+
     def step(self, text: str, *, duration: float = 4.0) -> None:
         self.add_step(text, duration=duration)
 
@@ -262,6 +274,16 @@ class RpaVideoSession:
         elapsed = time.monotonic() - self._t0
         self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration))
 
+    def _close_ffmpeg_record_log(self) -> None:
+        fh = self._ffmpeg_record_log_file
+        self._ffmpeg_record_log_file = None
+        if fh is None:
+            return
+        try:
+            fh.close()
+        except Exception:
+            pass
+
     def _start_ffmpeg_capture(self) -> None:
         if not self.enabled:
             return
@@ -271,26 +293,39 @@ class RpaVideoSession:
 
         cmd = _build_desktop_record_cmd(self._capture_path)
         if cmd is None:
-            self.warnings.append(
+            self._append_warning(
-                "ffmpeg_record_start_failed: desktop capture only supported on Windows"
+                "ffmpeg_record_start_failed",
+                "desktop capture only supported on Windows",
             )
             return
 
+        stderr_target: Any = subprocess.DEVNULL
+        try:
+            os.makedirs(self._logs_dir, exist_ok=True)
+            self._ffmpeg_record_log_file = open(
+                self._ffmpeg_record_log_path, "a", encoding="utf-8"
+            )
+            stderr_target = self._ffmpeg_record_log_file
+        except Exception as exc:
+            self.warnings.append(f"ffmpeg_record_log_open_failed: {exc}")
+
         try:
             self._ffmpeg_proc = subprocess.Popen(
                 cmd,
                 stdin=subprocess.PIPE,
                 stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
+                stderr=stderr_target,
             )
             self.capture_path = self._capture_path
         except Exception as exc:
-            self.warnings.append(f"ffmpeg_record_start_failed: {exc}")
+            self._append_warning("ffmpeg_record_start_failed", str(exc))
             self._ffmpeg_proc = None
+            self._close_ffmpeg_record_log()
 
     def _stop_ffmpeg_capture(self) -> None:
         proc = self._ffmpeg_proc
         if proc is None:
+            self._close_ffmpeg_record_log()
             return
         if proc.poll() is None:
             try:
@@ -314,8 +349,8 @@ class RpaVideoSession:
                 except Exception:
                     pass
         self._ffmpeg_proc = None
-        await_sleep = 0.3
+        self._close_ffmpeg_record_log()
-        time.sleep(await_sleep)
+        time.sleep(0.3)
 
     async def __aenter__(self) -> "RpaVideoSession":
         if self.enabled:
@@ -342,7 +377,7 @@ class RpaVideoSession:
         try:
             capture = Path(self._capture_path)
             if not capture.is_file() or capture.stat().st_size <= 0:
-                self.warnings.append("ffmpeg_capture_missing")
+                self._append_warning("ffmpeg_capture_missing")
                 self.artifacts = self.summary()
                 return
 
@@ -396,7 +431,7 @@ class RpaVideoSession:
                     ]
                 )
             if not ok:
-                self.warnings.append(f"ffmpeg_compose_failed: {err}")
+                self._append_warning("ffmpeg_compose_failed", err)
                 self.artifacts = self.summary()
                 return
 
@@ -407,10 +442,16 @@ class RpaVideoSession:
         self.artifacts = self.summary()
 
     def summary(self) -> Dict[str, Any]:
+        record_log = (
+            os.path.abspath(self._ffmpeg_record_log_path)
+            if self.enabled and self._ffmpeg_record_log_path
+            else None
+        )
         return {
             "enabled": self.enabled,
             "path": self.output_video_path,
             "capture_path": self.capture_path,
+            "record_log_path": record_log,
             "warnings": list(self.warnings),
         }
 

src/jiangchang_skill_core/unified_logging.py

@@ -10,17 +10,23 @@ import os
 import sys
 import uuid
 from logging.handlers import TimedRotatingFileHandler
-from typing import Optional
+from typing import Optional, Tuple
 
 from .runtime_env import get_data_root, get_user_id
 
 _skill_slug: str = ""
 _logger_name: str = ""
 
+# 避免 logging 内部错误再向 stderr 打印 "--- Logging error ---"
+logging.raiseExceptions = False
+
 
 def get_unified_logs_dir() -> str:
     path = os.path.join(get_data_root(), get_user_id(), "logs")
-    os.makedirs(path, exist_ok=True)
+    try:
+        os.makedirs(path, exist_ok=True)
+    except OSError:
+        pass
     return path
 
 
@@ -30,11 +36,27 @@ def get_skill_log_file_path() -> str:
     if override:
         parent = os.path.dirname(os.path.abspath(override))
         if parent:
-            os.makedirs(parent, exist_ok=True)
+            try:
+                os.makedirs(parent, exist_ok=True)
+            except OSError:
+                pass
         return os.path.abspath(override)
     return os.path.join(get_unified_logs_dir(), "jiangchang.log")
 
 
+def _can_open_log_file(log_path: str) -> Tuple[bool, str]:
+    """探测日志文件是否可追加写入（真实 open，非 os.access）。"""
+    try:
+        parent = os.path.dirname(os.path.abspath(log_path))
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        with open(log_path, "a", encoding="utf-8"):
+            pass
+        return True, ""
+    except Exception as exc:
+        return False, str(exc)
+
+
 def ensure_trace_for_process() -> str:
     """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。"""
     existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
@@ -81,44 +103,72 @@ def _backup_count() -> int:
         return 30
 
 
+def _clear_logger_handlers(log: logging.Logger) -> None:
+    for h in list(log.handlers):
+        log.removeHandler(h)
+        try:
+            h.close()
+        except Exception:
+            pass
+
+
+def _attach_fallback_handler(log: logging.Logger) -> None:
+    """文件日志不可用时仅挂 NullHandler，不向 stderr 打印 logging 异常。"""
+    log.addHandler(logging.NullHandler())
+
+
 def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
     """
     幂等：为指定 logger 挂载统一文件日志与可选 stderr。
     须在进程早期调用（如 CLI main 在业务日志之前）。
+    文件日志不可写时降级为 NullHandler，不影响 CLI 返回码。
     """
     global _skill_slug, _logger_name
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     _logger_name = logger_name
 
     log = logging.getLogger(logger_name)
-    if log.handlers:
+    _clear_logger_handlers(log)
-        return
     log.setLevel(_log_level_from_env())
-    path = get_skill_log_file_path()
-    fh = TimedRotatingFileHandler(
-        path,
-        when="midnight",
-        interval=1,
-        backupCount=_backup_count(),
-        encoding="utf-8",
-        delay=True,
-    )
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
-    fh.setFormatter(fmt)
+    filt = _SkillContextFilter()
-    fh.addFilter(_SkillContextFilter())
+
-    log.addHandler(fh)
+    log_path = get_skill_log_file_path()
+    ok, _err = _can_open_log_file(log_path)
+    if ok:
+        try:
+            fh = TimedRotatingFileHandler(
+                log_path,
+                when="midnight",
+                interval=1,
+                backupCount=_backup_count(),
+                encoding="utf-8",
+                delay=False,
+            )
+            fh.setFormatter(fmt)
+            fh.addFilter(filt)
+            log.addHandler(fh)
+        except Exception:
+            _attach_fallback_handler(log)
+    else:
+        _attach_fallback_handler(log)
+
     if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
         "1",
         "true",
         "yes",
         "on",
     ):
-        sh = logging.StreamHandler(sys.stderr)
+        try:
-        sh.setLevel(logging.WARNING)
+            sh = logging.StreamHandler(sys.stderr)
-        sh.setFormatter(fmt)
+            sh.setLevel(logging.WARNING)
-        sh.addFilter(_SkillContextFilter())
+            sh.setFormatter(fmt)
-        log.addHandler(sh)
+            sh.addFilter(filt)
+            log.addHandler(sh)
+        except Exception:
+            pass
     log.propagate = False
 
 
@@ -139,18 +189,25 @@ def attach_unified_file_handler(
     独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。
     """
     global _skill_slug
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     lg = logging.getLogger(logger_name)
-    lg.handlers.clear()
+    _clear_logger_handlers(lg)
     lg.setLevel(level)
-    parent = os.path.dirname(os.path.abspath(log_path))
-    if parent:
-        os.makedirs(parent, exist_ok=True)
-    fh = logging.FileHandler(log_path, encoding="utf-8")
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
-    fh.setFormatter(fmt)
+    filt = _SkillContextFilter()
-    fh.addFilter(_SkillContextFilter())
+
-    lg.addHandler(fh)
+    ok, _err = _can_open_log_file(log_path)
+    if ok:
+        try:
+            fh = logging.FileHandler(log_path, encoding="utf-8", delay=False)
+            fh.setFormatter(fmt)
+            fh.addFilter(filt)
+            lg.addHandler(fh)
+        except Exception:
+            _attach_fallback_handler(lg)
+    else:
+        _attach_fallback_handler(lg)
     lg.propagate = False
     return lg

tests/test_media_assets.py

@@ -0,0 +1,233 @@
+# -*- coding: utf-8 -*-
+"""media_assets 共享媒体资源解析测试。"""
+from __future__ import annotations
+
+import json
+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 _sample_manifest() -> dict:
+    bin_key = ma._platform_bin_key()
+    return {
+        "name": "media-assets",
+        "version": "0.2.0",
+        "directories": {"music": "music", "fonts": "fonts", "watermark": "watermark"},
+        "tools": {
+            "ffmpeg": {
+                bin_key: {
+                    "package": "ffmpeg-release-essentials",
+                    "archive_type": "zip",
+                    "urls": [{"name": "test", "url": "https://example.test/ffmpeg.zip"}],
+                    "archive_paths": {
+                        "ffmpeg": "*/bin/ffmpeg.exe",
+                        "ffprobe": "*/bin/ffprobe.exe",
+                    },
+                }
+            }
+        },
+    }
+
+
+def _install_tools_from_manifest(root: Path, warnings: list[str]) -> bool:
+    bin_key = ma._platform_bin_key()
+    bin_dir = root / "bin" / bin_key
+    bin_dir.mkdir(parents=True, exist_ok=True)
+    (bin_dir / ma._tool_filename("ffmpeg")).write_bytes(b"ffmpeg")
+    (bin_dir / ma._tool_filename("ffprobe")).write_bytes(b"ffprobe")
+    return True
+
+
+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:
+    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/.keep", "")
+        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"
+    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)
+    monkeypatch.setattr(ma, "_try_download_ffmpeg_tools", _install_tools_from_manifest)
+
+    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"
+
+
+def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    (root / "music").mkdir(parents=True)
+    (root / "fonts").mkdir(parents=True)
+    (root / "watermark").mkdir(parents=True)
+    (root / "manifest.json").write_text(
+        json.dumps(_sample_manifest(), ensure_ascii=False),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_ffmpeg_tools", _install_tools_from_manifest)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is True
+    assert status.ffmpeg_path is not None
+    assert status.ffprobe_path is not None

tests/test_unified_logging.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""统一日志：文件不可写时降级，首次 emit 不抛 PermissionError。"""
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+import unittest
+from logging.handlers import TimedRotatingFileHandler
+from unittest.mock import patch
+
+from jiangchang_skill_core import unified_logging as ul
+
+
+class TestUnifiedLogging(unittest.TestCase):
+    def setUp(self) -> None:
+        ul._skill_slug = ""
+        ul._logger_name = ""
+
+    def test_can_open_log_file_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "test.log")
+            ok, err = ul._can_open_log_file(path)
+            self.assertTrue(ok)
+            self.assertEqual(err, "")
+            self.assertTrue(os.path.isfile(path))
+
+    def test_setup_skill_logging_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "skill.log")
+            logger_name = "test.logger.writable"
+            with patch.object(ul, "get_skill_log_file_path", return_value=log_path):
+                ul.setup_skill_logging("receive-order", logger_name)
+                ul.get_skill_logger().info("hello")
+                self.assertTrue(os.path.isfile(log_path))
+                ul._clear_logger_handlers(logging.getLogger(logger_name))
+
+    def test_setup_skill_logging_unwritable_no_file_handler(self) -> None:
+        logger_name = "test.logger.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            ul.setup_skill_logging("receive-order", logger_name)
+            lg = logging.getLogger(logger_name)
+            lg.info("hello")
+        handlers = logging.getLogger(logger_name).handlers
+        self.assertFalse(any(isinstance(h, TimedRotatingFileHandler) for h in handlers))
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in handlers))
+
+    def test_attach_unified_file_handler_unwritable(self) -> None:
+        logger_name = "test.logger.attach.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            lg = ul.attach_unified_file_handler(
+                "/no/access/test.log",
+                skill_slug="receive-order",
+                logger_name=logger_name,
+            )
+            lg.info("hello")
+        self.assertFalse(
+            any(isinstance(h, logging.FileHandler) for h in lg.handlers)
+        )
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in lg.handlers))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_video_session.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""RpaVideoSession：ffmpeg 桌面录屏，无 Playwright record_video_dir。"""
+"""RpaVideoSession：ffmpeg 桌面录屏。"""
 from __future__ import annotations
 
 import asyncio
@@ -28,18 +28,17 @@ class TestRpaVideoSession(unittest.TestCase):
                     batch_id="batch_test",
                 )
                 self.assertFalse(session.enabled)
-                self.assertIsNone(session.record_video_dir)
-                self.assertIsNone(session.output_video_path)
-                self.assertIsNone(session.capture_path)
                 summ = session.summary()
                 self.assertFalse(summ["enabled"])
                 self.assertIsNone(summ["path"])
+                self.assertIsNone(summ["capture_path"])
+                self.assertIsNone(summ["record_log_path"])
         finally:
             if old is not None:
                 os.environ["OPENCLAW_RECORD_VIDEO"] = old
             config.reset_cache()
 
-    def test_enabled_paths_no_raw_video(self) -> None:
+    def test_enabled_paths_use_ffmpeg_capture(self) -> None:
         os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
         try:
             with tempfile.TemporaryDirectory() as tmp:
@@ -51,12 +50,16 @@ class TestRpaVideoSession(unittest.TestCase):
                     title="接收订单",
                 )
                 self.assertTrue(session.enabled)
-                self.assertIsNone(session.record_video_dir)
                 self.assertTrue(
                     session._capture_path.replace("\\", "/").endswith(
                         "/rpa-artifacts/exmail_20260602_120000/capture.mp4"
                     )
                 )
+                self.assertTrue(
+                    session._ffmpeg_record_log_path.replace("\\", "/").endswith(
+                        "/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log"
+                    )
+                )
                 self.assertTrue(
                     session._planned_output.replace("\\", "/").startswith(
                         tmp.replace("\\", "/") + "/videos/"
@@ -87,15 +90,17 @@ class TestRpaVideoSession(unittest.TestCase):
                         batch_id="batch_ff",
                     ) as video:
                         video.add_step("测试步骤")
-                        with patch.object(video, "finalize", return_value=None):
-                            pass
 
-                asyncio.run(_run())
+                with patch.object(RpaVideoSession, "finalize", return_value=None):
+                    asyncio.run(_run())
+
                 self.assertTrue(mock_popen.called)
                 cmd = mock_popen.call_args[0][0]
                 self.assertIn("ffmpeg", cmd)
                 self.assertIn("gdigrab", cmd)
                 self.assertIn("desktop", cmd)
+                kwargs = mock_popen.call_args[1]
+                self.assertIsNot(kwargs.get("stderr"), os.devnull)
         finally:
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
             config.reset_cache()
@@ -106,16 +111,6 @@ class TestRpaVideoSession(unittest.TestCase):
         try:
             with tempfile.TemporaryDirectory() as tmp:
                 config.reset_cache()
-
-                async def _run() -> None:
-                    async with RpaVideoSession(
-                        skill_slug="receive-order",
-                        skill_data_dir=tmp,
-                        batch_id="batch_no_ff",
-                    ):
-                        pass
-
-                asyncio.run(_run())
                 session = RpaVideoSession(
                     skill_slug="receive-order",
                     skill_data_dir=tmp,
@@ -138,10 +133,11 @@ class TestRpaVideoSession(unittest.TestCase):
                     batch_id="batch_warn",
                 )
                 asyncio.run(session.finalize())
-                self.assertIn("ffmpeg_capture_missing", session.warnings)
+                self.assertTrue(
+                    any("ffmpeg_capture_missing" in w for w in session.warnings)
+                )
+                self.assertIn("record_log=", "".join(session.warnings))
                 self.assertIsNone(session.output_video_path)
-                summ = session.summary()
-                self.assertIsNone(summ["path"])
         finally:
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
             config.reset_cache()