feat: 凹图次数限制

Author: Windsland52

custom/maa/farm.py

@@ -23,15 +23,16 @@
     sys.path.insert(0, _ROOT)
 
 from custom.core.timing import calibration  # noqa: E402
+from custom.maa.reco.click_stage import get_attempt_count, reset_attempt_count  # noqa: E402
 from custom.maa.registry import register_all  # noqa: E402
 from custom.utils.jsonc import load as load_jsonc  # noqa: E402
 from custom.utils.logger import setup_logging  # noqa: E402
 from custom.utils.runtime_paths import configure_paths  # noqa: E402
 
 logger = logging.getLogger(__name__)
 
-# 每轮估时（秒），用于 max-retries 超时估算
-_EST_SECONDS_PER_ROUND = 120
+# 次数监控线程的轮询间隔（秒）
+_POLL_INTERVAL = 2.0
 
 
 def main() -> int:
@@ -48,7 +49,10 @@ def main() -> int:
         help="难度：normal=普通 | sand=沙盘推演",
     )
     parser.add_argument(
-        "--max-retries", type=int, default=50, help="最大重试次数（0=无限，靠 Ctrl+C 停）"
+        "--max-retries",
+        type=int,
+        default=50,
+        help="最大凹图次数（0=无限，靠 Ctrl+C 停）。按实际尝试计数，不再用估时",
     )
     parser.add_argument(
         "--profile", default=None, help="校准文件名（可选，默认 config.DEFAULT_CALIBRATION）"
@@ -72,6 +76,7 @@ def main() -> int:
 
     if args.profile:
         calibration.load(args.profile)  # 校验可加载
+    reset_attempt_count()  # 每次运行从 0 计数
     logger.info("时间轴: %s（map_code 从文件读取）", args.timeline)
 
     wins = Toolkit.find_desktop_windows()
@@ -115,23 +120,29 @@ def main() -> int:
     pipeline["Farm"]["anchor"] = {"Farm@SwitchDifficulty": anchor_target}
     logger.info("难度：%s", "沙盘推演" if args.difficulty == "sand" else "普通")
 
-    # max-retries 超时定时器（到点 post_stop）
+    # max-retries 次数监控线程：累计凹图尝试达上限则 post_stop
+    # 用 > 而非 >=：count 在 ClickStage 识别成功时自增（约每轮一次），
+    # 故 count > max_retries 意味着已发起 max_retries 次完整尝试、第 N+1 次刚开始，此时停。
     if args.max_retries:
-        deadline = args.max_retries * _EST_SECONDS_PER_ROUND
 
-        def _timeout_stop() -> None:
-            time.sleep(deadline)
-            if not tasker.stopping:
-                logger.warning("已达 max-retries %d（≈%ds），停止", args.max_retries, deadline)
-                tasker.post_stop()
-
-        threading.Thread(target=_timeout_stop, daemon=True).start()
+        def _count_stop() -> None:
+            while not tasker.stopping:
+                time.sleep(_POLL_INTERVAL)
+                if get_attempt_count() > args.max_retries:
+                    if not tasker.stopping:
+                        logger.warning(
+                            "已达 max-retries %d 次（实际 %d 次），停止",
+                            args.max_retries,
+                            get_attempt_count(),
+                        )
+                        tasker.post_stop()
+                    return
+
+        threading.Thread(target=_count_stop, daemon=True).start()
 
     logger.info(
         "开始凹图%s。请在关卡列表页等待...",
-        f"（最多 {args.max_retries} 次，≈{args.max_retries * _EST_SECONDS_PER_ROUND}s）"
-        if args.max_retries
-        else "（无限，Ctrl+C 停）",
+        f"（最多 {args.max_retries} 次凹图）" if args.max_retries else "（无限，Ctrl+C 停）",
     )
 
     t_start = time.time()

custom/maa/reco/click_stage.py

@@ -8,6 +8,7 @@
 
 import json
 import logging
+import time
 from pathlib import Path
 
 from maa.context import Context
@@ -20,6 +21,41 @@
 
 _TIMELINE_DIR = project_root() / "config" / "timelines"
 
+# —— 凹图轮次计数 ——
+# ClickStage 每轮循环进入一次 ≈ 一次凹图尝试。Farm 用 JumpBack 反复跳回
+# ClickStage，同一轮里识别失败会重试，故用时间间隔去重避免重复计数。
+_MIN_ATTEMPT_INTERVAL = 5.0  # 秒：距上次计数不足此值则不计数
+_attempt_count = 0
+_last_attempt_time = 0.0  # monotonic，0 表示尚未计过
+
+
+def get_attempt_count() -> int:
+    """当前累计的有效凹图尝试次数（供外部读取，如 max-retries 判定）。"""
+    return _attempt_count
+
+
+def reset_attempt_count() -> None:
+    """重置计数（新一轮 farm 运行前调用）。"""
+    global _attempt_count, _last_attempt_time
+    _attempt_count = 0
+    _last_attempt_time = 0.0
+
+
+def _maybe_count_attempt() -> None:
+    """进入 ClickStage 即记一次尝试；距上次不足 _MIN_ATTEMPT_INTERVAL 则跳过。"""
+    global _attempt_count, _last_attempt_time
+    now = time.monotonic()
+    if _last_attempt_time and (now - _last_attempt_time) < _MIN_ATTEMPT_INTERVAL:
+        logger.debug(
+            "跳过凹图计数（距上次 %.1fs < %.0fs）",
+            now - _last_attempt_time,
+            _MIN_ATTEMPT_INTERVAL,
+        )
+        return
+    _attempt_count += 1
+    _last_attempt_time = now
+    logger.info("▶ 第 %d 次凹图尝试", _attempt_count)
+
 
 def _resolve_timeline_path(timeline_path: str) -> Path:
     """timeline_path 纯文件名 → config/timelines/ 下；带路径 → 相对项目根。"""
@@ -105,4 +141,5 @@ def analyze(
         if not box:
             return None
         logger.info("识别到关卡 %s at %s", stage_text, box)
+        _maybe_count_attempt()
         return box