From 39643dfc5355fd9756e6152a68377037bb388fd5 Mon Sep 17 00:00:00 2001 From: Beatrice-betty Date: Mon, 11 May 2026 17:57:30 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E9=99=90=E5=88=B6=E5=8F=8A?= =?UTF-8?q?=E6=9B=B4=E6=94=B9=E6=8E=A8=E9=80=81=E6=A0=B7=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- module/os/tasks/fleet_auto_change.py | 41 ++++++++ module/os/tasks/hazard_leveling.py | 142 +++++++++++++++++++++++++-- 2 files changed, 173 insertions(+), 10 deletions(-) diff --git a/module/os/tasks/fleet_auto_change.py b/module/os/tasks/fleet_auto_change.py index e9ab62147..06ed65898 100644 --- a/module/os/tasks/fleet_auto_change.py +++ b/module/os/tasks/fleet_auto_change.py @@ -59,6 +59,8 @@ class OpsiFleetAutoChange(CoinTaskMixin, DockMixin, OSMap): 3. 获取自定义舰位配置 4. 执行自动配队 5. 设置冷却时间 + 6. 运行经验检测 + 7. 推送结果 注意：此方法由经验检测触发，不需要重新收集舰船数据 """ @@ -76,11 +78,50 @@ class OpsiFleetAutoChange(CoinTaskMixin, DockMixin, OSMap): self._set_cooldown() logger.info("自动配队完成") + + self._run_exp_check_after_auto_change(custom_positions) + + self._notify_auto_change_complete(custom_positions) + except Exception as e: logger.error(f"自动配队执行失败: {e}") self._handle_auto_change_error(str(e)) raise + def _run_exp_check_after_auto_change(self, custom_positions): + """ + 自动配队后运行经验检测 + + Args: + custom_positions: 自定义舰位列表 + """ + logger.info("自动配队后运行经验检测") + + try: + from module.os.tasks.hazard_leveling import OpsiHazard1Leveling + + leveling = OpsiHazard1Leveling(config=self.config, device=self.device) + leveling.os_check_leveling() + logger.info("经验检测完成") + except Exception as e: + logger.warning(f"经验检测失败: {e}") + + def _notify_auto_change_complete(self, custom_positions): + """ + 推送自动配队完成通知 + + Args: + custom_positions: 自定义舰位列表 + """ + try: + positions_str = ', '.join(map(str, custom_positions)) + self.notify_push( + title="大世界自动配队完成", + content=f"<{self.config.config_name}>\n\n已更换舰位: {positions_str}\n\n自动配队冷却时间: {self.config.OpsiFleetAutoChange_CooldownHours} 小时" + ) + except Exception as e: + logger.warning(f"推送通知失败: {e}") + def _goto_azur_port(self): """前往最近的碧蓝航线港口""" logger.info("前往碧蓝航线港口") diff --git a/module/os/tasks/hazard_leveling.py b/module/os/tasks/hazard_leveling.py index b368c0bea..3c442f13c 100644 --- a/module/os/tasks/hazard_leveling.py +++ b/module/os/tasks/hazard_leveling.py @@ -390,9 +390,17 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): logger.warning(f"自定义舰位格式错误: {custom_positions_str}，将检测所有舰船") custom_positions = [] - ship_data_list = self._collect_ship_data_with_retry(target_level) - if ship_data_list['ships'] is None: - error_msg = ship_data_list['error'] or "未知错误" + if not self._check_auto_change_prerequisite(enable_custom_check, custom_positions): + logger.info("自动配队前置条件不满足，禁用自动配队") + self.config.OpsiFleetAutoChange_Enable = False + + if enable_custom_check and custom_positions: + ship_data_result = self._collect_custom_positions_data(target_level, custom_positions) + else: + ship_data_result = self._collect_ship_data_with_retry(target_level) + + if ship_data_result['ships'] is None: + error_msg = ship_data_result['error'] or "未知错误" logger.error(f"舰船数据收集失败: {error_msg}") report = self._format_check_report( None, target_level, self.config.OpsiFleet_Fleet, error_msg=error_msg @@ -405,7 +413,7 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): logger.info("检测失败，下次检测时间设为24小时后") return - ships = ship_data_list['ships'] + ships = ship_data_result['ships'] try: from module.statistics.ship_exp_stats import save_ship_exp_data @@ -432,7 +440,7 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): logger.warning(f"保存舰船经验数据失败: {e}") report = self._format_check_report( - ships, target_level, self.config.OpsiFleet_Fleet + ships, target_level, self.config.OpsiFleet_Fleet, custom_positions=custom_positions if enable_custom_check else None ) self.notify_push( title="舰船经验检测报告", @@ -474,7 +482,32 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): self.config.OpsiCheckLeveling_LastRun = datetime.now().replace(microsecond=0) - def _format_check_report(self, ship_data_list, target_level, fleet_index, error_msg=None): + def _check_auto_change_prerequisite(self, enable_custom_check, custom_positions): + """ + 检查自动配队前置条件 + + Args: + enable_custom_check: 是否启用自定义舰船检测 + custom_positions: 自定义舰位列表 + + Returns: + bool: 是否满足前置条件 + """ + if not self.config.OpsiFleetAutoChange_Enable: + return True + + if not enable_custom_check: + logger.warning("自动配队需要启用自定义舰船检测，将禁用自动配队") + return False + + if not custom_positions: + logger.warning("自动配队需要有效的自定义舰位配置，将禁用自动配队") + return False + + logger.info(f"自动配队前置条件满足: 启用自定义检测，舰位 {custom_positions}") + return True + + def _format_check_report(self, ship_data_list, target_level, fleet_index, error_msg=None, custom_positions=None): """ 格式化检测报告，用于推送通知 @@ -483,6 +516,7 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): target_level: 目标等级 fleet_index: 舰队索引 error_msg: 错误信息，成功时为None + custom_positions: 自定义舰位列表，None时显示所有舰船 Returns: str: 格式化的报告文本 @@ -499,6 +533,8 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): lines.append("检测状态: 成功") lines.append(f"检测舰队: 第 {fleet_index} 舰队") lines.append(f"目标等级: Lv.{target_level}") + if custom_positions: + lines.append(f"检测舰位: {', '.join(map(str, custom_positions))}") lines.append("") target_exp = LIST_SHIP_EXP[target_level - 1] if 1 <= target_level <= 125 else 0 @@ -512,7 +548,11 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): except Exception: exp_per_hour = 22000.0 - for ship in ship_data_list: + ships_to_report = ship_data_list + if custom_positions: + ships_to_report = [s for s in ship_data_list if s.get('position') in custom_positions] + + for ship in ships_to_report: position = ship.get('position', 0) level = ship.get('level', 0) current_exp = ship.get('current_exp', 0) @@ -545,17 +585,99 @@ class OpsiHazard1Leveling(CoinTaskMixin, OSMap): lines.append(f"舰位{position}: Lv.{level} | 经验：{current_exp:,} | 进度：{status} │ 预计时间：{time_str}") lines.append("") - all_full = all(ship.get('total_exp', 0) >= target_exp for ship in ship_data_list) + all_full = all(ship.get('total_exp', 0) >= target_exp for ship in ships_to_report) if all_full: - lines.append("★ 所有舰船已满经验！") + if custom_positions: + lines.append(f"★ 指定舰位 {', '.join(map(str, custom_positions))} 已满经验！") + else: + lines.append("★ 所有舰船已满经验！") else: - not_full = [s for s in ship_data_list if s.get('total_exp', 0) < target_exp] + not_full = [s for s in ships_to_report if s.get('total_exp', 0) < target_exp] lines.append(f"未满经验舰位: {len(not_full)} 艘") lines.append(f"检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") return "\n".join(lines) + def _collect_custom_positions_data(self, target_level, custom_positions): + """ + 收集指定舰位的舰船数据 + + Args: + target_level: 目标等级 + custom_positions: 自定义舰位列表，如 [1, 3, 6] + + Returns: + dict: {'ships': list, 'error': str} + ships为舰船数据列表，失败时为None + error为错误信息，成功时为None + """ + from module.os_handler.assets import ( + OS_FLEET_SLOT_NAV_1_BUTTON, + OS_FLEET_SLOT_NAV_2_BUTTON, + OS_FLEET_SLOT_NAV_3_BUTTON, + OS_FLEET_SLOT_NAV_4_BUTTON, + OS_FLEET_SLOT_NAV_5_BUTTON, + OS_FLEET_SLOT_NAV_6_BUTTON, + ) + + logger.info(f"开始收集指定舰位数据: {custom_positions}") + + slot_buttons = { + 1: OS_FLEET_SLOT_NAV_1_BUTTON, + 2: OS_FLEET_SLOT_NAV_2_BUTTON, + 3: OS_FLEET_SLOT_NAV_3_BUTTON, + 4: OS_FLEET_SLOT_NAV_4_BUTTON, + 5: OS_FLEET_SLOT_NAV_5_BUTTON, + 6: OS_FLEET_SLOT_NAV_6_BUTTON, + } + + ship_data_list = [] + + self.fleet_set(self.config.OpsiFleet_Fleet) + + for position in sorted(custom_positions): + button = slot_buttons.get(position) + if not button: + logger.warning(f"无效的舰位: {position}") + continue + + logger.info(f"检测舰位 {position}") + + self.equip_enter(button, check_button=EQUIPMENT_OPEN, long_click=True) + + self.device.screenshot() + level, exp = ship_info_get_level_exp(main=self) + + if level < 1 or level > len(LIST_SHIP_EXP): + logger.warning(f"舰位 {position} 等级识别异常: {level}") + ship_data_list.append({ + "position": position, + "level": level, + "current_exp": exp, + "total_exp": 0, + }) + else: + total_exp = LIST_SHIP_EXP[level - 1] + exp + logger.info( + f"舰位 {position}: 等级 {level}, 经验 {exp}, 总经验 {total_exp}" + ) + ship_data_list.append({ + "position": position, + "level": level, + "current_exp": exp, + "total_exp": total_exp, + }) + + self.ui_back(check_button=self.is_in_map) + self.device.sleep(0.5) + + if not ship_data_list: + return {'ships': None, 'error': '未收集到任何舰船数据'} + + logger.info(f"指定舰位数据收集完成，共 {len(ship_data_list)} 艘") + return {'ships': ship_data_list, 'error': None} + def _collect_ship_data_with_retry(self, target_level): """ 收集舰船数据，带重试机制