"""L2 LocatorRegistry：按 (app_version, resolution) 加载 YAML profile + 熔断接口。 加载优先级：精确匹配 > 版本匹配 > default 兜底。 熔断接口：record_image_failure / record_image_success / is_circuited， P1 阶段 no-op（opencv=None），P2 启用图像路径后生效。 """ from __future__ import annotations import logging import os import re import time from typing import Optional import yaml from woc_bridge.ui.locators.base import GeomSpec, Selector logger = logging.getLogger("woc-bridge") class LocatorRegistry: """选择器注册表。 Attributes: profile_dir: YAML profile 目录绝对路径 opencv: OpenCVBackend 实例（P1 阶段为 None，P2 启用图像路径） selectors: kind -> Selector 映射 theme: 当前主题（light / dark），P1 阶段固定 light """ # 熔断参数：连续失败 10 次熔断 5 分钟 _FAIL_THRESHOLD = 10 _CIRCUIT_OPEN_SEC = 300.0 def __init__(self, profile_dir: str, opencv=None) -> None: """初始化注册表。 Args: profile_dir: YAML profile 目录绝对路径 opencv: OpenCVBackend 实例（P1 阶段为 None） """ self.profile_dir = profile_dir self.opencv = opencv self.selectors: dict[str, Selector] = {} self.theme: str = "light" self._fail_counts: dict[str, int] = {} self._circuit_open_until: dict[str, float] = {} # ------------------------------------------------------------------ # Profile 加载 # ------------------------------------------------------------------ def load(self, app_version: str, resolution: tuple[int, int]) -> None: """按精确 > 最近分辨率 > 版本 > default 优先级加载 YAML profile。 Args: app_version: 微信版本号，如 "4.0" resolution: (width, height) 元组，如 (1920, 1080) """ exact_name = f"wechat_{app_version}_{resolution[0]}x{resolution[1]}.yaml" exact_path = os.path.join(self.profile_dir, exact_name) if os.path.exists(exact_path): logger.info( "[locator] load exact profile: %s (app_version=%s resolution=%s)", exact_name, app_version, resolution, ) self._load_yaml(exact_path) return # 无精确匹配时，找同版本最近分辨率并按比例缩放绝对偏移 nearest = self._find_nearest_profile(app_version, resolution) if nearest: name, profile_w, profile_h = nearest logger.info( "[locator] load nearest profile: %s for resolution=%s " "(scale=%.3fx%.3f)", name, resolution, resolution[0] / profile_w, resolution[1] / profile_h, ) self._load_yaml( os.path.join(self.profile_dir, name), target_resolution=resolution, profile_resolution=(profile_w, profile_h), ) return # 版本通用/默认兜底 for name in [f"wechat_{app_version}.yaml", "default.yaml"]: path = os.path.join(self.profile_dir, name) if os.path.exists(path): logger.info( "[locator] load profile: %s (app_version=%s resolution=%s)", name, app_version, resolution, ) self._load_yaml(path) return logger.warning( "[locator] no profile found in %s, fall back to built-in defaults " "(app_version=%s resolution=%s)", self.profile_dir, app_version, resolution, ) self._load_defaults() def _find_nearest_profile( self, app_version: str, resolution: tuple[int, int] ) -> Optional[tuple[str, int, int]]: """找与当前分辨率最接近的 wechat__x.yaml。""" pattern = re.compile(rf"wechat_{re.escape(app_version)}_(\d+)x(\d+)\.yaml$") target_w, target_h = resolution target_area = target_w * target_h best: Optional[tuple[str, int, int]] = None best_score: Optional[float] = None for name in os.listdir(self.profile_dir): m = pattern.match(name) if not m: continue pw, ph = int(m.group(1)), int(m.group(2)) area = pw * ph area_ratio = min(target_area, area) / max(target_area, area) aspect_diff = abs(target_w / target_h - pw / ph) # 面积越接近、宽高比越接近，分数越低 score = (1 - area_ratio) + aspect_diff if best_score is None or score < best_score: best_score = score best = (name, pw, ph) return best def set_theme(self, theme: str) -> None: """切换主题（light / dark）。""" if theme != self.theme: self.theme = theme logger.info("[locator] theme switched to %s", theme) # ------------------------------------------------------------------ # 选择器访问 + 熔断 # ------------------------------------------------------------------ def get(self, kind: str) -> Selector: """返回 kind 对应 Selector，无则返回 _default_selector(kind)。""" return self.selectors.get(kind) or self._default_selector(kind) def is_circuited(self, kind: str) -> bool: """检查 kind 是否处于熔断期。 熔断到期时自动重置失败计数，使半开状态允许完整次数的试探， 而非首次失败即重新熔断。 """ deadline = self._circuit_open_until.get(kind, 0) if time.monotonic() < deadline: return True # 熔断已到期：清理熔断标记和失败计数，进入半开状态 if kind in self._circuit_open_until: self._circuit_open_until.pop(kind, None) self._fail_counts[kind] = 0 logger.info( "[locator] kind=%s circuit recovered to HALF_OPEN (fail_count reset)", kind, ) return False def record_image_failure(self, kind: str) -> None: """记录一次图像匹配失败。连续 _FAIL_THRESHOLD 次后熔断 _CIRCUIT_OPEN_SEC 秒。""" self._fail_counts[kind] = self._fail_counts.get(kind, 0) + 1 if self._fail_counts[kind] >= self._FAIL_THRESHOLD: self._circuit_open_until[kind] = time.monotonic() + self._CIRCUIT_OPEN_SEC logger.warning( "[locator] kind=%s image match circuit OPEN for %.0fs " "(fail_count=%d)", kind, self._CIRCUIT_OPEN_SEC, self._fail_counts[kind], ) def record_image_success(self, kind: str) -> None: """记录一次图像匹配成功，重置失败计数与熔断。""" self._fail_counts[kind] = 0 self._circuit_open_until.pop(kind, None) # ------------------------------------------------------------------ # YAML 解析与默认值 # ------------------------------------------------------------------ def _load_yaml( self, path: str, target_resolution: Optional[tuple[int, int]] = None, profile_resolution: Optional[tuple[int, int]] = None, ) -> None: """用 yaml.safe_load 加载 YAML 并填充 self.selectors。 若提供 target_resolution 与 profile_resolution，则对 by_geom 中的 绝对偏移 x_offset / y_offset 按分辨率比例缩放（宽高比不变时坐标更接近目标）。 YAML 结构（每个 key 对应一个 Selector）： search_box: by_image: search_box.png # str，可省略 by_geom: # dict，可省略 relative_to: window x_ratio: 0.08 y_ratio: 0.05 x_offset: 0 y_offset: 0 threshold: 0.80 require_image: false description: 搜索框 """ try: with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} except (OSError, yaml.YAMLError) as exc: logger.error("[locator] load yaml failed: %s (%s)", path, exc) self._load_defaults() return scale_x = scale_y = 1.0 if target_resolution and profile_resolution: scale_x = target_resolution[0] / profile_resolution[0] scale_y = target_resolution[1] / profile_resolution[1] self.selectors.clear() for kind, spec in data.items(): if not isinstance(spec, dict): continue by_image = spec.get("by_image") by_geom_raw = spec.get("by_geom") by_geom: Optional[GeomSpec] = None if isinstance(by_geom_raw, dict): by_geom = GeomSpec( relative_to=by_geom_raw.get("relative_to", "window"), x_ratio=float(by_geom_raw.get("x_ratio", 0.0)), y_ratio=float(by_geom_raw.get("y_ratio", 0.0)), x_offset=int(int(by_geom_raw.get("x_offset", 0)) * scale_x), y_offset=int(int(by_geom_raw.get("y_offset", 0)) * scale_y), ) self.selectors[kind] = Selector( kind=kind, by_image=by_image, by_geom=by_geom, threshold=float(spec.get("threshold", 0.80)), require_image=bool(spec.get("require_image", False)), description=str(spec.get("description", "")), ) logger.info( "[locator] loaded %d selectors from %s (scale=%.3fx%.3f)", len(self.selectors), path, scale_x, scale_y, ) def _default_selector(self, kind: str) -> Selector: """返回 kind 的兜底 Selector（仅 description 标注 default）。""" return Selector(kind=kind, description=f"default {kind}") def _load_defaults(self) -> None: """从 xdotool_driver.py 现有硬编码常量迁移默认几何兜底。 参考 spec 4.2 节 YAML 示例与项目记忆中的硬编码值， 覆盖 6 个核心元素：search_box / send_button / input_box / input_box_empty / search_result_first / main_view。 relative_to 字段仅作语义占位，实际坐标计算统一用 ratio + offset （offset 已含正负号区分方向，详见 Actions._geom_find）。 """ defaults = { "session_manager_icon": Selector( kind="session_manager_icon", by_geom=GeomSpec( relative_to="window", x_ratio=0.018, y_ratio=0.109, x_offset=0, y_offset=0, ), description="左侧导航栏顶部微信消息/会话管理图标", ), "search_box": Selector( kind="search_box", by_geom=GeomSpec( relative_to="window", x_ratio=0.092, y_ratio=0.095, x_offset=0, y_offset=0, ), description="左侧顶部搜索框", ), "send_button": Selector( kind="send_button", by_geom=GeomSpec( relative_to="window", x_ratio=0.972, y_ratio=0.929, x_offset=0, y_offset=0, ), description="右下角发送按钮", ), "input_box": Selector( kind="input_box", by_geom=GeomSpec( relative_to="window", x_ratio=0.743, y_ratio=0.914, x_offset=0, y_offset=0, ), description="聊天输入框", ), "input_box_empty": Selector( kind="input_box_empty", by_geom=GeomSpec( relative_to="window", x_ratio=0.743, y_ratio=0.914, x_offset=0, y_offset=0, ), description="空输入框灰色提示态（发送成功后校验）", ), "search_result_first": Selector( kind="search_result_first", by_geom=GeomSpec( relative_to="window", x_ratio=0.078, y_ratio=0.174, x_offset=0, y_offset=0, ), description="搜索结果第一项", ), "main_view": Selector( kind="main_view", by_geom=GeomSpec( relative_to="window", x_ratio=0.65, y_ratio=0.50, x_offset=0, y_offset=0, ), description="主聊天视图（post_verify 校验用）", ), } self.selectors.clear() self.selectors.update(defaults) logger.info( "[locator] loaded %d built-in default selectors", len(self.selectors) )