feat(custom):新增 screenshot_on_fail.py节点级截图sink

Storage/customs/BQ/screenshot-on-fail/README.md

@@ -0,0 +1,49 @@
+# Screenshot On Fail
+
+节点级截图 Sink，每个识别节点完成时自动截图保存，用于调试和排查问题。
+
+## 设计动机
+
+在 MaaFramework 开发中，部分开发者会使用 `on_error` 来承载业务逻辑（而非仅用于错误处理），这会导致 Pipeline 节点失败时无法触发默认的截图保存行为。当用户提交 bug 反馈时，开发者往往拿不到失败时的运行截图，只能靠日志猜测问题，排查效率极低。
+
+本 Sink 绕过 `on_error` 机制，直接监听每个 Pipeline 节点的识别完成事件，**无论节点成功还是失败都会截图**，确保开发者始终能拿到完整的运行过程截图，大幅提升问题排查效率。
+
+## 功能
+
+- 每个 Pipeline 节点识别完成时自动截图保存（无论成功或失败）
+- 截图保存为 JPG 格式（如 cv2 不可用则回退为 BMP）
+- 截图文件名包含时间戳、节点名、识别 ID 和状态信息
+- 最多保留 300 张截图，环形覆盖旧图，避免磁盘空间无限增长
+
+## 截图保存路径
+
+```
+debug/screenshots/{年}.{月}.{日}-{时}.{分}.{秒}.{毫秒}_{节点名}_{识别ID}_{failed|success}.jpg
+```
+
+可通过环境变量 `MDNA_DEBUG_DIR` 自定义 debug 目录路径。
+
+## 文件
+
+- `maahub_meta.json`: 组件元数据
+- `README.md`: 使用说明
+- `main.py`: 入口文件，导入 Sink 模块即可自动注册
+- `screenshot_on_fail.py`: Sink 核心实现
+- `pipeline.json`: 空 Pipeline（Sink 组件无需 Pipeline 配置）
+
+## 使用方法
+
+1. 将本文件夹复制到你的 MaaFramework 项目的 `agent/custom/sink/` 目录下
+2. 确保已安装依赖：`pip install numpy opencv-python`
+3. 在你的 `main.py` 中导入模块：
+
+```python
+from agent.custom.sink.screenshot_on_fail import NodeScreenshotSink
+```
+
+或者直接运行本文件夹的 `main.py` 作为入口。
+
+## 依赖
+
+- `numpy`
+- `opencv-python`（可选，不可用时自动回退 BMP 格式）

Storage/customs/BQ/screenshot-on-fail/maahub_meta.json

@@ -0,0 +1,20 @@
+{
+  "id": "BQ/screenshot-on-fail",
+  "title": "Screenshot On Fail",
+  "description": "节点级截图 Sink，每个识别节点完成时自动截图保存，用于调试和排查问题。截图保存为 JPG 格式，最多保留 300 张(可在screenshot_on_fail.py中修改)，环形覆盖旧图。",
+  "author": "BQ",
+  "source": "MDNA",
+  "sourceGithub": "https://github.com/BQOvO/MDNA",
+  "tags": ["custom", "sink", "screenshot", "debug", "utility", "data"],
+  "createdAt": "2026-08-25",
+  "updatedAt": "2026-08-25",
+  "version": "1.0.0",
+  "mfwVersion": "5.12.1",
+  "entry": "main.py",
+  "readme": "./README.md",
+  "status": "stable",
+  "type": "custom",
+  "language": "python",
+  "runtime": "python 3.12.9",
+  "dependencies": ["numpy", "opencv-python"]
+}

Storage/customs/BQ/screenshot-on-fail/main.py

@@ -0,0 +1,9 @@
+import screenshot_on_fail  # noqa: F401 - 导入即注册 Sink，无需显式引用
+
+
+def main():
+    print("ScreenshotOnFail Sink 已加载")
+
+
+if __name__ == "__main__":
+    main()

Storage/customs/BQ/screenshot-on-fail/pipeline.json

@@ -0,0 +1 @@
+{}

Storage/customs/BQ/screenshot-on-fail/screenshot_on_fail.py

