Feat: clear exercise attempts based on downtime

module/exercise/exercise.py

@@ -1,11 +1,11 @@
 import datetime
-from module.config.utils import get_server_last_update
+from module.config.utils import get_server_next_update, get_server_last_update
 from module.exercise.assets import *
 from module.exercise.combat import ExerciseCombat
 from module.logger import logger
 from module.ocr.ocr import Digit, Ocr, OcrYuv
 from module.ui.page import page_exercise
-from module.config.utils import get_server_next_update
+from module.smart_mgmt.utils import parse_downtime
 
 class DatedDuration(Ocr):
     def __init__(self, buttons, lang='cnocr', letter=(255, 255, 255), threshold=128, alphabet='0123456789:IDS天日d',
@@ -77,6 +77,7 @@ class Exercise(ExerciseCombat):
     opponent_change_count = 0
     remain = 0
     preserve = 0
+    DOWNTIME_END_HOUR_THRESHOLD = 15
 
     def _new_opponent(self):
         logger.info('New opponent')
@@ -171,18 +172,6 @@ class Exercise(ExerciseCombat):
             self.config.set_record(Exercise_OpponentRefreshValue=0)
             return 0
 
-    def _get_thursday_strategy(self, remain_time):
-        """
-        Args:
-            remain_time(datetime.timedelta):
-
-        Returns:
-            bool: if use Thursday strategy
-        """
-        if 96 > remain_time >= 84 or 264 > remain_time >= 252:
-            return True
-        return False
-
     def _get_exercise_reset_remain(self):
         """
         Returns:
@@ -206,7 +195,43 @@ class Exercise(ExerciseCombat):
 
         return preserve, admiral_interval
 
-    def _should_force_clear(self, remain_time, admiral_interval):
+    def _get_thursday_strategy(self, remain_hours):
+        """
+        Args:
+            remain_hours(int):
+
+        Returns:
+            bool: if use Thursday strategy
+        """
+        if 96 > remain_hours >= 84 or 264 > remain_hours >= 252:
+            return True
+        return False
+
+    def _get_downtime_strategy(self, now):
+        """
+        Check smart_mgmt downtime window to decide whether to clear attempts.
+
+        Returns:
+            bool: 
+            True if today is maintenance day with end time at or after DOWNTIME_END_HOUR_THRESHOLD.
+            False if available but today is not maintenance day.
+            None if unavailable (disabled, non-CN server, window missing, or parse failed).
+        """
+        if self.config.SERVER != 'cn':
+            return None
+        if not self.config.cross_get('DowntimeMgmt.DowntimeStrategy.ManageExercise', False):
+            return None
+        window = self.config.cross_get('DowntimeMgmt.Downtime.Window', None)
+        if not window:
+            return None
+        end_dt = parse_downtime(window)
+        if end_dt is None:
+            return None
+        if end_dt <= now:
+            return False
+        return end_dt.date() == now.date() and end_dt.hour >= self.DOWNTIME_END_HOUR_THRESHOLD
+
+    def _should_force_clear(self, now, remain_time, admiral_interval):
         """Decide whether to clear all exercise attempts this run."""
         if admiral_interval is None or not remain_time:
             return False
@@ -214,9 +239,16 @@ class Exercise(ExerciseCombat):
         remain_hours = int(remain_time.total_seconds() // 3600)
         admiral_start, admiral_end = admiral_interval
 
-        if self._get_thursday_strategy(remain_hours):
-            logger.info('Reach Thursday, using all attempts')
+        downtime_strategy = self._get_downtime_strategy(now)
+        if downtime_strategy is True:
+            logger.info('Downtime today, using all attempts before downtime')
             return True
+        elif downtime_strategy is None:
+            # Fallback to thursday strategy when smart_mgmt is unavailable
+            if self._get_thursday_strategy(remain_hours):
+                logger.info('Reach Thursday, using all attempts before downtime')
+                return True
+
         if admiral_start > remain_hours >= admiral_end:
             logger.info('Reach set time for admiral trial, using all attempts')
             return True
@@ -226,11 +258,11 @@ class Exercise(ExerciseCombat):
 
         return False
 
-    def _should_delay(self, next_update, forced_clear):
+    def _should_delay(self, now, next_update, forced_clear):
         """Check whether to delay exercise to the configured time before next update."""
         if forced_clear:
             return False
-        seconds_until_update = (next_update - datetime.datetime.now()).seconds
+        seconds_until_update = (next_update - now).seconds
         return seconds_until_update > 3600 * self.config.Exercise_DelayUntilHoursBeforeNextUpdate
 
     def _exercise(self, preserve):
@@ -258,15 +290,17 @@ class Exercise(ExerciseCombat):
         logger.attr('Exercise_ExerciseStrategy', self.config.Exercise_ExerciseStrategy)
         self.preserve, admiral_interval = self._get_exercise_strategy()
 
+        now = datetime.datetime.now()
         remain_time = self._get_exercise_reset_remain()
         logger.info(f'Exercise period remain: {remain_time}')
 
-        forced_clear = self._should_force_clear(remain_time, admiral_interval)
+        forced_clear = self._should_force_clear(now, remain_time, admiral_interval)
         if forced_clear:
             self.preserve = 0
+        logger.attr("exercise preserve", self.preserve)
 
         next_update = get_server_next_update(server_update)
-        should_delay = self._should_delay(next_update, forced_clear)
+        should_delay = self._should_delay(now, next_update, forced_clear)
         if should_delay:
             logger.warning(f'Exercise should run at {self.config.Exercise_DelayUntilHoursBeforeNextUpdate} '
                            f'hours before next update. Delay task to it.')
@@ -280,7 +314,6 @@ class Exercise(ExerciseCombat):
             self.config.set_record(Exercise_OpponentRefreshValue=self.opponent_change_count)
             if self.remain <= self.preserve or self.opponent_change_count >= 5:
                 next_run = next_update - datetime.timedelta(hours=self.config.Exercise_DelayUntilHoursBeforeNextUpdate)
-                now = datetime.datetime.now()
                 if next_run >= now or should_delay:
                     minutes_to_delay = int((next_run - now).total_seconds() / 60 + 1)
                     self.config.task_delay(minute=minutes_to_delay)