feat(sites): auto-screenshot on final download failure for debugging

所有站点 with_retry 在最后一次重试失败、reset 之前自动截图,
保存到 logs/screenshots/。网页站点走 Playwright page.screenshot(),
安能走 CDP Page.captureScreenshot。截图失败绝不阻塞任务流程。

- paths.py: 新增 SCREENSHOT_DIR (BASE_DIR/logs/screenshots/)
- runtime.py: 新增 capture_error_screenshot() 工具函数
- .gitignore: 新增 logs/ 忽略规则

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-08-02 11:18:21 +08:00
parent 53c71aeeac
commit 3c32720985
8 changed files with 101 additions and 18 deletions

View File

@@ -43,10 +43,37 @@ from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
from inbound_verify import state_store
def with_retry(site_name, label, flow, reset, max_attempts=3):
def _capture_error_screenshot(site, kind, attempt, error):
"""安能 CDP 错误截图best-effort失败仅告警绝不外抛"""
try:
import base64, os
from datetime import datetime
from inbound_verify.paths import SCREENSHOT_DIR
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
pages = list_pages()
if not pages:
return
cdp = CDP(pages[0]["webSocketDebuggerUrl"])
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
err_short = (error or "unknown")[:40].replace("/", "_").replace("\\", "_")
fname = f"{site}_{kind}_{ts}_attempt{attempt}_{err_short}.png"
path = os.path.join(SCREENSHOT_DIR, fname)
result = cdp.call("Page.captureScreenshot", format="png")
with open(path, "wb") as f:
f.write(base64.b64decode(result["data"]))
cdp.close()
print(f"📸 【{site}-{kind}】错误截图已保存: {path}")
except Exception as se:
print(f"📸 【{site}-{kind}】截图失败(不影响任务): {se}")
def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
"""异常兜底flow 失败 → 重置回初始态 → 重试,最多 max_attempts 次(含首次)。
每次失败都重置(含最终放弃那一次):既是重试前的清场,也保证最终放弃时环境干净。
最后一次失败时(重置前)自动截图保存到 logs/screenshots/,供问题排查。
(安能通过 CDP 截图page 参数忽略;保留为统一签名兼容。)
flow 为零参可调用;返回 False 视为失败,其余视为成功。
返回 True=最终成功False=重试耗尽放弃(供调度层判断任务成败)。
"""
@@ -60,6 +87,11 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
return True
except Exception as e:
print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
if attempt == max_attempts:
try:
_capture_error_screenshot(site_name, label, attempt, str(e))
except Exception:
pass
print(f" → 重置【{site_name}】到初始态,清理环境 ...")
try:
reset()