@@ -0,0 +1,166 @@
+"""
+节点级截图 Sink，每个识别节点完成时自动截图保存。
+
+截图保存到: debug/screenshots/{年}.{月}.{日}-{时}.{分}.{秒}.{毫秒}_{节点名}_{识别ID}_{failed|success}.jpg
+最多保留 300 张，环形覆盖旧图。
+
+使用 ContextEventSink（节点级），每个节点识别后直接拿图保存为 JPG。
+"""
+
+import logging
+import os
+import struct
+from datetime import datetime
+from pathlib import Path
+
+import numpy as np
+
+from maa.agent.agent_server import AgentServer
+from maa.context import Context, ContextEventSink
+from maa.event_sink import NotificationType
+
+_SCREENSHOT_DIR = os.environ.get(
+    "MDNA_DEBUG_DIR",
+    str(Path.cwd() / "debug"),
+)
+_SCREENSHOT_DIR = str(Path(_SCREENSHOT_DIR) / "screenshots")
+_MAX_SCREENSHOTS = 300
+
+_log = logging.getLogger("ScreenshotOnFail")
+_log.info("ScreenshotOnFail 模块已加载")
+
+_HAS_CV2 = False
+try:
+    import cv2
+
+    _HAS_CV2 = True
+except ImportError:
+    _log.warning("cv2 不可用，将回退到 BMP 格式")
+
+
+def _cleanup_old_screenshots() -> int:
+    dirpath = Path(_SCREENSHOT_DIR)
+    dirpath.mkdir(parents=True, exist_ok=True)
+    files = sorted(
+        [f for f in dirpath.glob("*") if f.suffix.lower() in (".jpg", ".bmp")],
+        key=lambda p: p.stat().st_mtime,
+        reverse=True,
+    )
+    deleted = 0
+    for f in files[_MAX_SCREENSHOTS:]:
+        f.unlink()
+        deleted += 1
+    return deleted
+
+
+def _save_image(img: np.ndarray, filepath: Path) -> bool:
+    if img is None or img.size == 0:
+        return False
+
+    if _HAS_CV2:
+        success, encoded = cv2.imencode(
+            ".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 85]
+        )
+        if not success:
+            return False
+        with open(filepath, "wb") as f:
+            f.write(encoded.tobytes())
+        return True
+
+    return _save_bmp(img, filepath.with_suffix(".bmp"))
+
+
+def _save_bmp(img: np.ndarray, filepath: Path) -> bool:
+    if not img.flags.c_contiguous:
+        img = np.ascontiguousarray(img)
+
+    h, w = img.shape[:2]
+    if img.ndim == 2:
+        img = np.stack([img] * 3, axis=-1)
+    elif img.ndim == 3:
+        channels = img.shape[2]
+        if channels == 1:
+            img = np.stack([img[:, :, 0]] * 3, axis=-1)
+        elif channels == 4:
+            img = img[:, :, :3]
+        elif channels == 2:
+            img = np.dstack([img[:, :, :2], np.zeros((h, w), dtype=img.dtype)])
+        elif channels != 3:
+            return False
+
+    row_size = (w * 3 + 3) // 4 * 4
+    pixel_data_size = row_size * h
+    file_size = 14 + 40 + pixel_data_size
+
+    with open(filepath, "wb") as f:
+        f.write(b"BM")
+        f.write(struct.pack("<I", file_size))
+        f.write(struct.pack("<HH", 0, 0))
+        f.write(struct.pack("<I", 14 + 40))
+
+        f.write(struct.pack("<I", 40))
+        f.write(struct.pack("<i", w))
+        f.write(struct.pack("<i", h))
+        f.write(struct.pack("<H", 1))
+        f.write(struct.pack("<H", 24))
+        f.write(struct.pack("<I", 0))
+        f.write(struct.pack("<I", pixel_data_size))
+        f.write(struct.pack("<i", 2835))
+        f.write(struct.pack("<i", 2835))
+        f.write(struct.pack("<I", 0))
+        f.write(struct.pack("<I", 0))
+
+        padding = b"\x00" * (row_size - w * 3)
+        for r in range(h - 1, -1, -1):
+            f.write(img[r, :, :3].tobytes())
+            if padding:
+                f.write(padding)
+
+    return True
+
+
+@AgentServer.context_sink()
+class NodeScreenshotSink(ContextEventSink):
+    def on_node_pipeline_node(
+        self,
+        context: Context,
+        noti_type: NotificationType,
+        detail: ContextEventSink.NodePipelineNodeDetail,
+    ):
+        node_name = detail.name or "unknown"
+
+        img: np.ndarray | None = None
+        reco_id = 0
+        if noti_type in (NotificationType.Succeeded, NotificationType.Failed):
+            try:
+                node_detail = context.tasker.get_node_detail(detail.node_id)
+                if node_detail and node_detail.recognition:
+                    reco_id = node_detail.recognition.reco_id
+                    reco_detail = context.tasker.get_recognition_detail(reco_id)
+                    if reco_detail:
+                        img = getattr(reco_detail, "raw_image", None)
+                        if img is None or img.size == 0:
+                            draw_imgs = getattr(reco_detail, "draw_images", None)
+                            if draw_imgs:
+                                img = draw_imgs[0] if isinstance(draw_imgs, list) else draw_imgs
+            except Exception:
+                pass
+
+        if img is None or img.size == 0:
+            try:
+                img = context.tasker.controller.cached_image
+            except Exception:
+                pass
+
+        if img is None or img.size == 0:
+            return
+
+        dirpath = Path(_SCREENSHOT_DIR)
+        dirpath.mkdir(parents=True, exist_ok=True)
+        timestamp = datetime.now().strftime("%Y.%m.%d-%H.%M.%S.%f")[:-3]
+        status = "success" if noti_type == NotificationType.Succeeded else "failed"
+        filename = f"{timestamp}_{node_name}_{reco_id}_{status}.jpg"
+        filepath = dirpath / filename
+
+        _save_image(img, filepath)
+        _cleanup_old_screenshots()