fix(fleet): support selector rules and disambiguate same-name ships (#420)

* fix(fleet): support fleet_rules selectors and disambiguate same-name ships

* chore(release): bump autowsgr to 2.1.9.post5

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(review): address Copilot feedback for fleet rules

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(review): address new Copilot comments on PR #419

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: X-vedeax <65024689+veadex@users.noreply.github.com>

Author: 3 people

autowsgr/__init__.py

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

autowsgr/ops/event_fight.py

@@ -13,7 +13,7 @@
 from __future__ import annotations
 
 import time
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from autowsgr.combat import CombatMode, CombatPlan, CombatResult
 from autowsgr.combat.engine import run_combat
@@ -70,13 +70,15 @@ def __init__(
         event_name: str | None = None,
         fleet_id: int | None = None,
         fleet: list[str] | None = None,
+        fleet_rules: list[Any] | None = None,
     ) -> None:
         self._ctx = ctx
         self._ctrl = ctx.ctrl
         self._plan = plan
         self._entrance = entrance
         self._fleet_id = fleet_id if fleet_id is not None else (plan.fleet_id or 1)
         self._fleet = fleet if fleet is not None else plan.fleet
+        self._fleet_rules = fleet_rules
         self._skip_check = False  # 首次执行时检查难度和节点，后续重复执行时跳过检查以节省时间
 
         # 推导 map_code
@@ -111,6 +113,35 @@ def __init__(
         self._results: list[CombatResult] = []
         self._fleet_ships = None
 
+    @staticmethod
+    def _primary_names_from_rules(fleet_rules: list[Any] | None) -> list[str | None] | None:
+        if not fleet_rules:
+            return None
+
+        def _normalize_name(value: object) -> str | None:
+            if value is None:
+                return None
+            name = str(value).strip()
+            return name or None
+
+        names: list[str | None] = []
+        for slot in fleet_rules[:6]:
+            if isinstance(slot, str):
+                names.append(_normalize_name(slot))
+                continue
+
+            candidates = None
+            if isinstance(slot, dict):
+                candidates = slot.get('candidates')
+            else:
+                candidates = getattr(slot, 'candidates', None)
+
+            if isinstance(candidates, list) and len(candidates) > 0:
+                names.append(_normalize_name(candidates[0]))
+                continue
+            names.append(None)
+        return names
+
     # ── 公共接口 ──
 
     def run(self) -> CombatResult:
@@ -217,11 +248,20 @@ def _prepare_for_battle(self) -> list[ShipDamageState]:
         page.select_fleet(self._fleet_id)
         time.sleep(0.5)
 
-        # 换船 (如果指定了舰船列表)
-        if self._plan.fleet is not None:
+        resolved_ship_names: list[str | None] | None = None
+
+        # 换船 (若提供了规则则优先按规则执行)
+        if self._fleet_rules is not None:
             page.change_fleet(
                 self._fleet_id,
-                self._plan.fleet,
+                self._fleet_rules,
+            )
+            time.sleep(0.5)
+            resolved_ship_names = page.detect_fleet()
+        elif self._fleet is not None:
+            page.change_fleet(
+                self._fleet_id,
+                self._fleet,
             )
             time.sleep(0.5)
 
@@ -244,7 +284,14 @@ def _prepare_for_battle(self) -> list[ShipDamageState]:
         # 检测战前舰队信息 (血量 + 等级)
         fleet_info = page.detect_fleet_info()
         ship_stats = [fleet_info.ship_damage.get(i, ShipDamageState.NORMAL) for i in range(6)]
-        self._fleet_ships = fleet_info.to_ships(self._plan.fleet)
+        ship_names = resolved_ship_names
+        if ship_names is None:
+            ship_names = (
+                self._primary_names_from_rules(self._fleet_rules)
+                if self._fleet_rules is not None
+                else self._fleet
+            )
+        self._fleet_ships = fleet_info.to_ships(ship_names)
 
         # 出征
         page.start_battle()
@@ -302,6 +349,7 @@ def run_event_fight(
     gap: float = 0.0,
     fleet_id: int | None = None,
     fleet: list[str] | None = None,
+    fleet_rules: list[Any] | None = None,
 ) -> list[CombatResult]:
     """执行活动战的便捷函数。
 
@@ -331,6 +379,7 @@ def run_event_fight(
         entrance=entrance,
         fleet_id=fleet_id,
         fleet=fleet,
+        fleet_rules=fleet_rules,
     )
     return runner.run_for_times(times, gap=gap)
 
@@ -344,6 +393,7 @@ def run_event_fight_from_yaml(
     times: int = 1,
     fleet_id: int | None = None,
     fleet: list[str] | None = None,
+    fleet_rules: list[Any] | None = None,
 ) -> list[CombatResult]:
     """从 YAML 文件加载计划并执行活动战。
 
@@ -386,4 +436,5 @@ def run_event_fight_from_yaml(
         times=times,
         fleet_id=fleet_id,
         fleet=fleet,
+        fleet_rules=fleet_rules,
     )

autowsgr/ops/normal_fight.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from autowsgr.combat import CombatMode, CombatPlan, CombatResult
 from autowsgr.combat.engine import run_combat
@@ -37,12 +37,14 @@ def __init__(
         plan: CombatPlan,
         fleet_id: int | None = None,
         fleet: list[str] | None = None,
+        fleet_rules: list[Any] | None = None,
     ) -> None:
         self._ctx = ctx
         self._ctrl = ctx.ctrl
         self._plan = plan
         self._fleet_id = fleet_id if fleet_id is not None else plan.fleet_id
         self._fleet = fleet if fleet is not None else plan.fleet
+        self._fleet_rules = fleet_rules
 
         # 从 config 读取拆船配置
         self._dock_full_destroy = ctx.config.dock_full_destroy
@@ -61,6 +63,35 @@ def __init__(
         self._ship_acquired_count: int | None = None
         self._fleet_ships: list[Ship] | None = None
 
+    @staticmethod
+    def _primary_names_from_rules(fleet_rules: list[Any] | None) -> list[str | None] | None:
+        if not fleet_rules:
+            return None
+
+        def _normalize_name(value: object) -> str | None:
+            if value is None:
+                return None
+            name = str(value).strip()
+            return name or None
+
+        names: list[str | None] = []
+        for slot in fleet_rules[:6]:
+            if isinstance(slot, str):
+                names.append(_normalize_name(slot))
+                continue
+
+            candidates = None
+            if isinstance(slot, dict):
+                candidates = slot.get('candidates')
+            else:
+                candidates = getattr(slot, 'candidates', None)
+
+            if isinstance(candidates, list) and len(candidates) > 0:
+                names.append(_normalize_name(candidates[0]))
+                continue
+            names.append(None)
+        return names
+
     # ── 公共接口 ──
 
     def run(self) -> CombatResult:
@@ -292,8 +323,17 @@ def _prepare_for_battle(self) -> list[ShipDamageState]:
         page.select_fleet(self._fleet_id)
         time.sleep(0.5)
 
-        # 换船 (如果指定了舰船列表)
-        if self._fleet is not None:
+        resolved_ship_names: list[str | None] | None = None
+
+        # 换船 (若提供了规则则优先按规则执行)
+        if self._fleet_rules is not None:
+            page.change_fleet(
+                self._fleet_id,
+                self._fleet_rules,
+            )
+            time.sleep(0.5)
+            resolved_ship_names = page.detect_fleet()
+        elif self._fleet is not None:
             page.change_fleet(
                 self._fleet_id,
                 self._fleet,
@@ -322,7 +362,14 @@ def _prepare_for_battle(self) -> list[ShipDamageState]:
         if ShipDamageState.SEVERE in ship_stats:
             _log.error('[OPS] 出征前检测到大破舰船，退出程序')
             raise ActionFailedError('出征前检测到大破舰船，退出程序')
-        self._fleet_ships = fleet_info.to_ships(self._fleet)
+        ship_names = resolved_ship_names
+        if ship_names is None:
+            ship_names = (
+                self._primary_names_from_rules(self._fleet_rules)
+                if self._fleet_rules is not None
+                else self._fleet
+            )
+        self._fleet_ships = fleet_info.to_ships(ship_names)
 
         # 出征
         page.start_battle()
@@ -394,13 +441,15 @@ def run_normal_fight(
     gap: float = 0.0,
     fleet_id: int | None = None,
     fleet: list[str] | None = None,
+    fleet_rules: list[Any] | None = None,
 ) -> list[CombatResult]:
     """执行常规战的便捷函数。"""
     runner = NormalFightRunner(
         ctx,
         plan,
         fleet_id=fleet_id,
         fleet=fleet,
+        fleet_rules=fleet_rules,
     )
     return runner.run_for_times(times, gap=gap)
 
@@ -412,6 +461,7 @@ def run_normal_fight_from_yaml(
     times: int = 1,
     fleet_id: int | None = None,
     fleet: list[str] | None = None,
+    fleet_rules: list[Any] | None = None,
 ) -> list[CombatResult]:
     """从 YAML 文件加载计划并执行常规战。
 
@@ -422,4 +472,11 @@ def run_normal_fight_from_yaml(
       包数据目录中查找，可省略 ``.yaml`` 后缀。
     """
     plan = get_normal_fight_plan(yaml_path)
-    return run_normal_fight(ctx, plan, times=times, fleet_id=fleet_id, fleet=fleet)
+    return run_normal_fight(
+        ctx,
+        plan,
+        times=times,
+        fleet_id=fleet_id,
+        fleet=fleet,
+        fleet_rules=fleet_rules,
+    )

autowsgr/server/routes/task.py

@@ -111,6 +111,7 @@ def executor(task_info: Any) -> list[dict[str, Any]]:
         request_plan = request.plan
         override_fleet_id = request_plan.fleet_id if request_plan is not None else None
         override_fleet = request_plan.fleet if request_plan is not None else None
+        override_fleet_rules = request_plan.fleet_rules if request_plan is not None else None
 
         for i in range(request.times):
             if task_manager.should_stop():
@@ -126,6 +127,7 @@ def executor(task_info: Any) -> list[dict[str, Any]]:
                     times=1,
                     fleet_id=override_fleet_id,
                     fleet=override_fleet,
+                    fleet_rules=override_fleet_rules,
                 )[0]
                 results.append(convert_combat_result(result, i + 1))
                 task_manager.add_result(results[-1])
@@ -165,6 +167,7 @@ def executor(task_info: Any) -> list[dict[str, Any]]:
 
         request_plan = request.plan
         override_fleet = request_plan.fleet if request_plan is not None else None
+        override_fleet_rules = request_plan.fleet_rules if request_plan is not None else None
         # 优先级: 顶层 fleet_id > plan 覆盖 fleet_id > YAML 内 fleet_id
         if request.fleet_id is not None:
             fleet_id = request.fleet_id
@@ -187,6 +190,7 @@ def executor(task_info: Any) -> list[dict[str, Any]]:
                     times=1,
                     fleet_id=fleet_id,
                     fleet=override_fleet,
+                    fleet_rules=override_fleet_rules,
                 )[0]
                 results.append(convert_combat_result(result, i + 1))
                 task_manager.add_result(results[-1])

autowsgr/server/schemas.py

@@ -5,7 +5,7 @@
 from enum import StrEnum
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 # ═══════════════════════════════════════════════════════════════════════════════
@@ -66,6 +66,35 @@ class NodeDecisionRequest(BaseModel):
     model_config = {'extra': 'forbid'}
 
 
+class FleetRuleRequest(BaseModel):
+    """编队槽位候选规则。"""
+
+    candidates: list[str] = Field(min_length=1, description='候选舰船名（按优先级）')
+    search_name: str | None = Field(default=None, description='选船搜索关键词（用于同名舰船区分）')
+    min_level: int | None = Field(default=None, ge=1, description='等级下限（含）')
+    max_level: int | None = Field(default=None, ge=1, description='等级上限（含）')
+
+    @field_validator('candidates')
+    @classmethod
+    def _validate_candidates(cls, value: list[str]) -> list[str]:
+        normalized = [name.strip() for name in value if name and name.strip()]
+        if len(normalized) == 0:
+            raise ValueError('candidates 不能为空')
+        return normalized
+
+    @model_validator(mode='after')
+    def _validate_level_range(self) -> FleetRuleRequest:
+        if (
+            self.min_level is not None
+            and self.max_level is not None
+            and self.max_level < self.min_level
+        ):
+            raise ValueError('max_level 必须大于或等于 min_level')
+        return self
+
+    model_config = {'extra': 'forbid'}
+
+
 class CombatPlanRequest(BaseModel):
     """作战计划请求体。"""
 
@@ -75,6 +104,10 @@ class CombatPlanRequest(BaseModel):
     map: int | str = Field(default=1, description='地图号')
     fleet_id: int = Field(default=1, ge=1, le=6, description='舰队编号')
     fleet: list[str] | None = Field(default=None, description='舰队成员')
+    fleet_rules: list[str | FleetRuleRequest] | None = Field(
+        default=None,
+        description='舰队槽位规则（字符串或候选规则）',
+    )
     repair_mode: list[int] = Field(
         default_factory=lambda: [2, 2, 2, 2, 2, 2],
         description='修理策略 (6个位置)',