Compare commits

...

3 Commits

Author SHA1 Message Date
ae5466854f fix: sync unified logging probe before file handler attach
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-06-02 15:18:59 +08:00
b7163f12a5 chore: sync ffmpeg video session and logging fallback
All checks were successful
技能自动化发布 / release (push) Successful in 4s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-02 14:29:02 +08:00
f52536bdcc chore: sync ffmpeg RpaVideoSession, remove Playwright recording
All checks were successful
技能自动化发布 / release (push) Successful in 5s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-02 14:07:02 +08:00
5 changed files with 354 additions and 110 deletions

View File

@@ -159,10 +159,11 @@ skill 退出/抛错统一用 `ERROR:` 前缀 + 稳定码,方便宿主与上层
- **路径**`{CLAW_DATA_ROOT}/{user}/{slug}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.png`
- **失败必截图**:受 `OPENCLAW_ARTIFACTS_ON_FAILURE`(默认开)控制。
- **运行录屏成片**RPA 型技能默认 `OPENCLAW_RECORD_VIDEO=1`,表示录屏、中文字幕、背景音乐最终 MP4 **一体开启**。由 service 核心中的 `RpaVideoSession` 统一触发(见 `jiangchang_skill_core.rpa.video_session`,不在 CLI/E2E 入口单独实现。
- **最终视频**`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
- **中间产物**`{skill_data_dir}/rpa-artifacts/{batch_id}/`raw-video、字幕、失败截图等**不放最终 MP4**
- Playwright 使用 `record_video_dir` 指向 `rpa-artifacts/{batch_id}/raw-video/`
- **运行录屏成片**RPA 型技能默认 `OPENCLAW_RECORD_VIDEO=1`,表示 **ffmpeg 录制运行过程** + 字幕 + 背景音乐 + 最终 MP4 **一体开启**。由 `RpaVideoSession` 统一触发,不在 CLI/E2E 单独实现。
- **Playwright 不负责录屏**,仅浏览器自动化
- **ffmpeg 是唯一录屏器**Windows`gdigrab` + `desktop`
- **最终视频**`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
- **中间产物**`rpa-artifacts/{batch_id}/capture.mp4``subtitles/``logs/` 等,**不放最终 MP4**。
- **常见 tag**`before_submit` / `after_submit` / `captcha` / `login_fail` / `error`
---

View File

@@ -1,5 +1,4 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""统一 persistent context 启动封装。"""
"""统一 persistent context 启动封装(仅浏览器自动化,不负责录屏)。"""
from __future__ import annotations
@@ -25,12 +24,11 @@ async def launch_persistent_browser(
executable_path: str | None = None,
headless: bool | None = None,
extra_args: list[str] | None = None,
record_video_dir: str | None = None,
) -> "BrowserContext":
"""用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。
headless=None 时读 OPENCLAW_BROWSER_HEADLESS。
record_video_dir 非空则开录屏
录屏由 RpaVideoSessionffmpeg负责本函数不向 Playwright 传递任何录屏参数
"""
if headless is None:
v = (os.environ.get("OPENCLAW_BROWSER_HEADLESS") or "0").strip().lower()
@@ -53,8 +51,6 @@ async def launch_persistent_browser(
)
if ignore is not None:
launch_kwargs["ignore_default_args"] = ignore
if record_video_dir:
launch_kwargs["record_video_dir"] = record_video_dir
context = await playwright.chromium.launch_persistent_context(**launch_kwargs)
if stealth_enabled():

View File

@@ -1,21 +1,21 @@
"""RPA 运行录屏会话:Playwright raw video → 字幕 + 背景音乐 → 最终 MP4。"""
"""RPA 运行录屏会话:ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
from __future__ import annotations
import asyncio
import os
import random
import shutil
import subprocess
import sys
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
# 与 screencast/composer.py 保持一致的字幕样式
_JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14,"
"PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
@@ -27,6 +27,8 @@ _DEFAULT_MUSIC_ROOT = os.environ.get(
r"D:\OpenClaw\client-commons\media-assets",
)
_CAPTURE_FRAMERATE = 15
def _record_video_enabled() -> bool:
return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
@@ -50,7 +52,6 @@ class _StepEntry:
def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
"""根据步骤时间线生成 SRT防重叠"""
MIN_DURATION = 1.0
GAP = 0.1
steps = sorted(entries, key=lambda e: e.start)
@@ -86,16 +87,6 @@ def _escape_srt_path(path: str) -> str:
return path.replace("\\", "/").replace(":", "\\:")
def _find_raw_webm(raw_dir: str) -> Optional[Path]:
root = Path(raw_dir)
if not root.is_dir():
return None
webms = list(root.rglob("*.webm"))
if not webms:
return None
return max(webms, key=lambda p: p.stat().st_size)
def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
try:
result = subprocess.run(
@@ -113,8 +104,31 @@ def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
return True, ""
def _compose_webm_mp4(
webm_path: Path,
def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]:
"""Windows gdigrab 全桌面录制;非 Windows 返回 None。"""
if sys.platform != "win32":
return None
return [
"ffmpeg",
"-y",
"-f",
"gdigrab",
"-framerate",
str(_CAPTURE_FRAMERATE),
"-i",
"desktop",
"-c:v",
"libx264",
"-preset",
"ultrafast",
"-pix_fmt",
"yuv420p",
capture_path,
]
def _compose_capture_mp4(
capture_path: Path,
srt_path: Path,
output_path: Path,
*,
@@ -134,39 +148,60 @@ def _compose_webm_mp4(
f"[1:a]volume={music_volume},apad[aout]"
)
cmd = [
"ffmpeg", "-y",
"-i", str(webm_path),
"-i", str(music_file),
"-filter_complex", filter_complex,
"-map", "[vout]",
"-map", "[aout]",
"-c:v", "libx264",
"-preset", "fast",
"-crf", "23",
"-c:a", "aac",
"-b:a", "128k",
"ffmpeg",
"-y",
"-i",
str(capture_path),
"-i",
str(music_file),
"-filter_complex",
filter_complex,
"-map",
"[vout]",
"-map",
"[aout]",
"-c:v",
"libx264",
"-preset",
"fast",
"-crf",
"23",
"-c:a",
"aac",
"-b:a",
"128k",
"-shortest",
"-pix_fmt", "yuv420p",
"-pix_fmt",
"yuv420p",
str(output_path),
]
else:
cmd = [
"ffmpeg", "-y",
"-i", str(webm_path),
"-vf", subtitle_filter,
"-c:v", "libx264",
"-preset", "fast",
"-crf", "23",
"-c:a", "aac",
"-b:a", "128k",
"-pix_fmt", "yuv420p",
"ffmpeg",
"-y",
"-i",
str(capture_path),
"-vf",
subtitle_filter,
"-c:v",
"libx264",
"-preset",
"fast",
"-crf",
"23",
"-c:a",
"aac",
"-b:a",
"128k",
"-pix_fmt",
"yuv420p",
str(output_path),
]
return _run_ffmpeg(cmd)
class RpaVideoSession:
"""Service/RPA 层统一录屏会话OPENCLAW_RECORD_VIDEO=1 时启用完整成片流程。"""
"""Service/RPA 层统一 ffmpeg 桌面录屏OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
def __init__(
self,
@@ -183,7 +218,7 @@ class RpaVideoSession:
self.enabled = _record_video_enabled()
self.warnings: List[str] = []
self.output_video_path: Optional[str] = None
self.record_video_dir: Optional[str] = None
self.capture_path: Optional[str] = None
self.artifacts: Dict[str, Any] = {}
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -194,24 +229,44 @@ class RpaVideoSession:
self._artifact_root = os.path.join(
self.skill_data_dir, "rpa-artifacts", batch_id
)
self._raw_video_dir = os.path.join(self._artifact_root, "raw-video")
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")
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)
os.makedirs(self._raw_video_dir, exist_ok=True)
os.makedirs(self._subtitles_dir, exist_ok=True)
os.makedirs(self._logs_dir, exist_ok=True)
self.record_video_dir = self._raw_video_dir
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)
def add_step(self, text: str, *, duration: float = 4.0) -> None:
"""记录一步骤字幕(相对录屏开始时间)。"""
if not self.enabled:
return
if self._t0 is None:
@@ -219,71 +274,188 @@ 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
if not shutil.which("ffmpeg"):
self.warnings.append("ffmpeg_not_found")
return
cmd = _build_desktop_record_cmd(self._capture_path)
if cmd is None:
self._append_warning(
"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=stderr_target,
)
self.capture_path = self._capture_path
except Exception as 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:
if proc.stdin:
proc.stdin.write(b"q")
proc.stdin.flush()
except Exception:
try:
proc.terminate()
except Exception:
pass
try:
proc.wait(timeout=20)
except subprocess.TimeoutExpired:
try:
proc.kill()
except Exception:
pass
try:
proc.wait(timeout=5)
except Exception:
pass
self._ffmpeg_proc = None
self._close_ffmpeg_record_log()
time.sleep(0.3)
async def __aenter__(self) -> "RpaVideoSession":
if self.enabled:
self._t0 = time.monotonic()
if self.title:
self.step(self.title, duration=3.0)
self.add_step(self.title, duration=3.0)
self._start_ffmpeg_capture()
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
if self.enabled:
self._stop_ffmpeg_capture()
await self.finalize()
async def finalize(self) -> None:
"""在 Playwright context 已 close 后合成最终 MP4失败只记 warning。"""
"""停止录制后合成最终 MP4失败只记 warning,不抛异常"""
if not self.enabled:
self.artifacts = self.summary()
return
if not self._steps and self.title:
self.step(self.title)
self.add_step(self.title)
try:
await asyncio.sleep(0.5)
webm = _find_raw_webm(self._raw_video_dir)
if webm is None:
self.warnings.append("no_playwright_raw_video")
self.artifacts = self.to_artifact_dict()
capture = Path(self._capture_path)
if not capture.is_file() or capture.stat().st_size <= 0:
self._append_warning("ffmpeg_capture_missing")
self.artifacts = self.summary()
return
self.capture_path = str(capture.resolve())
entries = list(self._steps)
if entries:
entries[-1].duration = max(entries[-1].duration, 5.0)
_generate_srt(entries, self._srt_path)
try:
_generate_srt(entries, self._srt_path)
except Exception as exc:
self.warnings.append(f"subtitle_write_failed: {exc}")
music = _resolve_music_file()
if music is None:
self.warnings.append("background_music_missing")
ok, err = _compose_webm_mp4(
webm,
Path(self._srt_path),
srt_file = Path(self._srt_path)
use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
ok, err = _compose_capture_mp4(
capture,
srt_file if use_srt else Path(self._srt_path),
Path(self._planned_output),
music_file=music,
)
if not ok and music is not None:
self.warnings.append(f"ffmpeg_with_music_failed: {err}")
ok, err = _compose_webm_mp4(
webm,
Path(self._srt_path),
ok, err = _compose_capture_mp4(
capture,
srt_file if use_srt else Path(self._srt_path),
Path(self._planned_output),
music_file=None,
)
if not ok and not use_srt:
ok, err = _run_ffmpeg(
[
"ffmpeg",
"-y",
"-i",
str(capture),
"-c:v",
"libx264",
"-preset",
"fast",
"-crf",
"23",
"-pix_fmt",
"yuv420p",
str(self._planned_output),
]
)
if not ok:
self.warnings.append(f"ffmpeg_compose_failed: {err}")
self.artifacts = self.to_artifact_dict()
self._append_warning("ffmpeg_compose_failed", err)
self.artifacts = self.summary()
return
self.output_video_path = str(Path(self._planned_output).resolve())
self.artifacts = self.to_artifact_dict()
except Exception as exc:
self.warnings.append(f"video_finalize_error: {exc}")
self.artifacts = self.to_artifact_dict()
def to_artifact_dict(self) -> Dict[str, Any]:
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),
"batch_id": self.batch_id,
}
def to_artifact_dict(self) -> Dict[str, Any]:
data = self.summary()
data["batch_id"] = self.batch_id
return data

View File

@@ -1,5 +1,5 @@
"""
统一文件日志:{DATA_ROOT}/{USER_ID}/logs/jiangchang.log
统一文件日志:{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log
按日轮转;行内带 trace_id 与 skill_slug便于跨技能排查。
"""
@@ -10,31 +10,55 @@ 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
def get_skill_log_file_path() -> str:
"""与 setup 写入的主日志文件一致(供终端提示等)。"""
override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip()
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()
if existing:
return existing
@@ -44,8 +68,11 @@ def ensure_trace_for_process() -> str:
def subprocess_env_with_trace(environ: Optional[dict] = None) -> dict:
"""subprocess.run(..., env=...) 时使用,保证带上当前 trace。"""
ensure_trace_for_process()
tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or ensure_trace_for_process()
tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
if not tid:
tid = ensure_trace_for_process()
base = os.environ if environ is None else environ
return {**base, "JIANGCHANG_TRACE_ID": tid}
@@ -57,7 +84,9 @@ class _SkillContextFilter(logging.Filter):
return True
_FORMAT = "%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
_FORMAT = (
"%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
)
_DATEFMT = "%Y-%m-%dT%H:%M:%S"
@@ -74,35 +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:
return
_clear_logger_handlers(log)
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)
fh.addFilter(_SkillContextFilter())
log.addHandler(fh)
if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in ("1", "true", "yes", "on"):
sh = logging.StreamHandler(sys.stderr)
sh.setLevel(logging.WARNING)
sh.setFormatter(fmt)
sh.addFilter(_SkillContextFilter())
log.addHandler(sh)
filt = _SkillContextFilter()
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",
):
try:
sh = logging.StreamHandler(sys.stderr)
sh.setLevel(logging.WARNING)
sh.setFormatter(fmt)
sh.addFilter(filt)
log.addHandler(sh)
except Exception:
pass
log.propagate = False
@@ -119,19 +185,29 @@ def attach_unified_file_handler(
logger_name: str,
level: int = logging.DEBUG,
) -> logging.Logger:
"""
独立子进程内追加同一日志文件(如 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)
fh.addFilter(_SkillContextFilter())
lg.addHandler(fh)
filt = _SkillContextFilter()
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

View File

@@ -50,14 +50,13 @@ class SimRpaAdapter(AdapterBase):
batch_id=batch_id,
title="技能仿真演示",
) as video:
video.step("正在打开目标页面")
video.add_step("正在打开目标页面")
try:
async with async_playwright() as pw:
context = await launch_persistent_browser(
pw,
profile_dir=profile_dir,
headless=config.get_bool("OPENCLAW_BROWSER_HEADLESS"),
record_video_dir=video.record_video_dir,
)
page = context.pages[0] if context.pages else await context.new_page()
await page.goto(url, wait_until="domcontentloaded", timeout=60_000)