85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""任务运行辅助：录屏信息组装与 CLI 输出（通用模板，无业务逻辑）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, Optional
|
||
|
||
|
||
def build_video_info(
|
||
video_summary: dict[str, Any],
|
||
compose_context_before: Optional[dict[str, Any]] = None,
|
||
compose_context_after: Optional[dict[str, Any]] = None,
|
||
) -> dict[str, Any]:
|
||
"""将 RpaVideoSession.summary() 转为 finish() / CLI 可用的 video 块。"""
|
||
return {
|
||
"enabled": video_summary.get("enabled", False),
|
||
"path": video_summary.get("path"),
|
||
"capture_path": video_summary.get("capture_path"),
|
||
"record_log_path": video_summary.get("record_log_path"),
|
||
"warnings": list(video_summary.get("warnings") or []),
|
||
"voiceover_path": video_summary.get("voiceover_path"),
|
||
"music_path": video_summary.get("music_path"),
|
||
"audio_warnings": list(video_summary.get("audio_warnings") or []),
|
||
"compose_context": {
|
||
**(compose_context_before or {}),
|
||
"after_compose": compose_context_after or {},
|
||
},
|
||
}
|
||
|
||
|
||
def merge_video_into_result_summary(
|
||
payload: dict[str, Any],
|
||
video_info: Optional[dict[str, Any]],
|
||
) -> dict[str, Any]:
|
||
"""把 video artifact 字段写入 finish() 结果对象（video / video_path 等）。"""
|
||
if not video_info:
|
||
return payload
|
||
payload = dict(payload)
|
||
payload["video"] = video_info
|
||
payload["video_path"] = video_info.get("path")
|
||
payload["raw_video"] = video_info.get("capture_path")
|
||
payload["video_log"] = video_info.get("record_log_path")
|
||
payload["video_warnings"] = list(video_info.get("warnings") or [])
|
||
payload["voiceover_path"] = video_info.get("voiceover_path")
|
||
payload["music_path"] = video_info.get("music_path")
|
||
payload["audio_warnings"] = list(video_info.get("audio_warnings") or [])
|
||
return payload
|
||
|
||
|
||
def _print_video_summary(video_block: Optional[dict[str, Any]]) -> None:
|
||
"""任务完成后在 CLI 打印录屏路径、日志与诊断信息。"""
|
||
if not isinstance(video_block, dict) or not video_block.get("enabled"):
|
||
return
|
||
record_log = video_block.get("record_log_path")
|
||
if record_log:
|
||
print(f"录屏日志：{record_log}")
|
||
path = video_block.get("path")
|
||
if path:
|
||
print(f"录屏路径：{path}")
|
||
else:
|
||
print("录屏路径：未生成")
|
||
warnings = video_block.get("warnings") or []
|
||
if warnings:
|
||
print(f"视频诊断：{'; '.join(str(w) for w in warnings)}")
|
||
capture = video_block.get("capture_path")
|
||
if capture:
|
||
print(f"原始录屏路径：{capture}")
|
||
voiceover = video_block.get("voiceover_path")
|
||
if voiceover:
|
||
print(f"旁白路径：{voiceover}")
|
||
music = video_block.get("music_path")
|
||
if music:
|
||
print(f"背景音乐：{music}")
|
||
audio_warnings = video_block.get("audio_warnings") or []
|
||
if audio_warnings:
|
||
print(f"音频诊断：{'; '.join(str(w) for w in audio_warnings)}")
|
||
compose_ctx = video_block.get("compose_context")
|
||
if isinstance(compose_ctx, dict):
|
||
print(
|
||
"视频合成上下文："
|
||
f"media_assets_root={compose_ctx.get('media_assets_root', '')}; "
|
||
f"ffmpeg_path={compose_ctx.get('ffmpeg_path', '')}; "
|
||
f"background_music_issue={compose_ctx.get('background_music_issue') or ''}; "
|
||
f"background_music_sample_path={compose_ctx.get('background_music_sample_path', '')}"
|
||
)
|