mirror of
https://github.com/sui-feng-cb/AzurLaneAutoScript1.git
synced 2026-07-22 14:01:52 +08:00
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from module.config.config import AzurLaneConfig
|
|
from module.logger import logger
|
|
from module.smart_mgmt.downtime_fetcher import DowntimeFetcher, FetchError
|
|
from module.smart_mgmt.fetcher_test import DowntimeFetcherTest
|
|
|
|
|
|
DOWNTIME_FORMAT_RE = re.compile(
|
|
r'(\d{4})-(\d{1,2})-(\d{1,2}) '
|
|
r'(\d{1,2}):(\d{2})~(\d{1,2}):(\d{2})',
|
|
)
|
|
|
|
|
|
def parse_downtime(window: str) -> Optional[datetime]:
|
|
"""
|
|
Args:
|
|
window (str): Window string in format 'YYYY-MM-DD HH:MM~HH:MM'.
|
|
|
|
Returns:
|
|
Optional[datetime]:
|
|
End datetime parsed from the window, or None if no pattern matched.
|
|
"""
|
|
match = DOWNTIME_FORMAT_RE.search(window)
|
|
if not match:
|
|
logger.warning(f'Failed to parse downtime window: {window}')
|
|
return None
|
|
year, month, day = int(match.group(1)), int(match.group(2)), int(match.group(3))
|
|
end_hour, end_minute = int(match.group(6)), int(match.group(7))
|
|
end_dt = datetime(year, month, day, end_hour, end_minute)
|
|
logger.info(f'Parsed downtime window: end={end_dt:%Y-%m-%d %H:%M}')
|
|
return end_dt
|
|
|
|
|
|
def update_downtime(config: AzurLaneConfig, force_refresh=False) -> bool:
|
|
"""Fetch downtime and write config."""
|
|
try:
|
|
payload = DowntimeFetcher().run(force_refresh=force_refresh)
|
|
except FetchError as e:
|
|
# Expected operational failure degrade
|
|
logger.warning(f'Failed to fetch downtime: {e}')
|
|
return False
|
|
except Exception as e:
|
|
# Unexpected programming error; full traceback for diagnosis
|
|
logger.exception(f'Unexpected error: {e}')
|
|
return False
|
|
with config.multi_set():
|
|
config.cross_set('DowntimeMgmt.Downtime.Window', payload.get('window', ''))
|
|
config.cross_set('DowntimeMgmt.Downtime.Title', payload.get('title', ''))
|
|
return True
|
|
|
|
|
|
def get_downtime_end(config: AzurLaneConfig) -> Optional[datetime]:
|
|
task = config.task.command
|
|
if not config.cross_get(f'DowntimeMgmt.DowntimeStrategy.Manage{task}', False):
|
|
return None
|
|
if config.cross_get('DowntimeMgmt.DowntimeStrategy.AutoFetch', False):
|
|
if config.SERVER == 'cn':
|
|
update_downtime(config)
|
|
else:
|
|
logger.warning('Auto-fetch skipped as fetcher is for CN only, use window from config')
|
|
else:
|
|
logger.info('Auto-fetch disabled, use window from config')
|
|
window = config.cross_get('DowntimeMgmt.Downtime.Window', None)
|
|
if not window:
|
|
return None
|
|
return parse_downtime(window)
|
|
|
|
|
|
def run_downtime_fetch(config: AzurLaneConfig):
|
|
# Tool task entry is invoked with config bound to Alas by default,
|
|
# rebind to DowntimeFetch.
|
|
config.init_task('DowntimeFetch')
|
|
if config.DowntimeFetch_FetcherTest:
|
|
DowntimeFetcherTest().run()
|
|
else:
|
|
if config.SERVER != 'cn':
|
|
logger.warning('Downtime fetch targets CN Bilibili notices, data may be irrelevant on other servers')
|
|
update_downtime(config, force_refresh=True)
|