fix(fleet): ship_type validation and diagnostics improvements

autowsgr/__init__.py

@@ -1,3 +1,3 @@
 """AutoWSGR - 战舰少女R 自动化框架(v2)"""
 
-__version__ = '2.1.9.post6'
+__version__ = '2.1.9.post7'

autowsgr/server/schemas.py

@@ -8,6 +8,31 @@
 from pydantic import BaseModel, Field, field_validator, model_validator
 
 
+_ALLOWED_SHIP_TYPE_CODES = {
+    'dd',
+    'cl',
+    'ca',
+    'cav',
+    'clt',
+    'bb',
+    'bc',
+    'bbv',
+    'cv',
+    'cvl',
+    'av',
+    'ss',
+    'ssg',
+    'cg',
+    'cgaa',
+    'ddg',
+    'ddgaa',
+    'bm',
+    'cbg',
+    'cf',
+    'ss_or_ssg',
+}
+
+
 # ═══════════════════════════════════════════════════════════════════════════════
 # 枚举类型
 # ═══════════════════════════════════════════════════════════════════════════════
@@ -71,6 +96,7 @@ class FleetRuleRequest(BaseModel):
 
     candidates: list[str] = Field(min_length=1, description='候选舰船名（按优先级）')
     search_name: str | None = Field(default=None, description='选船搜索关键词（用于同名舰船区分）')
+    ship_type: str | None = Field(default=None, description='舰种约束（如 cl/cav/ss）')
     min_level: int | None = Field(default=None, ge=1, description='等级下限（含）')
     max_level: int | None = Field(default=None, ge=1, description='等级上限（含）')
 
@@ -82,6 +108,21 @@ def _validate_candidates(cls, value: list[str]) -> list[str]:
             raise ValueError('candidates 不能为空')
         return normalized
 
