AzurLaneAutoScript/module/smart_mgmt/downtime_fetch.py

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)