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:
- 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
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.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('assets', os.path.join(PACKAGE, 'assets'), 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_description = metadata.get('description')
metadata['readme_md'] = skill_readme_md
readme_path = os.path.join('references', 'README.md')
readme_path = 'README.md'
if os.path.isfile(readme_path):
try:
readme_post = frontmatter.load(readme_path)

View File

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

View File

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

View File

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

View File

@@ -1,9 +1,14 @@
"""RPA 运行录屏会话ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
"""RPA 运行录屏会话ffmpeg 桌面录制 → 字幕 + 旁白 + 背景音乐 → 最终 MP4。
背景音乐默认循环铺满录屏时长并在结尾淡出;旁白使用 Windows 本地 SAPI 合成,
失败时静默降级不影响视频生成。Skill 无需额外配置。
"""
from __future__ import annotations
import asyncio
import os
import re
import subprocess
import sys
import time
@@ -18,6 +23,7 @@ from ..media_assets import (
ensure_media_assets,
pick_background_music,
resolve_ffmpeg,
resolve_ffprobe,
)
_JIANGCHANG_SUBTITLE_STYLE = (
@@ -28,6 +34,54 @@ _JIANGCHANG_SUBTITLE_STYLE = (
_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:
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
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]:
music = pick_background_music()
if music is not None:
@@ -59,6 +175,183 @@ class _StepEntry:
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:
MIN_DURATION = 1.0
GAP = 0.1
@@ -117,27 +410,16 @@ def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: List[str]) -> tuple[bool, str]:
return True, ""
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,
)
def _build_music_filter(
audio_input_idx: int,
fade_start: Optional[float],
fade_duration: Optional[float],
output_label: str = "music",
) -> str:
filt = f"[{audio_input_idx}:a]volume={_MUSIC_VOLUME}"
if fade_start is not None and fade_duration is not None:
filt += f",afade=t=out:st={fade_start}:d={fade_duration}"
return filt + f"[{output_label}]"
def _compose_capture_mp4(
@@ -147,7 +429,8 @@ def _compose_capture_mp4(
output_path: Path,
*,
music_file: Optional[Path] = None,
music_volume: float = 0.15,
voiceover_file: Optional[Path] = None,
warnings: Optional[List[str]] = None,
) -> tuple[bool, str]:
output_path.parent.mkdir(parents=True, exist_ok=True)
srt_esc = _escape_srt_path(str(srt_path.resolve()))
@@ -156,19 +439,57 @@ def _compose_capture_mp4(
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
)
if music_file and music_file.is_file():
filter_complex = (
f"[0:v]{subtitle_filter}[vout];"
f"[1:a]volume={music_volume},apad[aout]"
has_music = music_file is not None and music_file.is_file()
has_voice = voiceover_file is not None and voiceover_file.is_file()
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)
)
cmd_tail = [
"-y",
"-i",
str(capture_path),
"-i",
str(music_file),
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",
filter_complex,
";".join(filter_parts),
"-map",
"[vout]",
"-map",
@@ -188,6 +509,7 @@ def _compose_capture_mp4(
"yuv420p",
str(output_path),
]
)
else:
cmd_tail = [
"-y",
@@ -212,6 +534,29 @@ def _compose_capture_mp4(
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:
"""Service/RPA 层统一 ffmpeg 桌面录屏OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
@@ -222,11 +567,13 @@ class RpaVideoSession:
skill_data_dir: str,
batch_id: str,
title: str = "",
closing_title: str = "",
) -> None:
self.skill_slug = skill_slug
self.skill_data_dir = os.path.abspath(skill_data_dir)
self.batch_id = batch_id
self.title = title
self.closing_title = closing_title
self.enabled = _record_video_enabled()
self.warnings: List[str] = []
self.output_video_path: Optional[str] = None
@@ -242,12 +589,17 @@ class RpaVideoSession:
self.skill_data_dir, "rpa-artifacts", batch_id
)
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._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._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._music_path: Optional[str] = None
self._voiceover_output: Optional[str] = None
self._steps: List[_StepEntry] = []
self._t0: Optional[float] = None
self._ffmpeg_proc: Optional[subprocess.Popen] = None
@@ -256,6 +608,7 @@ class RpaVideoSession:
if self.enabled:
os.makedirs(self._videos_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._artifact_root, exist_ok=True)
@@ -366,19 +719,80 @@ class RpaVideoSession:
self._close_ffmpeg_record_log()
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":
if self.enabled:
self._t0 = time.monotonic()
if self.title:
self.add_step(self.title, duration=3.0)
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
async def __aexit__(self, exc_type, exc, tb) -> None:
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()
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:
"""停止录制后合成最终 MP4失败只记 warning不抛异常。"""
if not self.enabled:
@@ -406,7 +820,12 @@ class RpaVideoSession:
self.warnings.append(f"subtitle_write_failed: {exc}")
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")
ffmpeg_exe = _resolve_ffmpeg_exe()
@@ -415,24 +834,31 @@ class RpaVideoSession:
self.artifacts = self.summary()
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)
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,
capture,
srt_file if use_srt else Path(self._srt_path),
srt_file,
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_capture_mp4(
ffmpeg_exe,
capture,
srt_file if use_srt else Path(self._srt_path),
Path(self._planned_output),
music_file=None,
music=music,
voiceover=voiceover,
use_srt=use_srt,
)
if not ok and not use_srt:
ok, err = _run_ffmpeg(
@@ -469,12 +895,28 @@ class RpaVideoSession:
if self.enabled and self._ffmpeg_record_log_path
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 {
"enabled": self.enabled,
"path": self.output_video_path,
"capture_path": self.capture_path,
"record_log_path": record_log,
"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]:

View File

@@ -71,9 +71,23 @@ def _resolve_data_root(env: Mapping[str, str]) -> str:
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] = []
for piece in version.strip().split("."):
for piece in text.split("."):
digits = ""
for ch in piece:
if ch.isdigit():
@@ -82,11 +96,15 @@ def _parse_version(version: str) -> tuple[int, ...]:
break
if 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:
"""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)
@@ -281,7 +299,7 @@ def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]:
f"record_video_enabled: {diagnostics.record_video_enabled}",
]
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

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
@@ -8,17 +8,52 @@ import re
import pytest
_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"
)
_PUBLISH_WORKFLOW_PATH = os.path.join(
_REPO_ROOT, ".github", "workflows", "publish.yml"
)
@pytest.fixture(scope="module")
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()
@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:
assert ".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['\"]\)",
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", "1.0.10") is True
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:
@@ -311,3 +316,32 @@ def test_runtime_diagnostics_frozen_dataclass() -> None:
)
assert diag.has_fatal_issues is False
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 -*-
"""RpaVideoSessionffmpeg 桌面录屏。"""
"""RpaVideoSessionffmpeg 桌面录屏、背景音乐循环淡出、旁白混音"""
from __future__ import annotations
import asyncio
@@ -8,9 +8,21 @@ import sys
import tempfile
import unittest
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.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):
@@ -33,6 +45,8 @@ class TestRpaVideoSession(unittest.TestCase):
self.assertIsNone(summ["path"])
self.assertIsNone(summ["capture_path"])
self.assertIsNone(summ["record_log_path"])
self.assertIsNone(summ["voiceover_path"])
self.assertIsNone(summ["music_path"])
finally:
if old is not None:
os.environ["OPENCLAW_RECORD_VIDEO"] = old
@@ -60,6 +74,11 @@ class TestRpaVideoSession(unittest.TestCase):
"/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(
session._planned_output.replace("\\", "/").startswith(
tmp.replace("\\", "/") + "/videos/"
@@ -144,5 +163,413 @@ class TestRpaVideoSession(unittest.TestCase):
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__":
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 会:
- 加密 scripts/(递归 PyArmor -r输出到包内 scripts/,与源码目录树一致)
- 复制 SKILL.md
- 若根目录存在 README.md则复制到包根明文用于技能市场用户说明
- 复制 references/(排除 REQUIREMENTS.md
- 复制 assets/
- 复制 tests/