diff --git a/module/campaign/campaign_base.py b/module/campaign/campaign_base.py index be3dfaae0..b54c0b012 100644 --- a/module/campaign/campaign_base.py +++ b/module/campaign/campaign_base.py @@ -37,6 +37,24 @@ class CampaignBase(CampaignUI, Map, AutoSearchCombat): """ FUNCTION_NAME_BASE = 'battle_' MAP: CampaignMap + low_emotion_withdrawn = False + + def handle_combat_low_emotion(self): + """处理战役中的低心情强制出击提示。 + + 仅在“计算心情消耗”模式下取消提示并撤退；其他心情设置维持原行为。 + """ + if self.config.Emotion_Mode != 'calculate': + return super().handle_combat_low_emotion() + + if self.handle_popup_cancel('IGNORE_LOW_EMOTION'): + logger.hr('低心情撤退') + logger.info('[低心情] 已取消强制出击，退出战斗准备后撤退当前地图') + self.low_emotion_withdrawn = True + # 不能在战斗准备页内嵌撤退循环；由 CampaignRun 退出此页面后统一撤退。 + raise CampaignEnd('LowEmotionWithdraw') + + return False def battle_default(self): """默认战斗策略：清除所有敌人。 diff --git a/module/campaign/run.py b/module/campaign/run.py index 3aaaa87b0..85177d600 100644 --- a/module/campaign/run.py +++ b/module/campaign/run.py @@ -21,6 +21,7 @@ from module.shop.shop_status import ShopStatus from module.campaign.campaign_ui import MODE_SWITCH_1 from module.config.config import AzurLaneConfig +from module.config.task_priority import parse_task_priority from module.exception import CampaignEnd, RequestHumanTakeover, ScriptEnd from module.handler.fast_forward import map_files, to_map_file_name from module.logger import logger @@ -418,6 +419,69 @@ def after_campaign_run(self): """单次战役完成后的扩展钩子。""" pass + def get_low_emotion_next_event_task(self, task): + """从任务优先级配置中获取当前活动图的后继活动图任务。""" + event_tasks = [ + candidate + for candidate in parse_task_priority(self.config.SCHEDULER_PRIORITY) + if candidate == 'Event' or (candidate.startswith('Event') and candidate[5:].isdigit()) + ] + try: + return event_tasks[event_tasks.index(task) + 1] + except (IndexError, ValueError): + return None + + def handle_low_emotion_withdrawal(self): + """延后因低心情撤退的当前任务，并切换到后续活动图。""" + if not getattr(self.campaign, 'low_emotion_withdrawn', False): + return False + + task = self.config.task.command + try: + # 低心情弹窗可能出现在战斗准备页。先退出该页面，回到地图后再撤退。 + self.campaign.withdraw(skip_first_screenshot=False) + except CampaignEnd: + pass + + emotion = self.campaign.emotion + emotion.update() + if emotion.using_public: + public_fleet = getattr(emotion, 'public_fleet', None) + fleets = [public_fleet] if public_fleet is not None else [] + else: + # 初次出击时地图还未初始化，无法可靠判断是哪一队触发弹窗。 + # 双舰队配置下保守地延后两队，避免按错误舰队提前重试。 + fleets = list(getattr(emotion, 'fleets', [])) + if not self.config.FLEET_2: + fleets = fleets[:1] + + if not fleets: + logger.critical('[低心情] 未找到舰队心情记录，无法安全延后当前任务') + raise RequestHumanTakeover + + # 游戏已显示低心情强制出击弹窗，说明本地记录的心情值已经失真。 + # 不能继续用过高的旧值（例如 75）计算，否则会把当前任务排回现在。 + for fleet in fleets: + fleet.current = 0 + emotion.record() + emotion.show() + recovered = emotion.get_recovered_for_battle(self.campaign._map_battle) + + fleet_names = ', '.join(str(fleet.fleet) for fleet in fleets) + logger.info( + f'[低心情] 游戏端低心情，{task} 的 {fleet_names} 队按 0 心情恢复至 {recovered}' + ) + self.config.task_delay(target=recovered) + + # 后继活动图由用户的任务优先级配置决定；最后一张活动图交由调度器处理。 + next_event = self.get_low_emotion_next_event_task(task) + if next_event and self.config.task_call(next_event, force_call=False): + logger.info(f'[低心情] {task} 已撤退，立即切换到 {next_event}') + self.config.update() + + self.campaign.low_emotion_withdrawn = False + return True + def handle_commission_notice(self): """ 检查委托通知。如果发现委托完成，停止当前任务并调用委托处理。 @@ -511,6 +575,9 @@ def run(self, name, folder='campaign_main', mode='normal', total=0): self.device.click_record_clear() try: self.campaign.run() + except CampaignEnd: + if not getattr(self.campaign, 'low_emotion_withdrawn', False): + raise except ScriptEnd as e: logger.hr('脚本结束') logger.info(str(e)) @@ -519,6 +586,9 @@ def run(self, name, folder='campaign_main', mode='normal', total=0): self.config.Scheduler_Enable = False break + if self.handle_low_emotion_withdrawal(): + self.config.task_stop('[低心情] 已撤退并延后当前任务') + # 更新配置 if len(self.campaign.config.modified): logger.info('[战役-运行] 更新仪表盘配置') diff --git a/module/combat/emotion.py b/module/combat/emotion.py index bc67ec658..bbddabf7e 100644 --- a/module/combat/emotion.py +++ b/module/combat/emotion.py @@ -336,23 +336,16 @@ def reduce_per_battle_before_entering(self): def reduce_shipwreck(self): return 10 - def _check_reduce(self, battle): - """检查战斗带来的情绪减少。 + def get_recovered_for_battle(self, battle): + """计算完成下一次战役所需的情绪恢复时间。 Returns: - recovered (datetime): 预期恢复时间。 - delay (bool): 是否需要延迟。 + datetime: 扣除下一次战役的预计情绪后达到控制阈值的时间。 """ if self.using_public: reduce = battle * self.reduce_per_battle_before_entering logger.info(f'[情绪-检查] 预期情绪扣减: {reduce}') - - self.update() - self.record() - self.show() - recovered = self.public_fleet.get_recovered(reduce) - delay = recovered > current_time() - return recovered, delay + return self.public_fleet.get_recovered(reduce) method = self.config.Fleet_FleetOrder @@ -369,13 +362,15 @@ def _check_reduce(self, battle): battle = tuple(np.array(battle) * self.reduce_per_battle_before_entering) logger.info(f'[情绪-检查] 预期情绪扣减: {battle}') + return max([f.get_recovered(b) for f, b in zip(self.fleets, battle)]) + def _check_reduce(self, battle): + """检查战斗带来的情绪减少。""" self.update() self.record() self.show() - recovered = max([f.get_recovered(b) for f, b in zip(self.fleets, battle)]) - delay = recovered > current_time() - return recovered, delay + recovered = self.get_recovered_for_battle(battle) + return recovered, recovered > current_time() def check_reduce(self, battle): """进入战役前检查情绪。 diff --git a/module/map/map_operation.py b/module/map/map_operation.py index a18c27daa..ece589a6d 100644 --- a/module/map/map_operation.py +++ b/module/map/map_operation.py @@ -17,6 +17,7 @@ import cv2 from module.base.timer import Timer +from module.combat.assets import BATTLE_PREPARATION from module.exception import CampaignEnd, RequestHumanTakeover, ScriptEnd from module.handler.fast_forward import FastForwardHandler from module.handler.mystery import MysteryHandler @@ -44,6 +45,12 @@ class MapOperation(MysteryHandler, FleetPreparation, Retirement, FastForwardHand map_cat_attack_timer = Timer(2) map_clear_percentage_prev = -1 map_clear_percentage_timer = Timer(0.3, count=1) + WITHDRAW_RESULT_HANDLERS = ( + 'handle_battle_status', + 'handle_exp_info', + 'handle_get_ship', + 'handle_get_items', + ) # 屏幕上显示的舰队编号。 fleet_show_index = 1 @@ -301,11 +308,7 @@ def enter_map_cancel(self, skip_first_screenshot=True): if self.is_in_stage(): break - if self.appear(MAP_PREPARATION, offset=(20, 20), interval=2): - self.device.click(MAP_PREPARATION_CANCEL) - continue - if self.appear(FLEET_PREPARATION, offset=(20, 50), interval=2): - self.device.click(MAP_PREPARATION_CANCEL) + if self.handle_enter_map_preparation_cancel(): continue return True @@ -451,6 +454,12 @@ def withdraw(self, skip_first_screenshot=True): else: self.device.screenshot() + if self.handle_withdraw_battle_preparation(): + continue + if self.handle_enter_map_preparation_cancel(): + continue + if self.handle_withdraw_result(): + continue if self.appear_then_click(FLEET_SWITCH_CONFIRM, offset=(30, 30)): continue if self.handle_popup_confirm('WITHDRAW'): @@ -469,6 +478,35 @@ def withdraw(self, skip_first_screenshot=True): if self.handle_in_stage(): raise CampaignEnd('Withdraw') + def handle_withdraw_battle_preparation(self): + """退出撤退前仍残留的战斗准备页。""" + if self.appear(BATTLE_PREPARATION, offset=(20, 20), interval=2): + logger.info(f'{BATTLE_PREPARATION} -> {BACK_ARROW}') + self.device.click(BACK_ARROW) + return True + return False + + def handle_enter_map_preparation_cancel(self): + """从舰队准备或地图准备页退回关卡选择页。""" + if self.appear(MAP_PREPARATION, offset=(20, 20), interval=2): + logger.info(f'{MAP_PREPARATION} -> {MAP_PREPARATION_CANCEL}') + self.device.click(MAP_PREPARATION_CANCEL) + return True + if self.appear(FLEET_PREPARATION, offset=(20, 50), interval=2): + logger.info(f'{FLEET_PREPARATION} -> {MAP_PREPARATION_CANCEL}') + self.device.click(MAP_PREPARATION_CANCEL) + return True + return False + + def handle_withdraw_result(self): + """推进撤退过程中可能出现的战斗结算页面。""" + # MapOperation 也由不混入 Combat 的任务复用，结算处理器在该场景下是可选的。 + for name in self.WITHDRAW_RESULT_HANDLERS: + handler = getattr(self, name, None) + if handler and handler(): + return True + return self.handle_popup_confirm('COMBAT_STATUS') + def handle_map_cat_attack(self): """ 处理猫猫攻击动画，点击跳过。 diff --git a/tests/test_low_emotion_withdraw.py b/tests/test_low_emotion_withdraw.py new file mode 100644 index 000000000..4925904c1 --- /dev/null +++ b/tests/test_low_emotion_withdraw.py @@ -0,0 +1,320 @@ +from datetime import datetime +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +# 本地旧虚拟环境的 RapidOCR 仍未提供源码要求的 PPOCRV6 枚举。 +# 仅为加载待测战役模块补齐别名，不影响生产运行时的依赖版本。 +from rapidocr import OCRVersion + +if not hasattr(OCRVersion, 'PPOCRV6'): + OCRVersion.PPOCRV6 = OCRVersion.PPOCRV5 + +from module.campaign.campaign_base import CampaignBase +from module.campaign.run import CampaignRun +from module.combat.emotion import Emotion +from module.event.maritime_escort import MaritimeEscort +from module.exception import CampaignEnd, RequestHumanTakeover +from module.map.assets import FLEET_PREPARATION, MAP_PREPARATION, MAP_PREPARATION_CANCEL +from module.map.map_operation import MapOperation + + +class TestLowEmotionWithdraw(unittest.TestCase): + EVENT_PRIORITY = 'Event > Event2 > Event3' + + def test_calculate_mode_cancels_then_exits_combat_preparation(self): + campaign = CampaignBase.__new__(CampaignBase) + campaign.config = SimpleNamespace(Emotion_Mode='calculate') + campaign.handle_popup_cancel = Mock(return_value=True) + campaign.withdraw = Mock() + + with self.assertRaises(CampaignEnd): + campaign.handle_combat_low_emotion() + + campaign.handle_popup_cancel.assert_called_once_with('IGNORE_LOW_EMOTION') + campaign.withdraw.assert_not_called() + self.assertTrue(campaign.low_emotion_withdrawn) + + def test_non_calculate_mode_keeps_original_confirm_behavior(self): + campaign = CampaignBase.__new__(CampaignBase) + campaign.config = SimpleNamespace(Emotion_Mode='calculate_ignore') + campaign.__dict__['emotion'] = SimpleNamespace(is_ignore=True) + campaign.handle_popup_cancel = Mock() + campaign.handle_popup_confirm = Mock(return_value=True) + campaign.interval_reset = Mock() + + self.assertTrue(campaign.handle_combat_low_emotion()) + + campaign.handle_popup_cancel.assert_not_called() + campaign.handle_popup_confirm.assert_called_once_with('IGNORE_LOW_EMOTION') + + def test_withdrawal_delays_current_task_to_emotion_recovery(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + fleet = Mock() + fleet.fleet = 1 + fleet.current = 75 + fleet.get_recovered.return_value = recovered + emotion = Mock(using_public=False) + emotion.fleets = [fleet, Mock()] + emotion.get_recovered_for_battle.return_value = recovered + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + _map_battle=5, + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event' + runner.config.FLEET_2 = 0 + runner.config.SCHEDULER_PRIORITY = self.EVENT_PRIORITY + + self.assertTrue(runner.handle_low_emotion_withdrawal()) + + campaign.withdraw.assert_called_once_with(skip_first_screenshot=False) + emotion.update.assert_called_once_with() + emotion.record.assert_called_once_with() + emotion.show.assert_called_once_with() + self.assertEqual(fleet.current, 0) + emotion.get_recovered_for_battle.assert_called_once_with(5) + runner.config.task_delay.assert_called_once_with(target=recovered) + runner.config.task_call.assert_called_once_with('Event2', force_call=False) + runner.config.update.assert_called_once_with() + self.assertFalse(campaign.low_emotion_withdrawn) + + def test_withdrawal_from_second_event_runs_third_event(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + fleet = Mock() + emotion = Mock(using_public=False) + emotion.fleets = [fleet] + emotion.get_recovered_for_battle.return_value = recovered + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + _map_battle=4, + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event2' + runner.config.FLEET_2 = 0 + runner.config.SCHEDULER_PRIORITY = self.EVENT_PRIORITY + + self.assertTrue(runner.handle_low_emotion_withdrawal()) + + runner.config.task_call.assert_called_once_with('Event3', force_call=False) + runner.config.update.assert_called_once_with() + emotion.get_recovered_for_battle.assert_called_once_with(4) + + def test_withdrawal_uses_public_fleet_emotion(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + public_fleet = Mock(fleet='Public', current=75) + public_fleet.get_recovered.return_value = recovered + fleet_1 = Mock(fleet=1, current=80) + fleet_2 = Mock(fleet=2, current=90) + emotion = Mock(using_public=True, public_fleet=public_fleet) + emotion.fleets = [fleet_1, fleet_2] + emotion.get_recovered_for_battle.return_value = recovered + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + _map_battle=3, + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event3' + runner.config.FLEET_2 = 2 + + self.assertTrue(runner.handle_low_emotion_withdrawal()) + + self.assertEqual(public_fleet.current, 0) + self.assertEqual(fleet_1.current, 80) + self.assertEqual(fleet_2.current, 90) + emotion.get_recovered_for_battle.assert_called_once_with(3) + runner.config.task_delay.assert_called_once_with(target=recovered) + + def test_withdrawal_without_fleet_record_requires_human_takeover(self): + emotion = Mock(using_public=False) + emotion.fleets = [] + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event' + runner.config.FLEET_2 = 0 + + with self.assertRaises(RequestHumanTakeover): + runner.handle_low_emotion_withdrawal() + + runner.config.task_delay.assert_not_called() + + def test_withdrawal_from_third_event_returns_to_scheduler_queue(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + fleet = Mock() + emotion = Mock(using_public=False) + emotion.fleets = [fleet] + emotion.get_recovered_for_battle.return_value = recovered + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + _map_battle=2, + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event3' + runner.config.FLEET_2 = 0 + + self.assertTrue(runner.handle_low_emotion_withdrawal()) + + emotion.get_recovered_for_battle.assert_called_once_with(2) + runner.config.task_delay.assert_called_once_with(target=recovered) + runner.config.task_call.assert_not_called() + runner.config.update.assert_not_called() + + def test_withdrawal_delays_both_configured_fleets(self): + first_recovered = datetime(2026, 8, 14, 12, 0, 0) + second_recovered = datetime(2026, 8, 14, 12, 12, 0) + fleet_1 = Mock(fleet=1, current=75) + fleet_1.get_recovered.return_value = first_recovered + fleet_2 = Mock(fleet=2, current=90) + fleet_2.get_recovered.return_value = second_recovered + emotion = Mock(using_public=False) + emotion.fleets = [fleet_1, fleet_2] + emotion.get_recovered_for_battle.return_value = second_recovered + campaign = SimpleNamespace( + low_emotion_withdrawn=True, + emotion=emotion, + withdraw=Mock(side_effect=CampaignEnd('Withdraw')), + _map_battle=6, + ) + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['campaign'] = campaign + runner.__dict__['config'] = Mock() + runner.config.task.command = 'Event' + runner.config.FLEET_2 = 2 + runner.config.SCHEDULER_PRIORITY = self.EVENT_PRIORITY + + self.assertTrue(runner.handle_low_emotion_withdrawal()) + + self.assertEqual(fleet_1.current, 0) + self.assertEqual(fleet_2.current, 0) + emotion.get_recovered_for_battle.assert_called_once_with(6) + runner.config.task_delay.assert_called_once_with(target=second_recovered) + + def test_next_event_task_follows_scheduler_priority_configuration(self): + runner = CampaignRun.__new__(CampaignRun) + runner.__dict__['config'] = SimpleNamespace( + SCHEDULER_PRIORITY='Event3 > Main > Event > Event2 > Raid' + ) + + self.assertEqual(runner.get_low_emotion_next_event_task('Event'), 'Event2') + self.assertIsNone(runner.get_low_emotion_next_event_task('Event2')) + + def test_recovery_for_battle_includes_next_sortie_emotion_cost(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + public_fleet = Mock() + public_fleet.get_recovered.return_value = recovered + emotion = Emotion.__new__(Emotion) + emotion.config = SimpleNamespace(Campaign_Use2xBook=False) + emotion.using_public = True + emotion.public_fleet = public_fleet + emotion.map_is_2x_book = False + emotion.update = Mock() + emotion.record = Mock() + emotion.show = Mock() + + self.assertEqual(emotion.get_recovered_for_battle(5), recovered) + + public_fleet.get_recovered.assert_called_once_with(10) + emotion.update.assert_not_called() + emotion.record.assert_not_called() + emotion.show.assert_not_called() + + def test_check_reduce_updates_and_records_emotion_once(self): + recovered = datetime(2026, 8, 14, 12, 0, 0) + emotion = Emotion.__new__(Emotion) + emotion.update = Mock() + emotion.record = Mock() + emotion.show = Mock() + emotion.get_recovered_for_battle = Mock(return_value=recovered) + + with patch('module.combat.emotion.current_time', return_value=datetime(2026, 8, 14, 13, 0, 0)): + self.assertEqual(emotion._check_reduce(5), (recovered, False)) + + emotion.update.assert_called_once_with() + emotion.record.assert_called_once_with() + emotion.show.assert_called_once_with() + emotion.get_recovered_for_battle.assert_called_once_with(5) + + def test_withdraw_processes_battle_result_before_waiting_for_stage(self): + operation = MapOperation.__new__(MapOperation) + operation.device = SimpleNamespace(screenshot=Mock()) + operation.handle_battle_status = Mock(side_effect=[True, False]) + operation.handle_exp_info = Mock(return_value=False) + operation.handle_get_ship = Mock(return_value=False) + operation.handle_get_items = Mock(return_value=False) + operation.handle_popup_confirm = Mock(return_value=False) + operation.appear_then_click = Mock(return_value=False) + operation.handle_auto_search_exit = Mock(return_value=False) + operation.appear = Mock(return_value=False) + operation.handle_in_stage = Mock(return_value=True) + + with self.assertRaises(CampaignEnd): + operation.withdraw(skip_first_screenshot=False) + + operation.handle_battle_status.assert_called() + operation.handle_in_stage.assert_called_once_with() + + def test_withdraw_exits_battle_preparation_before_opening_withdraw_menu(self): + operation = MapOperation.__new__(MapOperation) + operation.device = SimpleNamespace(click=Mock()) + operation.appear = Mock(return_value=True) + + self.assertTrue(operation.handle_withdraw_battle_preparation()) + + operation.device.click.assert_called_once() + + def test_withdraw_exits_initial_preparation_pages(self): + for preparation_page in (MAP_PREPARATION, FLEET_PREPARATION): + with self.subTest(preparation_page=preparation_page): + operation = MapOperation.__new__(MapOperation) + operation.device = SimpleNamespace(click=Mock()) + operation.appear = Mock(side_effect=lambda button, **_: button is preparation_page) + + self.assertTrue(operation.handle_enter_map_preparation_cancel()) + + operation.device.click.assert_called_once_with(MAP_PREPARATION_CANCEL) + + def test_handle_withdraw_result_falls_back_to_combat_status_popup(self): + operation = MapOperation.__new__(MapOperation) + operation.handle_battle_status = Mock(return_value=False) + operation.handle_exp_info = Mock(return_value=False) + operation.handle_get_ship = Mock(return_value=False) + operation.handle_get_items = Mock(return_value=False) + operation.handle_popup_confirm = Mock(return_value=True) + + self.assertTrue(operation.handle_withdraw_result()) + + operation.handle_popup_confirm.assert_called_once_with('COMBAT_STATUS') + + def test_withdraw_result_supports_non_combat_maritime_escort(self): + escort = MaritimeEscort.__new__(MaritimeEscort) + escort.handle_popup_confirm = Mock(return_value=True) + + self.assertTrue(escort.handle_withdraw_result()) + + escort.handle_popup_confirm.assert_called_once_with('COMBAT_STATUS') + + +if __name__ == '__main__': + unittest.main()