4 Commits

Author SHA1 Message Date
367df10755 chore: update skill release readme source
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 26s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-14 10:59:37 +08:00
857e7e3d73 feat: improve rpa video timing and narration
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 20s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-08 11:50:54 +08:00
0e2b73e1ad feat: improve rpa video audio composition
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 50s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-08 10:14:31 +08:00
36a411810d ci: publish formal version from pyproject.toml on main push
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 25s
Remove .postN CI versioning, fix runtime version_ge for legacy post releases, and use ASCII issue separators in health output.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-07 09:26:41 +08:00
12 changed files with 1068 additions and 183 deletions

View File

@@ -22,13 +22,6 @@ jobs:
steps: steps:
- uses: https://git.jc2009.com/admin/actions-checkout@v4 - uses: https://git.jc2009.com/admin/actions-checkout@v4
- name: Resolve and patch package version
env:
GITHUB_REF: ${{ github.ref }}
GITHUB_RUN_NUMBER: ${{ github.run_number }}
GITHUB_RUN_ID: ${{ github.run_id }}
run: python3.12 tools/ci_set_package_version.py
- name: Install build tools - name: Install build tools
run: python3.12 -m pip install --upgrade build twine --break-system-packages run: python3.12 -m pip install --upgrade build twine --break-system-packages

View File