+    @field_validator('ship_type')
+    @classmethod
+    def _validate_ship_type(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+
+        normalized = value.strip().lower()
+        if not normalized:
+            return None
+
+        if normalized not in _ALLOWED_SHIP_TYPE_CODES:
+            allowed = ', '.join(sorted(_ALLOWED_SHIP_TYPE_CODES))
+            raise ValueError(f'ship_type 不合法: {value!r}, 可选值: {allowed}')
+        return normalized
+
     @model_validator(mode='after')
     def _validate_level_range(self) -> FleetRuleRequest:
         if (

autowsgr/ui/battle/fleet_change/_change.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import re
 import time
 from collections import Counter
 from typing import TYPE_CHECKING, TypedDict
@@ -34,12 +35,16 @@
 # 等待选船页面出现的超时 (秒)
 _CHOOSE_PAGE_TIMEOUT: float = 5.0
 
+# 舰名尾部别名后缀，如“(苍青幻影)”
+_SHIP_ALIAS_SUFFIX_RE = re.compile(r'\s*[（(][^（）()]*[)）]\s*$')
+
 
 class FleetSlotSelector(TypedDict, total=False):
     """编队槽位规则。"""
 
     candidates: list[str]
     search_name: str
+    ship_type: str
     min_level: int
     max_level: int
 
@@ -89,14 +94,14 @@ def change_fleet(
         fleet_id:
             舰队编号 (2-4); ``None`` 代表不指定舰队。1 队不支持更换。
         ship_names:
-                        目标槽位列表 (按槽位 0-5 顺序); 每个元素可为:
+            目标槽位列表 (按槽位 0-5 顺序); 每个元素可为:
 
-                        - ``str``: 目标舰船名
-                        - ``dict``: 规则对象 (``candidates`` / ``search_name`` /
-                            ``min_level`` / ``max_level``)
-                        - ``None``: 留空
+            - ``str``: 目标舰船名
+            - ``dict``: 规则对象 (``candidates`` / ``search_name`` /
+              ``ship_type`` / ``min_level`` / ``max_level``)
+            - ``None``: 留空
 
-                        另外也兼容具备同名属性的 selector-like 对象。
+            另外也兼容具备同名属性的 selector-like 对象。
 
         Returns
         -------
@@ -293,17 +298,20 @@ def _extract_selector(slot: object | None) -> dict | None:
 
         raw_candidates = None
         raw_search_name = None
+        raw_ship_type = None
         raw_min = None
         raw_max = None
 
         if isinstance(slot, dict):
             raw_candidates = slot.get('candidates')
             raw_search_name = slot.get('search_name')
+            raw_ship_type = slot.get('ship_type')
             raw_min = slot.get('min_level')
             raw_max = slot.get('max_level')
         else:
             raw_candidates = getattr(slot, 'candidates', None)
             raw_search_name = getattr(slot, 'search_name', None)
+            raw_ship_type = getattr(slot, 'ship_type', None)
             raw_min = getattr(slot, 'min_level', None)
             raw_max = getattr(slot, 'max_level', None)
 
@@ -317,6 +325,8 @@ def _extract_selector(slot: object | None) -> dict | None:
         selector: dict[str, object] = {'candidates': candidates}
         if isinstance(raw_search_name, str) and raw_search_name.strip():
             selector['search_name'] = raw_search_name.strip()
+        if isinstance(raw_ship_type, str) and raw_ship_type.strip():
+            selector['ship_type'] = raw_ship_type.strip().lower()
         if isinstance(raw_min, int) and raw_min > 0:
             selector['min_level'] = raw_min
         if isinstance(raw_max, int) and raw_max > 0:
@@ -338,6 +348,29 @@ def _slot_candidates(cls, name: str | None, selector: dict | None) -> list[str]:
             out.append(normalized_name)
         return out
 
+    @classmethod
+    def _normalize_search_name_for_compare(cls, value: str) -> str:
+        normalized = value.strip()
+        if normalized.endswith('·改'):
+            normalized = normalized.removesuffix('·改').strip()
+        normalized = _SHIP_ALIAS_SUFFIX_RE.sub('', normalized)
+        return normalized.strip()
+
+    @classmethod
+    def _matches_search_name(cls, current_name: str | None, raw_search_name: object) -> bool:
+        if current_name is None:
+            return False
+        if not isinstance(raw_search_name, str):
+            return True
+        if not raw_search_name.strip():
+            return True
+
+        search_name = raw_search_name.strip()
+        if current_name == search_name:
+            return True
+
+        return current_name == cls._normalize_search_name_for_compare(search_name)
+
     @classmethod
     def _can_short_circuit(
         cls,
@@ -436,7 +469,7 @@ def _match_existing_members(
                     raw_search_name = selector.get('search_name')
                     if isinstance(raw_search_name, str) and raw_search_name.strip():
                         # 指定了搜索关键词时，不能仅凭同名判定已满足。
-                        if ship != raw_search_name.strip():
+                        if not cls._matches_search_name(ship, raw_search_name):
                             continue
                 ok[i] = True
                 matched_slots.add(i)
@@ -454,7 +487,7 @@ def _match_existing_members(
                     if isinstance(selector, dict):
                         raw_search_name = selector.get('search_name')
                         if isinstance(raw_search_name, str) and raw_search_name.strip():
-                            if ship != raw_search_name.strip():
+                            if not cls._matches_search_name(ship, raw_search_name):
                                 continue
                     ok[j] = True
                     matched_slots.add(i)

autowsgr/ui/choose_ship_page.py

@@ -59,6 +59,29 @@
 _OCR_MAX_ATTEMPTS: int = 3
 _SHIP_ALIAS_SUFFIX_RE = re.compile(r'\s*[（(][^（）()]*[)）]\s*$')
 
+_SHIP_TYPE_KEYWORDS: dict[str, tuple[str, ...]] = {
+    'dd': ('驱逐',),
+    'cl': ('轻巡',),
+    'ca': ('重巡',),
+    'cav': ('航巡',),
+    'clt': ('雷巡',),
+    'bb': ('战列',),
+    'bc': ('战巡',),
+    'bbv': ('航战',),
+    'cv': ('航母',),
+    'cvl': ('轻母',),
+    'av': ('装母',),
+    'ss': ('潜艇',),
+    'ssg': ('导潜',),
+    'cg': ('导巡',),
+    'cgaa': ('防巡',),
+    'ddg': ('导驱',),
+    'ddgaa': ('防驱',),
+    'bm': ('重炮',),
+    'cbg': ('大巡',),
+    'cf': ('旗舰',),
+}
+
 PAGE_SIGNATURE = PixelSignature(
     name='choose_ship_page',
     strategy=MatchStrategy.ALL,
@@ -189,10 +212,11 @@ def change_single_ship(
             (决战选船界面没有搜索框)。
         selector:
             可选规则，支持 ``candidates`` / ``search_name`` /
-            ``min_level`` / ``max_level``。
+            ``ship_type`` / ``min_level`` / ``max_level``。
             其中 ``search_name`` 用于指定搜索框关键字（仅在
             ``use_search=True`` 且界面存在搜索框时生效），
             ``candidates`` 用于限定允许点击的舰船名集合，
+            ``ship_type`` 用于按舰种筛选同名舰船，
             ``min_level`` / ``max_level`` 用于按等级范围筛选。
 
         Returns
@@ -211,6 +235,7 @@ def change_single_ship(
 
         candidates = [name]
         search_name: str | None = None
+        ship_type: str | None = None
         min_level: int | None = None
         max_level: int | None = None
 
@@ -223,8 +248,11 @@ def change_single_ship(
             raw_min = selector.get('min_level')
             raw_max = selector.get('max_level')
             raw_search = selector.get('search_name')
+            raw_ship_type = selector.get('ship_type')
             if isinstance(raw_search, str) and raw_search.strip():
-                search_name = raw_search.strip()
+                search_name = self._normalize_search_keyword(raw_search)
+            if isinstance(raw_ship_type, str) and raw_ship_type.strip():
+                ship_type = raw_ship_type.strip().lower()
             if isinstance(raw_min, int) and raw_min > 0:
                 min_level = raw_min
             if isinstance(raw_max, int) and raw_max > 0:
@@ -236,6 +264,7 @@ def change_single_ship(
             self.ensure_dismiss_keyboard()
             matched = self._click_ship_in_list(
                 name,
+                ship_type=ship_type,
                 min_level=min_level,
                 max_level=max_level,
             )
@@ -244,12 +273,14 @@ def change_single_ship(
                 return matched
 
         for candidate in candidates:
+            search_candidate = self._normalize_search_keyword(candidate)
             if use_search:
                 self.ensure_search_box()
-                self.input_ship_name(candidate)
+                self.input_ship_name(search_candidate)
                 self.ensure_dismiss_keyboard()
             matched = self._click_ship_in_list(
                 candidate,
+                ship_type=ship_type,
                 min_level=min_level,
                 max_level=max_level,
             )
@@ -265,8 +296,18 @@ def change_single_ship(
                 level_hint = f' (等级限制: >= {min_level})'
             else:
                 level_hint = f' (等级限制: <= {max_level})'
-        _log.error('[UI] 未在选船列表中找到可用候选: {}{}', candidates, level_hint)
-        raise RuntimeError(f'未找到满足条件的目标舰船: {candidates}{level_hint}')
+
+        ship_type_hint = ''
+        if ship_type is not None:
+            ship_type_hint = f' (舰种限制: {ship_type})'
+
+        _log.error(
+            '[UI] 未在选船列表中找到可用候选: {}{}{}',
+            candidates,
+            level_hint,
+            ship_type_hint,
+        )
+        raise RuntimeError(f'未找到满足条件的目标舰船: {candidates}{level_hint}{ship_type_hint}')
 
     @staticmethod
     def _normalize_hit_entry(hit: object) -> tuple[str, float, float, float]:
@@ -321,6 +362,7 @@ def _click_ship_in_list(
         self,
         name: str,
         *,
+        ship_type: str | None = None,
         min_level: int | None = None,
         max_level: int | None = None,
     ) -> str | None:
@@ -403,6 +445,22 @@ def _click_ship_in_list(
                     )
                     continue
 
+                if ship_type is not None:
+                    detected_ship_type = self._detect_ship_type_near_hit(
+                        screen,
+                        cx,
+                        cy,
+                        row_key,
+                    )
+                    if not self._is_ship_type_in_rule(detected_ship_type, ship_type):
+                        _log.warning(
+                            "[UI] 命中 '{}' 舰种 '{}' 不满足要求 '{}'",
+                            matched,
+                            detected_ship_type if detected_ship_type is not None else '未知',
+                            ship_type,
+                        )
+                        continue
+
                 _log.info(
                     "[UI] 选船 DLL+OCR -> '{}' (第 {}/{} 次), 点击 ({:.3f}, {:.3f})",
                     name,
@@ -427,6 +485,66 @@ def _click_ship_in_list(
 
         return None
 
+    def _detect_ship_type_near_hit(
+        self,
+        screen: np.ndarray,
+        cx: float,
+        cy: float,
+        row_key: float,
+    ) -> str | None:
+        """在命中卡片附近 OCR 识别舰种。"""
+        assert self._ctx.ocr is not None
+
+        h, w = screen.shape[:2]
+        x_px = int(max(0, min(w - 1, cx * w)))
+        y_px = int(max(0, min(h - 1, cy * h)))
+        row_y = int(max(0, min(h - 1, row_key * h))) if row_key >= 0 else y_px
+
+        probes: list[tuple[int, int, int, int]] = [
+            (max(0, x_px - 110), max(0, row_y - 120), min(w, x_px + 110), max(0, row_y - 12)),
+            (max(0, x_px - 130), max(0, y_px - 150), min(w, x_px + 130), max(0, y_px - 18)),
+            (max(0, x_px - 140), max(0, y_px - 170), min(w, x_px + 140), min(h, y_px + 20)),
+        ]
+
+        for x1, y1, x2, y2 in probes:
+            if x2 - x1 < 16 or y2 - y1 < 16:
+                continue
+            crop = screen[y1:y2, x1:x2]
+            results = self._ctx.ocr.recognize(crop)
+            for result in results:
+                text = str(getattr(result, 'text', '')).strip()
+                ship_type = self._extract_ship_type_from_text(text)
+                if ship_type is not None:
+                    return ship_type
+        return None
+
+    @staticmethod
+    def _extract_ship_type_from_text(text: str) -> str | None:
+        if not text:
+            return None
+        normalized = text.replace(' ', '')
+        for ship_type, keywords in _SHIP_TYPE_KEYWORDS.items():
+            if any(keyword in normalized for keyword in keywords):
+                return ship_type
+        return None
+
+    @staticmethod
+    def _is_ship_type_in_rule(detected: str | None, expected: str) -> bool:
+        if detected is None:
+            return False
+        rule = expected.strip().lower()
+        if rule == 'ss_or_ssg':
+            return detected in {'ss', 'ssg'}
+        return detected == rule
+
+    @staticmethod
+    def _normalize_search_keyword(name: str) -> str:
+        normalized = name.strip()
+        if normalized.endswith('·改'):
+            normalized = normalized.removesuffix('·改').strip()
+        normalized = _SHIP_ALIAS_SUFFIX_RE.sub('', normalized)
+        return normalized.strip()
+
     @staticmethod
     def _normalize_ship_name(name: str) -> str:
         normalized = name.strip()