30 lines
1.1 KiB
Python
30 lines
1.1 KiB
Python
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
|
|
"""RPA 失败存证路径与截图。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
|
|
def _artifacts_enabled() -> bool:
|
|
v = (os.environ.get("OPENCLAW_ARTIFACTS_ON_FAILURE") or "1").strip().lower()
|
|
return v not in ("0", "false", "no", "off")
|
|
|
|
|
|
def artifact_path(data_dir: str, batch_id: str, tag: str, ext: str = "png") -> str:
|
|
"""返回 {data_dir}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.{ext}"""
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
directory = os.path.join(data_dir, "rpa-artifacts", batch_id)
|
|
os.makedirs(directory, exist_ok=True)
|
|
return os.path.join(directory, f"{tag}_{ts}.{ext}")
|
|
|
|
|
|
async def capture_failure(page, data_dir: str, batch_id: str, tag: str) -> str:
|
|
"""受 OPENCLAW_ARTIFACTS_ON_FAILURE 控制,截图并返回路径;禁用时返回空字符串。"""
|
|
if not _artifacts_enabled():
|
|
return ""
|
|
path = artifact_path(data_dir, batch_id, tag, ext="png")
|
|
await page.screenshot(path=path, full_page=True)
|
|
return path
|