@@ -108,6 +108,13 @@ jobs:
' shutil.copytree(src, dst, ignore=ignore_fn, dirs_exist_ok=True)' \ ' shutil.copytree(src, dst, ignore=ignore_fn, dirs_exist_ok=True)' \
'' \ '' \
"shutil.copy2('SKILL.md', os.path.join(PACKAGE, 'SKILL.md'))" \ "shutil.copy2('SKILL.md', os.path.join(PACKAGE, 'SKILL.md'))" \
"readme_src = os.path.abspath('README.md')" \
"readme_dst = os.path.join(PACKAGE, 'README.md')" \
'if os.path.isfile(readme_src):' \
" shutil.copy2(readme_src, readme_dst)" \
" print('Copied README.md')" \
'if os.path.isfile(readme_src) and not os.path.isfile(readme_dst):' \
" raise RuntimeError('README.md exists in skill root but was not packaged')" \
"copy_dir('references', os.path.join(PACKAGE, 'references'), references_ignore)" \ "copy_dir('references', os.path.join(PACKAGE, 'references'), references_ignore)" \
"copy_dir('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \ "copy_dir('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \
"copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \ "copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \
@@ -144,7 +151,7 @@ jobs:
skill_readme_md = (post.content or '').strip() skill_readme_md = (post.content or '').strip()
skill_description = metadata.get('description') skill_description = metadata.get('description')
metadata['readme_md'] = skill_readme_md metadata['readme_md'] = skill_readme_md
readme_path = os.path.join('references', 'README.md') readme_path = 'README.md'
if os.path.isfile(readme_path): if os.path.isfile(readme_path):
try: try:
readme_post = frontmatter.load(readme_path) readme_post = frontmatter.load(readme_path)

View File

@@ -142,17 +142,18 @@ run_screencast(
## 5. 发版流程(维护者) ## 5. 发版流程(维护者)
### main 分支自动预发布 ### 发布规则
每次推送到 `main` 会触发 `.github/workflows/publish.yml`,自动构建并发布 **post 预发布版本**(仅供 CI 流水线内部使用,**不是**正式版本线) - jiangchang-platform-kit **只发布正式版本**
- 每次需要发布公共能力时,先 bump `pyproject.toml``version`(例如 `1.0.10``1.0.11`)。
正式版本线使用 **1.0.x** semver通过 git tag 发布 - 推送 `main` 后,`.github/workflows/publish.yml` 会构建并上传该正式版本到 Gitea PyPI
- 发布包版本必须等于 `pyproject.toml` 中的 `version`;若该版本已存在,上传会失败,需继续 bump 版本号后再推送。
```bash ```bash
git push origin main git push origin main
``` ```
测试人员升级安装(始终安装当前正式 tag 对应的版本,例如 `1.0.9` 升级安装
```bash ```bash
python -m pip install --upgrade \ python -m pip install --upgrade \
@@ -161,15 +162,6 @@ python -m pip install --upgrade \
jiangchang-platform-kit jiangchang-platform-kit
``` ```
### 正式版本tag
打 tag 并推送后,发布 **去掉 `v` 前缀** 的正式版本号(例如 `v1.0.9` → PyPI 包 `1.0.9`
```bash
git tag v1.0.9
git push origin v1.0.9
```
### 安装后验证 ### 安装后验证
```bash ```bash

View File

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

View File

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

View File

@@ -1,9 +1,14 @@
"""RPA 运行录屏会话ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。""" """RPA 运行录屏会话ffmpeg 桌面录制 → 字幕 + 旁白 + 背景音乐 → 最终 MP4。
背景音乐默认循环铺满录屏时长并在结尾淡出;旁白使用 Windows 本地 SAPI 合成,
失败时静默降级不影响视频生成。Skill 无需额外配置。
"""
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import os import os
import re
import subprocess import subprocess
import sys import sys
import time import time
@@ -18,6 +23,7 @@ from ..media_assets import (
ensure_media_assets, ensure_media_assets,
pick_background_music, pick_background_music,
resolve_ffmpeg, resolve_ffmpeg,
resolve_ffprobe,
) )
_JIANGCHANG_SUBTITLE_STYLE = ( _JIANGCHANG_SUBTITLE_STYLE = (
@@ -28,6 +34,54 @@ _JIANGCHANG_SUBTITLE_STYLE = (
_CAPTURE_FRAMERATE = 15 _CAPTURE_FRAMERATE = 15
# 背景音乐循环 + 结尾淡出(默认策略,非 env 配置)
_MUSIC_VOLUME = 0.12
_VOICE_VOLUME = 1.0
_MUSIC_FADE_OUT_SECONDS = 3.0
_MIN_MUSIC_FADE_OUT_SECONDS = 1.0
_SHORT_VIDEO_FADE_RATIO = 0.25
_VOICEOVER_MAX_CHARS = 60
_VOICEOVER_DEDUP_SECONDS = 10.0
_VOICEOVER_MIN_GAP_SECONDS = 2.0
_INTRO_BUFFER_SECONDS = 5.0
_INTRO_STABILIZE_SECONDS = 1.0
_OUTRO_BUFFER_SECONDS = 5.0
_VOICEOVER_SKIP_SUBSTRINGS = (
"等待 1-5s",
"跳过重复",
"写入联系人库",
"探针模式",
"异常",
"失败",
)
_VOICEOVER_IMPORTANT_SUBSTRINGS = (
"开始",
"打开",
"启动",
"定位",
"输入",
"点击",
"搜索",
"等待搜索结果",
"检查登录状态",
"处理验证码",
"等待人工验证",
"解析",
"提取",
"采集完成",
"部分完成",
"完成",
)
# 明显技术日志:时间戳前缀、纯数字进度等
_TECH_LOG_RE = re.compile(
r"^\d{4}-\d{2}-\d{2}|^\[\w+\]|^DEBUG:|^INFO:|^WARN:|^ERROR:"
)
def _record_video_enabled() -> bool: def _record_video_enabled() -> bool:
return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False) return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
@@ -41,6 +95,68 @@ def _resolve_ffmpeg_exe() -> Optional[Path]:
return path if path.is_file() else None return path if path.is_file() else None
def _resolve_ffprobe_exe() -> Optional[Path]:
"""定位 ffprobe优先 media_assets bundle否则尝试 ffmpeg 同目录。"""
ffprobe = resolve_ffprobe()
if ffprobe is not None:
path = Path(ffprobe)
if path.is_file():
return path
ffmpeg = _resolve_ffmpeg_exe()
if ffmpeg is not None:
sibling = ffmpeg.parent / ("ffprobe.exe" if sys.platform == "win32" else "ffprobe")
if sibling.is_file():
return sibling
return None
def _probe_media_duration_seconds(
ffprobe_exe: Optional[Path], media_path: Path
) -> Optional[float]:
if ffprobe_exe is None or not media_path.is_file():
return None
try:
result = subprocess.run(
[
str(ffprobe_exe),
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=nk=1:nw=1",
str(media_path),
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=30,
)
if result.returncode != 0:
return None
raw = (result.stdout or "").strip()
if not raw:
return None
value = float(raw)
return value if value > 0 else None
except Exception:
return None
def _music_fade_params(
duration_seconds: Optional[float],
) -> tuple[Optional[float], Optional[float]]:
if duration_seconds is None or duration_seconds <= 0:
return (None, None)
fade = min(
_MUSIC_FADE_OUT_SECONDS,
max(_MIN_MUSIC_FADE_OUT_SECONDS, duration_seconds * _SHORT_VIDEO_FADE_RATIO),
)
fade_start = max(0.0, duration_seconds - fade)
return (fade_start, fade)
def _resolve_music_file(warnings: List[str]) -> Optional[Path]: def _resolve_music_file(warnings: List[str]) -> Optional[Path]:
music = pick_background_music() music = pick_background_music()
if music is not None: if music is not None:
@@ -59,6 +175,183 @@ class _StepEntry:
duration: float = 4.0 duration: float = 4.0
@dataclass
class _VoiceClip:
path: Path
delay_ms: int
def _voiceover_text_for_step(text: str) -> Optional[str]:
"""过滤适合朗读的步骤文本;噪音/技术日志返回 None。"""
cleaned = (text or "").strip()
if not cleaned:
return None
if _TECH_LOG_RE.search(cleaned):
return None
for skip in _VOICEOVER_SKIP_SUBSTRINGS:
if skip in cleaned:
return None
if len(cleaned) > _VOICEOVER_MAX_CHARS:
cleaned = cleaned[:_VOICEOVER_MAX_CHARS]
return cleaned
def _is_important_voiceover(text: str) -> bool:
for important in _VOICEOVER_IMPORTANT_SUBSTRINGS:
if important in text:
return True
return False
def _select_voiceover_clips(
entries: List[_StepEntry],
) -> List[tuple[_StepEntry, str]]:
"""按时间排序,去重并保证旁白间隔;关键动作可绕过最小间隔。"""
result: List[tuple[_StepEntry, str]] = []
last_text_at: Dict[str, float] = {}
last_spoken_at = -_VOICEOVER_MIN_GAP_SECONDS
for entry in sorted(entries, key=lambda e: e.start):
spoken = _voiceover_text_for_step(entry.text)
if not spoken:
continue
prev = last_text_at.get(spoken)
if prev is not None and entry.start - prev < _VOICEOVER_DEDUP_SECONDS:
continue
if (
not _is_important_voiceover(spoken)
and entry.start - last_spoken_at < _VOICEOVER_MIN_GAP_SECONDS
):
continue
last_text_at[spoken] = entry.start
last_spoken_at = entry.start
result.append((entry, spoken))
return result
def _tts_available() -> bool:
return sys.platform == "win32"
def _sapi_synthesize_wav(text: str, output_wav: Path) -> bool:
if not _tts_available():
return False
output_wav.parent.mkdir(parents=True, exist_ok=True)
out_str = str(output_wav.resolve()).replace("'", "''")
text_esc = text.replace("'", "''")
ps_script = (
"Add-Type -AssemblyName System.Speech; "
"$s = New-Object System.Speech.Synthesis.SpeechSynthesizer; "
f"$s.SetOutputToWaveFile('{out_str}'); "
f"$s.Speak('{text_esc}'); "
"$s.Dispose()"
)
try:
result = subprocess.run(
["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_script],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=120,
)
return (
result.returncode == 0
and output_wav.is_file()
and output_wav.stat().st_size > 0
)
except Exception:
return False
def _mix_voice_clips(
ffmpeg_exe: Path, clips: List[_VoiceClip], output_wav: Path
) -> bool:
if not clips:
return False
output_wav.parent.mkdir(parents=True, exist_ok=True)
filter_parts: List[str] = []
mix_labels: List[str] = []
cmd_inputs: List[str] = []
for i, clip in enumerate(clips):
cmd_inputs.extend(["-i", str(clip.path)])
label = f"v{i}"
filter_parts.append(
f"[{i}:a]adelay={clip.delay_ms}|{clip.delay_ms},"
f"volume={_VOICE_VOLUME}[{label}]"
)
mix_labels.append(f"[{label}]")
if len(clips) == 1:
filter_complex = filter_parts[0].replace(f"[v0]", "[out]")
# single clip: relabel output
filter_complex = (
f"[0:a]adelay={clips[0].delay_ms}|{clips[0].delay_ms},"
f"volume={_VOICE_VOLUME}[out]"
)
else:
filter_complex = (
";".join(filter_parts)
+ ";"
+ "".join(mix_labels)
+ f"amix=inputs={len(clips)}:duration=longest:dropout_transition=0:normalize=0[out]"
)
cmd_tail = [
"-y",
*cmd_inputs,
"-filter_complex",
filter_complex,
"-map",
"[out]",
str(output_wav),
]
ok, _ = _run_ffmpeg(ffmpeg_exe, cmd_tail)
return ok and output_wav.is_file()
def _synthesize_step_voiceover(
entries: List[_StepEntry],
output_wav: Path,
warnings: List[str],
*,
ffmpeg_exe: Optional[Path] = None,
) -> Optional[Path]:
"""本地 Windows SAPI 逐条合成旁白并按步骤 start 时间对齐;失败返回 None。"""
if not _tts_available():
warnings.append("tts_unsupported_platform")
return None
clips_to_speak = _select_voiceover_clips(entries)
if not clips_to_speak:
return None
voice_dir = output_wav.parent
voice_dir.mkdir(parents=True, exist_ok=True)
temp_clips: List[_VoiceClip] = []
for i, (entry, spoken) in enumerate(clips_to_speak):
clip_path = voice_dir / f"voice_{i:03d}.wav"
if not _sapi_synthesize_wav(spoken, clip_path):
warnings.append("tts_synthesis_failed")
return None
delay_ms = max(0, int(entry.start * 1000))
temp_clips.append(_VoiceClip(path=clip_path, delay_ms=delay_ms))
if ffmpeg_exe is None:
ffmpeg_exe = _resolve_ffmpeg_exe()
if ffmpeg_exe is None:
warnings.append("tts_mix_ffmpeg_missing")
return None
if not _mix_voice_clips(ffmpeg_exe, temp_clips, output_wav):
warnings.append("tts_mix_failed")
return None
return output_wav
def _generate_srt(entries: List[_StepEntry], output_path: str) -> None: def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
MIN_DURATION = 1.0 MIN_DURATION = 1.0
GAP = 0.1 GAP = 0.1
@@ -117,27 +410,16 @@ def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: List[str]) -> tuple[bool, str]:
return True, "" return True, ""
def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]: def _build_music_filter(
"""Windows gdigrab 全桌面录制;非 Windows 返回 None。""" audio_input_idx: int,
if sys.platform != "win32": fade_start: Optional[float],
return None fade_duration: Optional[float],
return _ffmpeg_cmd( output_label: str = "music",
ffmpeg_exe, ) -> str:
"-y", filt = f"[{audio_input_idx}:a]volume={_MUSIC_VOLUME}"
"-f", if fade_start is not None and fade_duration is not None:
"gdigrab", filt += f",afade=t=out:st={fade_start}:d={fade_duration}"
"-framerate", return filt + f"[{output_label}]"
str(_CAPTURE_FRAMERATE),
"-i",
"desktop",
"-c:v",
"libx264",
"-preset",
"ultrafast",
"-pix_fmt",
"yuv420p",
capture_path,
)
def _compose_capture_mp4( def _compose_capture_mp4(
@@ -147,7 +429,8 @@ def _compose_capture_mp4(
output_path: Path, output_path: Path,
*, *,
music_file: Optional[Path] = None, music_file: Optional[Path] = None,
music_volume: float = 0.15, voiceover_file: Optional[Path] = None,
warnings: Optional[List[str]] = None,
) -> tuple[bool, str]: ) -> tuple[bool, str]:
output_path.parent.mkdir(parents=True, exist_ok=True) output_path.parent.mkdir(parents=True, exist_ok=True)
srt_esc = _escape_srt_path(str(srt_path.resolve())) srt_esc = _escape_srt_path(str(srt_path.resolve()))
@@ -156,38 +439,77 @@ def _compose_capture_mp4(
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'" f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
) )
if music_file and music_file.is_file(): has_music = music_file is not None and music_file.is_file()
filter_complex = ( has_voice = voiceover_file is not None and voiceover_file.is_file()
f"[0:v]{subtitle_filter}[vout];"
f"[1:a]volume={music_volume},apad[aout]" fade_start: Optional[float] = None
fade_duration: Optional[float] = None
if has_music:
ffprobe = _resolve_ffprobe_exe()
duration = _probe_media_duration_seconds(ffprobe, capture_path)
fade_start, fade_duration = _music_fade_params(duration)
if fade_start is None and warnings is not None:
warnings.append("video_duration_probe_failed_for_music_fade")
if has_music or has_voice:
cmd_tail: List[str] = ["-y", "-i", str(capture_path)]
input_idx = 1
filter_parts: List[str] = [f"[0:v]{subtitle_filter}[vout]"]
audio_labels: List[str] = []
if has_music:
cmd_tail.extend(["-stream_loop", "-1", "-i", str(music_file)])
music_label = "aout" if not has_voice else "music"
filter_parts.append(
_build_music_filter(input_idx, fade_start, fade_duration, music_label)
)
if has_voice:
audio_labels.append(f"[{music_label}]")
input_idx += 1
if has_voice:
cmd_tail.extend(["-i", str(voiceover_file)])
if has_music:
filter_parts.append(
f"[{input_idx}:a]volume={_VOICE_VOLUME}[voice]"
)
audio_labels.append("[voice]")
else:
# voice-onlyapad 补静音,避免 -shortest 按旁白长度截短视频
filter_parts.append(
f"[{input_idx}:a]volume={_VOICE_VOLUME},apad[aout]"
)
input_idx += 1
if len(audio_labels) == 2:
filter_parts.append(
"".join(audio_labels) + "amix=inputs=2:normalize=0[aout]"
)
cmd_tail.extend(
[
"-filter_complex",
";".join(filter_parts),
"-map",
"[vout]",
"-map",
"[aout]",
"-c:v",
"libx264",
"-preset",
"fast",
"-crf",
"23",
"-c:a",
"aac",
"-b:a",
"128k",
"-shortest",
"-pix_fmt",
"yuv420p",
str(output_path),
]
) )
cmd_tail = [
"-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",
str(output_path),
]
else: else:
cmd_tail = [ cmd_tail = [
"-y", "-y",
@@ -212,6 +534,29 @@ def _compose_capture_mp4(
return _run_ffmpeg(ffmpeg_exe, cmd_tail) return _run_ffmpeg(ffmpeg_exe, cmd_tail)
def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]:
"""Windows gdigrab 全桌面录制;非 Windows 返回 None。"""
if sys.platform != "win32":
return None
return _ffmpeg_cmd(
ffmpeg_exe,
"-y",
"-f",
"gdigrab",
"-framerate",
str(_CAPTURE_FRAMERATE),
"-i",
"desktop",
"-c:v",
"libx264",
"-preset",
"ultrafast",
"-pix_fmt",
"yuv420p",
capture_path,
)
class RpaVideoSession: class RpaVideoSession:
"""Service/RPA 层统一 ffmpeg 桌面录屏OPENCLAW_RECORD_VIDEO=1 时启用成片流程。""" """Service/RPA 层统一 ffmpeg 桌面录屏OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
@@ -222,11 +567,13 @@ class RpaVideoSession:
skill_data_dir: str, skill_data_dir: str,
batch_id: str, batch_id: str,
title: str = "", title: str = "",
closing_title: str = "",
) -> None: ) -> None:
self.skill_slug = skill_slug self.skill_slug = skill_slug
self.skill_data_dir = os.path.abspath(skill_data_dir) self.skill_data_dir = os.path.abspath(skill_data_dir)
self.batch_id = batch_id self.batch_id = batch_id
self.title = title self.title = title
self.closing_title = closing_title
self.enabled = _record_video_enabled() self.enabled = _record_video_enabled()
self.warnings: List[str] = [] self.warnings: List[str] = []
self.output_video_path: Optional[str] = None self.output_video_path: Optional[str] = None
@@ -242,12 +589,17 @@ class RpaVideoSession:
self.skill_data_dir, "rpa-artifacts", batch_id self.skill_data_dir, "rpa-artifacts", batch_id
) )
self._subtitles_dir = os.path.join(self._artifact_root, "subtitles") self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
self._voiceover_dir = os.path.join(self._artifact_root, "voiceover")
self._logs_dir = os.path.join(self._artifact_root, "logs") 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._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._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._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt")
self._voiceover_path = os.path.join(self._voiceover_dir, "voiceover.wav")
self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4") self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
self._music_path: Optional[str] = None
self._voiceover_output: Optional[str] = None
self._steps: List[_StepEntry] = [] self._steps: List[_StepEntry] = []
self._t0: Optional[float] = None self._t0: Optional[float] = None
self._ffmpeg_proc: Optional[subprocess.Popen] = None self._ffmpeg_proc: Optional[subprocess.Popen] = None
@@ -256,6 +608,7 @@ class RpaVideoSession:
if self.enabled: if self.enabled:
os.makedirs(self._videos_dir, exist_ok=True) os.makedirs(self._videos_dir, exist_ok=True)
os.makedirs(self._subtitles_dir, exist_ok=True) os.makedirs(self._subtitles_dir, exist_ok=True)
os.makedirs(self._voiceover_dir, exist_ok=True)
os.makedirs(self._logs_dir, exist_ok=True) os.makedirs(self._logs_dir, exist_ok=True)
os.makedirs(self._artifact_root, exist_ok=True) os.makedirs(self._artifact_root, exist_ok=True)
@@ -366,19 +719,80 @@ class RpaVideoSession:
self._close_ffmpeg_record_log() self._close_ffmpeg_record_log()
time.sleep(0.3) time.sleep(0.3)
async def _safe_add_step(self, text: str, *, duration: float = 4.0) -> None:
try:
self.add_step(text, duration=duration)
except Exception:
pass
async def __aenter__(self) -> "RpaVideoSession": async def __aenter__(self) -> "RpaVideoSession":
if self.enabled: if self.enabled:
self._t0 = time.monotonic() self._t0 = time.monotonic()
if self.title:
self.add_step(self.title, duration=3.0)
self._start_ffmpeg_capture() self._start_ffmpeg_capture()
await asyncio.sleep(_INTRO_STABILIZE_SECONDS)
if self.title:
await self._safe_add_step(self.title, duration=4.0)
intro_wait = _INTRO_BUFFER_SECONDS - _INTRO_STABILIZE_SECONDS
if intro_wait > 0:
await asyncio.sleep(intro_wait)
return self return self
async def __aexit__(self, exc_type, exc, tb) -> None: async def __aexit__(self, exc_type, exc, tb) -> None:
if self.enabled: if self.enabled:
if self.closing_title:
await self._safe_add_step(self.closing_title, duration=4.0)
await asyncio.sleep(_OUTRO_BUFFER_SECONDS)
self._stop_ffmpeg_capture() self._stop_ffmpeg_capture()
await self.finalize() await self.finalize()
def _try_compose_variants(
self,
ffmpeg_exe: Path,
capture: Path,
srt_file: Path,
output: Path,
*,
music: Optional[Path],
voiceover: Optional[Path],
use_srt: bool,
) -> tuple[bool, str]:
"""按 music/voice 组合尝试合成,失败时逐级降级。"""
srt = srt_file if use_srt else Path(self._srt_path)
variants: List[tuple[Optional[Path], Optional[Path]]] = [
(music, voiceover),
]
if music and voiceover:
variants.append((None, voiceover))
if music:
variants.append((music, None))
variants.append((None, None))
seen: set[tuple[Optional[str], Optional[str]]] = set()
last_err = ""
for m, v in variants:
key = (str(m) if m else None, str(v) if v else None)
if key in seen:
continue
seen.add(key)
ok, err = _compose_capture_mp4(
ffmpeg_exe,
capture,
srt,
output,
music_file=m,
voiceover_file=v,
warnings=self.warnings,
)
if ok:
return True, ""
last_err = err
if m is not None and v is not None:
self.warnings.append(f"ffmpeg_with_music_and_voice_failed: {err}")
elif m is not None:
self.warnings.append(f"ffmpeg_with_music_failed: {err}")
return False, last_err
async def finalize(self) -> None: async def finalize(self) -> None:
"""停止录制后合成最终 MP4失败只记 warning不抛异常。""" """停止录制后合成最终 MP4失败只记 warning不抛异常。"""
if not self.enabled: if not self.enabled:
@@ -406,7 +820,12 @@ class RpaVideoSession:
self.warnings.append(f"subtitle_write_failed: {exc}") self.warnings.append(f"subtitle_write_failed: {exc}")
music = _resolve_music_file(self.warnings) music = _resolve_music_file(self.warnings)
if music is None and "background_music_lfs_pointer_only" not in self.warnings and "background_music_invalid" not in self.warnings: if music is not None:
self._music_path = str(music.resolve())
elif (
"background_music_lfs_pointer_only" not in self.warnings
and "background_music_invalid" not in self.warnings
):
self.warnings.append("background_music_missing") self.warnings.append("background_music_missing")
ffmpeg_exe = _resolve_ffmpeg_exe() ffmpeg_exe = _resolve_ffmpeg_exe()
@@ -415,25 +834,32 @@ class RpaVideoSession:
self.artifacts = self.summary() self.artifacts = self.summary()
return return
voiceover: Optional[Path] = None
try:
vo = _synthesize_step_voiceover(
entries,
Path(self._voiceover_path),
self.warnings,
ffmpeg_exe=ffmpeg_exe,
)
if vo is not None:
voiceover = vo
self._voiceover_output = str(vo.resolve())
except Exception as exc:
self.warnings.append(f"tts_unexpected_error: {exc}")
srt_file = Path(self._srt_path) srt_file = Path(self._srt_path)
use_srt = srt_file.is_file() and srt_file.stat().st_size > 0 use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
ok, err = _compose_capture_mp4( ok, err = self._try_compose_variants(
ffmpeg_exe, ffmpeg_exe,
capture, capture,
srt_file if use_srt else Path(self._srt_path), srt_file,
Path(self._planned_output), Path(self._planned_output),
music_file=music, music=music,
voiceover=voiceover,
use_srt=use_srt,
) )
if not ok and music is not None:
self.warnings.append(f"ffmpeg_with_music_failed: {err}")
ok, err = _compose_capture_mp4(
ffmpeg_exe,
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: if not ok and not use_srt:
ok, err = _run_ffmpeg( ok, err = _run_ffmpeg(
ffmpeg_exe, ffmpeg_exe,
@@ -469,12 +895,28 @@ class RpaVideoSession:
if self.enabled and self._ffmpeg_record_log_path if self.enabled and self._ffmpeg_record_log_path
else None else None
) )
audio_warnings = [
w
for w in self.warnings
if w.startswith(
(
"tts_",
"video_duration_probe_failed",
"ffmpeg_with_music",
"background_music_",
)
)
or w == "background_music_missing"
]
return { return {
"enabled": self.enabled, "enabled": self.enabled,
"path": self.output_video_path, "path": self.output_video_path,
"capture_path": self.capture_path, "capture_path": self.capture_path,
"record_log_path": record_log, "record_log_path": record_log,
"warnings": list(self.warnings), "warnings": list(self.warnings),
"voiceover_path": self._voiceover_output,
"music_path": self._music_path,
"audio_warnings": audio_warnings,
} }
def to_artifact_dict(self) -> Dict[str, Any]: def to_artifact_dict(self) -> Dict[str, Any]:

View File

@@ -71,9 +71,23 @@ def _resolve_data_root(env: Mapping[str, str]) -> str:
return platform_default_data_root() return platform_default_data_root()
def _parse_version(version: str) -> tuple[int, ...]: def _parse_version(version: str) -> tuple[int, int, int, int]:
"""Parse major.minor.patch[.postN] into (major, minor, patch, post)."""
text = version.strip()
post = 0
if ".post" in text:
base, _, post_part = text.partition(".post")
post_digits = ""
for ch in post_part:
if ch.isdigit():
post_digits += ch
else:
break
post = int(post_digits) if post_digits else 0
text = base
parts: list[int] = [] parts: list[int] = []
for piece in version.strip().split("."): for piece in text.split("."):
digits = "" digits = ""
for ch in piece: for ch in piece:
if ch.isdigit(): if ch.isdigit():
@@ -82,11 +96,15 @@ def _parse_version(version: str) -> tuple[int, ...]:
break break
if digits: if digits:
parts.append(int(digits)) parts.append(int(digits))
return tuple(parts)
while len(parts) < 3:
parts.append(0)
return parts[0], parts[1], parts[2], post
def version_ge(installed: str, required: str) -> bool: def version_ge(installed: str, required: str) -> bool:
"""Compare PEP 440-like versions without external deps (post release suffix ignored).""" """Compare semver-like versions without external deps (supports legacy .postN)."""
return _parse_version(installed) >= _parse_version(required) return _parse_version(installed) >= _parse_version(required)
@@ -281,7 +299,7 @@ def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]:
f"record_video_enabled: {diagnostics.record_video_enabled}", f"record_video_enabled: {diagnostics.record_video_enabled}",
] ]
for issue in diagnostics.issues: for issue in diagnostics.issues:
lines.append(f"runtime_issue[{issue.severity}]: {issue.code} {issue.message}") lines.append(f"runtime_issue[{issue.severity}]: {issue.code} - {issue.message}")
return lines return lines

View File

@@ -1,4 +1,4 @@
"""Text-level checks that reusable-release-skill packages .env.example at package root.""" """Text-level checks for release-related GitHub Actions workflows."""
from __future__ import annotations from __future__ import annotations
@@ -8,17 +8,52 @@ import re
import pytest import pytest
_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
_WORKFLOW_PATH = os.path.join( _SKILL_WORKFLOW_PATH = os.path.join(
_REPO_ROOT, ".github", "workflows", "reusable-release-skill.yaml" _REPO_ROOT, ".github", "workflows", "reusable-release-skill.yaml"
) )
_PUBLISH_WORKFLOW_PATH = os.path.join(
_REPO_ROOT, ".github", "workflows", "publish.yml"
)
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def workflow_text() -> str: def workflow_text() -> str:
with open(_WORKFLOW_PATH, encoding="utf-8") as f: with open(_SKILL_WORKFLOW_PATH, encoding="utf-8") as f:
return f.read() return f.read()
@pytest.fixture(scope="module")
def publish_workflow_text() -> str:
with open(_PUBLISH_WORKFLOW_PATH, encoding="utf-8") as f:
return f.read()
def test_workflow_packages_readme_at_root(workflow_text: str) -> None:
assert "readme_src = os.path.abspath('README.md')" in workflow_text
assert "readme_dst = os.path.join(PACKAGE, 'README.md')" in workflow_text
assert "Copied README.md" in workflow_text
assert (
"raise RuntimeError('README.md exists in skill root but was not packaged')"
in workflow_text
)
def test_workflow_metadata_reads_root_readme(workflow_text: str) -> None:
assert "readme_path = 'README.md'" in workflow_text
assert "frontmatter.load(readme_path)" in workflow_text
assert "metadata['readme_md'] = body" in workflow_text
def test_workflow_does_not_use_references_readme_for_marketplace(
workflow_text: str,
) -> None:
assert "os.path.join('references', 'README.md')" not in workflow_text
assert not re.search(
r"frontmatter\.load\([^)]*references[^)]*README",
workflow_text,
)
def test_workflow_packages_env_example_at_root(workflow_text: str) -> None: def test_workflow_packages_env_example_at_root(workflow_text: str) -> None:
assert ".env.example" in workflow_text assert ".env.example" in workflow_text
assert "Copied .env.example" in workflow_text assert "Copied .env.example" in workflow_text
@@ -40,3 +75,18 @@ def test_workflow_does_not_package_dot_env(workflow_text: str) -> None:
r"os\.path\.join\(PACKAGE,\s*['\"]\.env['\"]\)", r"os\.path\.join\(PACKAGE,\s*['\"]\.env['\"]\)",
workflow_text, workflow_text,
) )
def test_publish_workflow_triggers_on_main_only(publish_workflow_text: str) -> None:
assert "branches:" in publish_workflow_text
assert "- main" in publish_workflow_text
assert "tags:" not in publish_workflow_text
def test_publish_workflow_uses_pyproject_version_only(publish_workflow_text: str) -> None:
assert "ci_set_package_version" not in publish_workflow_text
assert "GITHUB_RUN_NUMBER" not in publish_workflow_text
assert "GITHUB_RUN_ID" not in publish_workflow_text
assert ".post" not in publish_workflow_text
assert "python3.12 -m build" in publish_workflow_text
assert "twine upload" in publish_workflow_text

View File

@@ -25,6 +25,11 @@ def test_version_ge_post_release() -> None:
assert version_ge("1.0.10.post23", "1.0.10") is True assert version_ge("1.0.10.post23", "1.0.10") is True
assert version_ge("1.0.10", "1.0.10") is True assert version_ge("1.0.10", "1.0.10") is True
assert version_ge("1.0.9.post19", "1.0.10") is False assert version_ge("1.0.9.post19", "1.0.10") is False
assert version_ge("1.0.10.post23", "1.0.10.post21") is True
assert version_ge("1.0.10.post21", "1.0.10.post23") is False
assert version_ge("1.0.10", "1.0.10.post1") is False
assert version_ge("1.0.11", "1.0.10.post99") is True
assert version_ge("1.0.11", "1.0.10") is True
def test_is_jiangchang_skill_core_from_skill_tree(tmp_path: Path) -> None: def test_is_jiangchang_skill_core_from_skill_tree(tmp_path: Path) -> None:
@@ -311,3 +316,32 @@ def test_runtime_diagnostics_frozen_dataclass() -> None:
) )
assert diag.has_fatal_issues is False assert diag.has_fatal_issues is False
assert diag.issue_codes() == ["ffmpeg_unavailable"] assert diag.issue_codes() == ["ffmpeg_unavailable"]
def test_format_runtime_health_lines_issue_separator_ascii() -> None:
diag = RuntimeDiagnostics(
skill_slug="x",
python_executable="python",
platform_kit_version="1.0.11",
platform_kit_min_version=None,
platform_kit_version_ok=None,
jiangchang_skill_core_file=None,
claw_data_root=None,
jiangchang_data_root=None,
resolved_data_root="/tmp",
media_assets_root="/tmp/media",
ffmpeg_available=False,
ffmpeg_path=None,
background_music_mp3_count=0,
background_music_usable_count=0,
background_music_issue=None,
background_music_sample_path=None,
record_video_enabled=False,
issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
)
lines = format_runtime_health_lines(diag)
issue_lines = [line for line in lines if line.startswith("runtime_issue[")]
assert len(issue_lines) == 1
assert " - " in issue_lines[0]
assert "\u2014" not in issue_lines[0]
assert "\u2013" not in issue_lines[0]

View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""RpaVideoSessionffmpeg 桌面录屏。""" """RpaVideoSessionffmpeg 桌面录屏、背景音乐循环淡出、旁白混音"""
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
@@ -8,9 +8,21 @@ import sys
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from jiangchang_skill_core import config from jiangchang_skill_core import config
from jiangchang_skill_core.rpa.video_session import RpaVideoSession from jiangchang_skill_core.rpa import video_session as vs
from jiangchang_skill_core.rpa.video_session import (
RpaVideoSession,
_StepEntry,
_INTRO_BUFFER_SECONDS,
_INTRO_STABILIZE_SECONDS,
_OUTRO_BUFFER_SECONDS,
_compose_capture_mp4,
_music_fade_params,
_select_voiceover_clips,
_voiceover_text_for_step,
)
class TestRpaVideoSession(unittest.TestCase): class TestRpaVideoSession(unittest.TestCase):
@@ -33,6 +45,8 @@ class TestRpaVideoSession(unittest.TestCase):
self.assertIsNone(summ["path"]) self.assertIsNone(summ["path"])
self.assertIsNone(summ["capture_path"]) self.assertIsNone(summ["capture_path"])
self.assertIsNone(summ["record_log_path"]) self.assertIsNone(summ["record_log_path"])
self.assertIsNone(summ["voiceover_path"])
self.assertIsNone(summ["music_path"])
finally: finally:
if old is not None: if old is not None:
os.environ["OPENCLAW_RECORD_VIDEO"] = old os.environ["OPENCLAW_RECORD_VIDEO"] = old
@@ -60,6 +74,11 @@ class TestRpaVideoSession(unittest.TestCase):
"/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log" "/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log"
) )
) )
self.assertTrue(
session._voiceover_path.replace("\\", "/").endswith(
"/rpa-artifacts/exmail_20260602_120000/voiceover/voiceover.wav"
)
)
self.assertTrue( self.assertTrue(
session._planned_output.replace("\\", "/").startswith( session._planned_output.replace("\\", "/").startswith(
tmp.replace("\\", "/") + "/videos/" tmp.replace("\\", "/") + "/videos/"
@@ -144,5 +163,413 @@ class TestRpaVideoSession(unittest.TestCase):
config.reset_cache() config.reset_cache()
class TestMusicFadeParams(unittest.TestCase):
def test_normal_duration(self) -> None:
start, dur = _music_fade_params(20.0)
self.assertAlmostEqual(start, 17.0)
self.assertAlmostEqual(dur, 3.0)
def test_short_video(self) -> None:
start, dur = _music_fade_params(4.0)
self.assertAlmostEqual(dur, 1.0)
self.assertAlmostEqual(start, 3.0)
def test_invalid_duration(self) -> None:
self.assertEqual(_music_fade_params(None), (None, None))
self.assertEqual(_music_fade_params(0), (None, None))
self.assertEqual(_music_fade_params(-1), (None, None))
class TestVoiceoverTextFilter(unittest.TestCase):
def test_keeps_meaningful_steps(self) -> None:
for text in (
"开始采集",
"获取1688账号",
"启动浏览器",
"检查登录状态",
"搜索关键词:螺丝",
"解析第 1 页店铺列表",
"打开店铺联系方式页",
"提取联系人信息",
"采集完成",
):
self.assertIsNotNone(_voiceover_text_for_step(text), msg=text)
def test_skips_noise(self) -> None:
for text in (
"等待 1-5s",
"跳过重复店铺",
"写入联系人库",
"探针模式",
"操作异常",
"登录失败",
"2024-01-01 12:00:00 INFO: debug",
"",
" ",
):
self.assertIsNone(_voiceover_text_for_step(text), msg=text)
def test_truncates_long_text(self) -> None:
long_text = "" * 80
result = _voiceover_text_for_step(long_text)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual(len(result), 60)
def test_dedup_within_ten_seconds(self) -> None:
entries = [
_StepEntry(0.0, "启动浏览器"),
_StepEntry(5.0, "启动浏览器"),
_StepEntry(12.0, "启动浏览器"),
]
clips = _select_voiceover_clips(entries)
texts = [t for _, t in clips]
self.assertEqual(texts.count("启动浏览器"), 2)
def test_min_gap_two_seconds(self) -> None:
entries = [
_StepEntry(0.0, "获取1688账号"),
_StepEntry(1.0, "连接CDP"),
_StepEntry(3.0, "检查登录状态"),
]
clips = _select_voiceover_clips(entries)
self.assertEqual(len(clips), 2)
self.assertEqual(clips[0][1], "获取1688账号")
self.assertEqual(clips[1][1], "检查登录状态")
def test_important_bypasses_min_gap(self) -> None:
entries = [
_StepEntry(0.0, "定位搜索框"),
_StepEntry(0.5, "输入关键词:宁德电池"),
_StepEntry(1.0, "点击搜索"),
_StepEntry(1.5, "等待搜索结果"),
]
clips = _select_voiceover_clips(entries)
texts = [t for _, t in clips]
self.assertEqual(len(clips), 4)
self.assertIn("输入关键词:宁德电池", texts)
self.assertIn("点击搜索", texts)
def test_noise_still_skipped_with_important_nearby(self) -> None:
entries = [
_StepEntry(0.0, "点击搜索"),
_StepEntry(0.5, "跳过重复店铺"),
_StepEntry(1.0, "写入联系人库"),
]
clips = _select_voiceover_clips(entries)
texts = [t for _, t in clips]
self.assertEqual(texts, ["点击搜索"])
class TestComposeCaptureMp4(unittest.TestCase):
def setUp(self) -> None:
self.ffmpeg = Path(r"C:\ffmpeg\ffmpeg.exe")
self.capture = Path(r"C:\tmp\capture.mp4")
self.srt = Path(r"C:\tmp\sub.srt")
self.output = Path(r"C:\tmp\out.mp4")
self.music = Path(r"C:\tmp\music.mp3")
self.voice = Path(r"C:\tmp\voiceover.wav")
def _run_compose(self, **kwargs) -> list[str]:
captured: list[str] = []
def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
captured.extend(cmd_tail)
return True, ""
with (
patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
patch.object(vs, "_resolve_ffprobe_exe", return_value=Path(r"C:\ffmpeg\ffprobe.exe")),
patch.object(vs, "_probe_media_duration_seconds", return_value=kwargs.pop("duration", 20.0)),
patch.object(Path, "is_file", return_value=True),
):
_compose_capture_mp4(
self.ffmpeg,
self.capture,
self.srt,
self.output,
music_file=kwargs.get("music_file"),
voiceover_file=kwargs.get("voiceover_file"),
warnings=kwargs.get("warnings"),
)
return captured
def test_music_loop_params(self) -> None:
cmd = self._run_compose(music_file=self.music)
cmd_str = " ".join(cmd)
loop_idx = cmd.index("-stream_loop")
minus_one_idx = cmd.index("-1", loop_idx)
music_i_idx = cmd.index("-i", minus_one_idx)
self.assertEqual(cmd[music_i_idx + 1], str(self.music))
self.assertIn("-shortest", cmd)
filter_idx = cmd.index("-filter_complex")
filt = cmd[filter_idx + 1]
self.assertNotIn("apad", filt)
def test_music_fade_filter_long_video(self) -> None:
cmd = self._run_compose(music_file=self.music, duration=20.0)
filt = cmd[cmd.index("-filter_complex") + 1]
self.assertIn("afade=t=out:st=17", filt)
self.assertIn("d=3", filt)
self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
def test_music_fade_filter_short_video(self) -> None:
cmd = self._run_compose(music_file=self.music, duration=4.0)
filt = cmd[cmd.index("-filter_complex") + 1]
self.assertIn("afade=t=out:st=3", filt)
self.assertIn("d=1", filt)
def test_duration_probe_failed_no_afade(self) -> None:
warnings: list[str] = []
captured: list[str] = []
def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
captured.extend(cmd_tail)
return True, ""
with (
patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
patch.object(vs, "_resolve_ffprobe_exe", return_value=None),
patch.object(vs, "_probe_media_duration_seconds", return_value=None),
patch.object(Path, "is_file", return_value=True),
):
ok, _ = _compose_capture_mp4(
self.ffmpeg,
self.capture,
self.srt,
self.output,
music_file=self.music,
warnings=warnings,
)
self.assertTrue(ok)
self.assertIn("video_duration_probe_failed_for_music_fade", warnings)
filt = captured[captured.index("-filter_complex") + 1]
self.assertNotIn("afade", filt)
self.assertIn("-stream_loop", captured)
def test_music_and_voice_amix(self) -> None:
cmd = self._run_compose(music_file=self.music, voiceover_file=self.voice)
filt = cmd[cmd.index("-filter_complex") + 1]
self.assertIn("amix", filt)
self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
self.assertLess(vs._MUSIC_VOLUME, vs._VOICE_VOLUME)
self.assertIn("afade", filt)
self.assertNotIn("apad", filt)
self.assertIn("-stream_loop", cmd)
self.assertIn("-shortest", cmd)
def test_voice_only_no_amix(self) -> None:
cmd = self._run_compose(voiceover_file=self.voice)
filt = cmd[cmd.index("-filter_complex") + 1]
self.assertNotIn("amix", filt)
self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
self.assertIn("apad", filt)
self.assertIn("-shortest", cmd)
self.assertNotIn("-stream_loop", cmd)
class TestVideoSessionLifecycle(unittest.TestCase):
def setUp(self) -> None:
config.reset_cache()
def tearDown(self) -> None:
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
config.reset_cache()
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
@patch.object(RpaVideoSession, "_start_ffmpeg_capture")
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
def test_enter_starts_capture_before_title(
self,
mock_finalize: AsyncMock,
mock_stop: MagicMock,
mock_start: MagicMock,
mock_sleep: AsyncMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
call_order: list[str] = []
mock_start.side_effect = lambda: call_order.append("start_capture")
with tempfile.TemporaryDirectory() as tmp:
config.reset_cache()
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch_intro",
title="开始执行测试",
)
original_add = session.add_step
def _track_add(text: str, *, duration: float = 4.0) -> None:
call_order.append(f"add_step:{text}")
original_add(text, duration=duration)
with patch.object(session, "add_step", side_effect=_track_add):
asyncio.run(session.__aenter__())
self.assertEqual(call_order[0], "start_capture")
self.assertIn("add_step:开始执行测试", call_order)
mock_sleep.assert_awaited()
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
@patch.object(RpaVideoSession, "_start_ffmpeg_capture")
def test_enter_no_title_still_buffers(
self,
mock_start: MagicMock,
mock_sleep: AsyncMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
with tempfile.TemporaryDirectory() as tmp:
config.reset_cache()
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch_no_title",
)
with patch.object(session, "add_step") as mock_add:
asyncio.run(session.__aenter__())
mock_start.assert_called_once()
mock_add.assert_not_called()
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
def test_exit_closing_title_before_stop(
self,
mock_finalize: AsyncMock,
mock_stop: MagicMock,
mock_sleep: AsyncMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
call_order: list[str] = []
mock_stop.side_effect = lambda: call_order.append("stop_capture")
with tempfile.TemporaryDirectory() as tmp:
config.reset_cache()
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch_outro",
closing_title="任务已完成",
)
original_add = session.add_step
def _track_add(text: str, *, duration: float = 4.0) -> None:
call_order.append(f"add_step:{text}")
original_add(text, duration=duration)
with patch.object(session, "add_step", side_effect=_track_add):
asyncio.run(session.__aexit__(None, None, None))
self.assertIn("add_step:任务已完成", call_order)
self.assertEqual(call_order[-1], "stop_capture")
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
def test_exit_no_closing_title_still_outro_buffer(
self,
mock_finalize: AsyncMock,
mock_stop: MagicMock,
mock_sleep: AsyncMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
with tempfile.TemporaryDirectory() as tmp:
config.reset_cache()
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch_outro_empty",
)
with patch.object(session, "add_step") as mock_add:
asyncio.run(session.__aexit__(None, None, None))
mock_add.assert_not_called()
mock_stop.assert_called_once()
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
class TestFinalizeDegradation(unittest.TestCase):
def setUp(self) -> None:
config.reset_cache()
def tearDown(self) -> None:
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
config.reset_cache()
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
@patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
@patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
@patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
def test_tts_failure_still_composes(
self,
mock_compose: MagicMock,
mock_tts: MagicMock,
mock_music: MagicMock,
mock_ffmpeg: MagicMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
mock_music.return_value = Path(r"C:\tmp\music.mp3")
mock_tts.return_value = None
mock_compose.return_value = (True, "")
with tempfile.TemporaryDirectory() as tmp:
capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
capture.parent.mkdir(parents=True)
capture.write_bytes(b"fake")
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch",
)
session.add_step("启动浏览器")
asyncio.run(session.finalize())
self.assertTrue(mock_compose.called)
self.assertIsNotNone(session.output_video_path)
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
@patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
@patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
@patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
def test_compose_fallback_without_music(
self,
mock_compose: MagicMock,
mock_tts: MagicMock,
mock_music: MagicMock,
mock_ffmpeg: MagicMock,
) -> None:
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
mock_music.return_value = Path(r"C:\tmp\music.mp3")
mock_tts.return_value = None
mock_compose.side_effect = [(False, "music err"), (True, "")]
with tempfile.TemporaryDirectory() as tmp:
capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
capture.parent.mkdir(parents=True)
capture.write_bytes(b"fake")
session = RpaVideoSession(
skill_slug="test",
skill_data_dir=tmp,
batch_id="batch",
)
asyncio.run(session.finalize())
self.assertEqual(mock_compose.call_count, 2)
self.assertTrue(any("ffmpeg_with_music_failed" in w for w in session.warnings))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@@ -1,79 +0,0 @@
"""Patch pyproject.toml and __init__.py version for CI builds (not committed)."""
from __future__ import annotations
import os
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
def _read_base_version() -> str:
text = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
match = re.search(r'^version\s*=\s*"([^"]+)"', text, re.MULTILINE)
if not match:
raise SystemExit("ERROR: version not found in pyproject.toml")
base = match.group(1)
return re.sub(r"\.post\d+$", "", base)
def resolve_version() -> str:
ref = os.environ.get("GITHUB_REF", "")
if ref.startswith("refs/tags/v"):
return ref.removeprefix("refs/tags/v")
if ref.startswith("refs/tags/"):
return ref.removeprefix("refs/tags/")
run_number = (
os.environ.get("GITHUB_RUN_NUMBER")
or os.environ.get("GITHUB_RUN_ID")
or "0"
)
base = _read_base_version()
return f"{base}.post{run_number}"
def patch_version(version: str) -> None:
pyproject = ROOT / "pyproject.toml"
py_text = pyproject.read_text(encoding="utf-8")
py_text, n = re.subn(
r'^version\s*=\s*"[^"]*"',
f'version = "{version}"',
py_text,
count=1,
flags=re.MULTILINE,
)
if n != 1:
raise SystemExit("ERROR: failed to patch version in pyproject.toml")
pyproject.write_text(py_text, encoding="utf-8")
init_py = ROOT / "src" / "jiangchang_desktop_sdk" / "__init__.py"
init_text = init_py.read_text(encoding="utf-8")
init_text, n = re.subn(
r'^__version__\s*=\s*"[^"]*"',
f'__version__ = "{version}"',
init_text,
count=1,
flags=re.MULTILINE,
)
if n != 1:
raise SystemExit("ERROR: failed to patch __version__ in __init__.py")
init_py.write_text(init_text, encoding="utf-8")
def main() -> None:
version = resolve_version()
print(f"Resolved package version: {version}")
patch_version(version)
if os.environ.get("GITHUB_OUTPUT"):
with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as fh:
fh.write(f"package_version={version}\n")
if __name__ == "__main__":
try:
main()
except SystemExit as exc:
print(exc, file=sys.stderr)
raise

View File

@@ -28,6 +28,7 @@
加密与 ZIP 内容由 CI 工作流 reusable-release-skill.yaml 的「Encrypt Source Code」步骤执行。CI 会: 加密与 ZIP 内容由 CI 工作流 reusable-release-skill.yaml 的「Encrypt Source Code」步骤执行。CI 会:
- 加密 scripts/(递归 PyArmor -r输出到包内 scripts/,与源码目录树一致) - 加密 scripts/(递归 PyArmor -r输出到包内 scripts/,与源码目录树一致)
- 复制 SKILL.md - 复制 SKILL.md
- 若根目录存在 README.md则复制到包根明文用于技能市场用户说明
- 复制 references/(排除 REQUIREMENTS.md - 复制 references/(排除 REQUIREMENTS.md
- 复制 assets/ - 复制 assets/
- 复制 tests/ - 复制 tests/