AzurLaneAutoScript/module/smart_mgmt/downtime_fetcher.py

from __future__ import annotations

import json
import os
import re
import time
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Optional

import requests

from deploy.atomic import atomic_read_text, atomic_write
from module.logger import logger
from module.smart_mgmt.happy_eyeballs import HappyEyeballsHTTPAdapter


class FetchError(RuntimeError):
    pass


@dataclass
class CollectionMeta:
    collection_id: int
    collection_name: str
    update_time: int
    article_count: int


@dataclass
class Article:
    pub_ts: int
    id: str
    title: str
    summary: str


@dataclass
class DowntimeWindow:
    window: str


class DowntimeFetcher:
    # Bilibili UP mid of the AzurLane official account
    DEFAULT_MID = 233114659
    DEFAULT_COLLECTION_NAME = '《碧蓝航线》维护公告'
    ARTICLE_LISTS_API = 'https://api.bilibili.com/x/article/up/lists'
    ARTICLE_COLLECTION_API = 'https://api.bilibili.com/x/article/list/web/articles'
    CN_TZ = timezone(timedelta(hours=8))
    # connect timeout is taken over by Happy Eyeballs; only the read timeout applies
    REQUEST_TIMEOUT = (None, 5.0)

    # Local cache and cross-process fetch lock
    CACHE_FILE = './log/downtime_cache.json'
    FETCH_LOCK_FILE = './log/downtime_fetch.lock'
    # Multi-instance duplicate fetch avoidance window (seconds)
    CACHE_FRESH_SECONDS = 2*3600
    # Upper bound on how long another instance may hold the fetch lock
    LOCK_WAIT_SECONDS = 25

    WS = r'[\s　]*'
    WINDOW_RE = re.compile(
        r'司令部将于' + WS +
        r'(\d{1,2})月(\d{1,2})日' + WS +
        r'(\d{1,2})[:：](\d{2})' + WS +
        r'[~～至到—\-]+' + WS +
        r'(\d{1,2})[:：](\d{2})',
    )
    DURATION_RE = re.compile(r'为期' + WS + r'(\d+)' + WS + r'个?小时')
    TITLE_TIME_RE = re.compile(r'(\d{1,2})月(\d{1,2})日' + WS + r'(\d{1,2})[:：](\d{2})')


    def __init__(self, instance_name=''):
        self.session = self._session_for()
        # Lock ownership token: the ALAS instance name
        self._lock_token = instance_name

    def _session_for(self):
        """Build the API session with a browser-like User-Agent; HTTPS goes through Happy Eyeballs."""
        s = requests.Session()
        s.headers.update({
            'User-Agent': (
                'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                '(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
            ),
            'Referer': f'https://space.bilibili.com/{self.DEFAULT_MID}/article',
        })
        s.mount('https://', HappyEyeballsHTTPAdapter())
        return s

    def _request_api(self, url, params) -> dict:
        """Request Bilibili API and return inner data field (code == 0)."""
        try:
            http_resp = self.session.get(url, params=params, timeout=self.REQUEST_TIMEOUT)
            http_resp.raise_for_status()
            body = http_resp.json()
        except requests.RequestException as e:
            # Network/HTTP failures are operational, degrade as FetchError
            raise FetchError(f'Request failed: {url}: {e}') from e
        except ValueError as e:
            raise FetchError(f'Invalid JSON response: {url}: {e}') from e
        if not isinstance(body, dict):
            raise FetchError(f'Unexpected JSON structure: {url}: {type(body).__name__}')
        if 'code' not in body:
            raise FetchError(f'Missing code field: {url}')
        if body['code'] != 0:
            raise FetchError(body.get('message') or body.get('msg') or 'unknown error')
        if 'data' not in body:
            raise FetchError(f'Missing data field: {url}')
        data = body['data']
        if not isinstance(data, dict):
            raise FetchError(f'Unexpected data structure: {url}: {type(data).__name__}')
        return data

    def find_collection(self) -> CollectionMeta:
        """
        Find target collection by name from UP's article lists.

        Returns:
            CollectionMeta: Metadata of the matched collection.
        """
        data = self._request_api(self.ARTICLE_LISTS_API, {'mid': str(self.DEFAULT_MID), 'sort': 0})
        lists = data.get('lists')
        if not isinstance(lists, list):
            raise FetchError(f'Unexpected lists structure: {self.ARTICLE_LISTS_API}')
        for item in lists:
            if not isinstance(item, dict):
                raise FetchError(f'Unexpected collection structure: {type(item).__name__}')
            if item.get('name') != self.DEFAULT_COLLECTION_NAME:
                continue
            try:
                collection_meta = CollectionMeta(
                    collection_id=int(item['id']),
                    collection_name=item['name'],
                    update_time=int(item['update_time']),
                    article_count=int(item['articles_count']),
                )
            except (KeyError, TypeError, ValueError) as e:
                raise FetchError(f'Unexpected collection fields: {e}') from e
            logger.info(
                f'Found collection: id={collection_meta.collection_id} '
                f'update_time={collection_meta.update_time} '
                f'article_count={collection_meta.article_count}'
            )
            return collection_meta
        raise FetchError(
            f'collection not found: mid={self.DEFAULT_MID} name={self.DEFAULT_COLLECTION_NAME!r}'
        )

    @staticmethod
    def _clean_text(text: str):
        return text.replace('[图片]', '').replace('\u3000', ' ').strip()

    @staticmethod
    def _format_window(start: datetime, end: datetime) -> str:
        """Format downtime window as 'YYYY-MM-DD HH:MM~HH:MM'."""
        return f'{start:%Y-%m-%d} {start:%H:%M}~{end:%H:%M}'

    def _infer_year(self, published, month, day):
        year = published.year
        delta = (datetime(year, month, day, tzinfo=self.CN_TZ).date() - published.date()).days
        if delta < -30:
            year += 1
        return year

    def parse_window_from_article(self, article: Article) -> Optional[DowntimeWindow]:
        """
        Returns:
            Optional[DowntimeWindow]: Parsed downtime window, or None if no pattern matched.
            The window string uses '~' as time range separator to prevent config loader
            from mis-parsing 'HH:MM-HH:MM' as an ISO 8601 datetime with offset.
        """
        if not isinstance(article.summary, str):
            raise FetchError(f'Unexpected article summary: {article.summary}')
        published = datetime.fromtimestamp(article.pub_ts, tz=self.CN_TZ)
        summary = self._clean_text(article.summary)

        window_match = self.WINDOW_RE.search(summary)
        if window_match:
            month = int(window_match.group(1))
            day = int(window_match.group(2))
            start_hour = int(window_match.group(3))
            start_minute = int(window_match.group(4))
            end_hour = int(window_match.group(5))
            end_minute = int(window_match.group(6))
            year = self._infer_year(published, month, day)
            start = datetime(year, month, day, start_hour, start_minute, tzinfo=self.CN_TZ)
            end = datetime(year, month, day, end_hour, end_minute, tzinfo=self.CN_TZ)
            return DowntimeWindow(window=self._format_window(start, end))

        duration_match = self.DURATION_RE.search(summary)
        if duration_match:
            if not isinstance(article.title, str):
                raise FetchError(f'Unexpected article title: {article.title}')
            title_match = self.TITLE_TIME_RE.search(article.title)
            if title_match:
                month = int(title_match.group(1))
                day = int(title_match.group(2))
                start_hour = int(title_match.group(3))
                start_minute = int(title_match.group(4))
                hours = int(duration_match.group(1))
                year = self._infer_year(published, month, day)
                start = datetime(year, month, day, start_hour, start_minute, tzinfo=self.CN_TZ)
                end = start + timedelta(hours=hours)
                return DowntimeWindow(window=self._format_window(start, end))

        return None

    @staticmethod
    def _normalize_article(raw_dict) -> Article:
        """Map a raw Bilibili article dict onto the Article dataclass."""
        if not isinstance(raw_dict, dict):
            raise FetchError(f'Unexpected article structure: {type(raw_dict).__name__}')
        try:
            return Article(
                pub_ts=int(raw_dict['publish_time']),
                id=str(raw_dict['id']),
                title=raw_dict['title'],
                summary=raw_dict['summary'],
            )
        except (KeyError, TypeError, ValueError) as e:
            raise FetchError(f'Unexpected article fields: {e}') from e

    def fetch_articles(
            self,
            cache_update_time=None,
            cache_article_count=None,
        ) -> tuple[CollectionMeta, Optional[list[Article]]]:
        """
        find_collection -> incremental check -> fetch articles -> consistency check

        Returns:
            tuple[CollectionMeta, Optional[list[Article]]]:
                Collection metadata and article list fetched from collection API.
                Article list is None when collection not updated (incremental check hit)
                or consistency check failed (article API lagged, returning stale list).
        """
        collection_meta = self.find_collection()

        # Incremental check: reuse the cache when both the collection
        # update_time and the declared article count are unchanged.
        if (
            cache_update_time is not None and cache_article_count is not None
            and cache_update_time == collection_meta.update_time
            and cache_article_count == collection_meta.article_count
        ):
            logger.info(
                f'Collection not updated, reuse local cache: '
                f'update_time={collection_meta.update_time} '
                f'article_count={collection_meta.article_count}'
            )
            return collection_meta, None

        logger.info('Collection updated, fetching articles')
        response = self._request_api(
            self.ARTICLE_COLLECTION_API, {'id': str(collection_meta.collection_id)},
        )
        articles = response.get('articles')
        if not isinstance(articles, list):
            raise FetchError(f'Unexpected articles structure: {self.ARTICLE_COLLECTION_API}')
        articles_list = [self._normalize_article(raw_dict) for raw_dict in articles]
        logger.info(f'Fetched {len(articles_list)} articles')

        if not articles_list:
            logger.warning('Article API returned an empty list, skip this fetch')
            return collection_meta, None

        # Consistency check: article count must match collection's articles_count.
        # Article API may lag behind collection API when publish time is close to
        # fetch time, returning stale list with fewer articles than declared.
        # Skip the fetch so a stale list never lands in the cache.
        if len(articles_list) != collection_meta.article_count:
            logger.warning(
                f'Article API lagged behind collection API: '
                f'fetched={len(articles_list)} expected={collection_meta.article_count}, '
                f'skip this fetch'
            )
            return collection_meta, None

        return collection_meta, articles_list

    @staticmethod
    def _latest_article(articles_list: list[Article]) -> Article:
        return max(articles_list, key=lambda a: a.pub_ts)

    def collect_info(
            self,
            cache_update_time=None,
            cache_article_count=None,
            cache_latest_pub_ts=None,
            cache_latest_article_id=None,
        ) -> tuple[CollectionMeta, Optional[Article], Optional[DowntimeWindow]]:
        """
        Full fetch and parse pipeline.

        Returns:
            tuple[CollectionMeta, Optional[Article], Optional[DowntimeWindow]]:
                Collection metadata, latest article, and parsed downtime window.
                Downtime window is None if parse failed.
                Latest article is None if collection not updated or consistency check failed.
        """
        collection_meta, articles_list = self.fetch_articles(cache_update_time, cache_article_count)
        if articles_list is None:
            return collection_meta, None, None
        latest_article = self._latest_article(articles_list)

        # Secondary consistency check:
        # verify the fetched latest article differs from the cache to confirm a real update.
        if (
            cache_latest_pub_ts is not None and cache_latest_article_id is not None
            and latest_article.pub_ts == cache_latest_pub_ts
            and latest_article.id == cache_latest_article_id
        ):
            logger.warning(
                f'Latest article unchanged despite collection update: '
                f'pub_ts={latest_article.pub_ts} id={latest_article.id}, '
                f'skip this fetch'
            )
            return collection_meta, None, None

        downtime_window = self.parse_window_from_article(latest_article)
        logger.info(
            f'Fetched downtime: window={downtime_window.window if downtime_window else "<none>"} '
            f'title={latest_article.title!r}'
        )

        return collection_meta, latest_article, downtime_window

    @classmethod
    def load_cache(cls) -> Optional[dict]:
        try:
            text = atomic_read_text(cls.CACHE_FILE)
        except FileNotFoundError:
            return None
        if not text:
            return None
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            logger.warning('Failed to parse downtime_cache.json, will re-fetch')
            return None

    @classmethod
    def save_cache(cls, payload):
        Path(cls.CACHE_FILE).parent.mkdir(parents=True, exist_ok=True)
        atomic_write(cls.CACHE_FILE, json.dumps(payload, ensure_ascii=False, indent=2))
        logger.info(f'Saved downtime cache to {cls.CACHE_FILE}')

    @staticmethod
    def _file_age(path: Path) -> float:
        return time.time() - path.stat().st_mtime

    def get_downtime_info(self, force_refresh=False) -> dict:
        """
        Fetch from API, parse, and save cache.

        Args:
            force_refresh (bool): Skip incremental check, always fetch from API and overwrite cache.

        Returns:
            dict: payload with window, title, collection metadata, and latest article info.
        """
        cache = self.load_cache()
        if force_refresh or cache is None:
            cache_update_time = cache_article_count = cache_latest_pub_ts = cache_latest_article_id = None
        else:
            cache_update_time = cache.get('update_time')
            cache_article_count = cache.get('article_count')
            cache_latest_pub_ts = cache.get('latest_pub_ts')
            cache_latest_article_id = cache.get('latest_article_id')

        collection_meta, latest_article, downtime_window = self.collect_info(
            cache_update_time,
            cache_article_count,
            cache_latest_pub_ts,
            cache_latest_article_id,
        )

        if latest_article is None:
            if cache is None:
                raise FetchError(
                    'No available cache and collection not updated or consistency check failed'
                )
            return cache

        # Degrade on parse failure: reuse cached window, or raise if no cache
        if downtime_window is None:
            if cache and cache.get('window'):
                logger.warning(f'Failed to parse, reuse cached window: {cache["window"]}')
                window = cache['window']
            else:
                raise FetchError('Failed to parse and no cache available')
        else:
            window = downtime_window.window

        cache_payload = {
            'collection_id': collection_meta.collection_id,
            'update_time': collection_meta.update_time,
            'article_count': collection_meta.article_count,
            'latest_pub_ts': latest_article.pub_ts,
            'latest_article_id': latest_article.id,
            'window': window,
            'title': latest_article.title,
        }
        self.save_cache(cache_payload)
        return cache_payload

    def acquire_lock(self) -> bool:
        """
        Acquire cross-process fetch lock.

        The lock file carries the owning instance name; release_lock() deletes
        it only while the read-back token still matches. Read and unlink are
        not atomic; a takeover in between degrades to a duplicate fetch.

        Returns:
            bool: True if acquired, False on timeout or OSError.
        """
        p = Path(self.FETCH_LOCK_FILE)
        p.parent.mkdir(parents=True, exist_ok=True)
        deadline = time.monotonic() + self.LOCK_WAIT_SECONDS
        while time.monotonic() < deadline:
            try:
                fd = os.open(str(p), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            except FileExistsError:
                # Lock held: check staleness only on conflict, take over when stale.
                # os.replace moves the stale lock aside atomically.
                try:
                    age = self._file_age(p)
                    if age < self.LOCK_WAIT_SECONDS:
                        time.sleep(1)
                        continue
                    logger.warning(f'Stale lock assumed (age={age:.1f}s), overriding')
                    stale = p.with_name(f'{p.name}.{self._lock_token}.stale')
                    try:
                        os.replace(str(p), str(stale))
                    except FileNotFoundError:
                        pass  # already moved by another instance
                    try:
                        stale.unlink()
                    except OSError:
                        pass
                except (OSError, FileNotFoundError):
                    pass
                continue
            except OSError as e:
                logger.warning(f'Failed to acquire fetch lock: {e}')
                return False
            write_failed = False
            try:
                # os.write() may write partially; write the whole token.
                data = self._lock_token.encode()
                while data:
                    written = os.write(fd, data)
                    data = data[written:]
            except OSError as e:
                logger.warning(f'Failed to write fetch lock: {e}')
                write_failed = True
            finally:
                os.close(fd)
            if write_failed:
                # Remove the incomplete lock after close; Windows cannot
                # unlink an open file.
                try:
                    os.unlink(str(p))
                except OSError:
                    pass
                return False
            return True
        return False

    def release_lock(self):
        p = Path(self.FETCH_LOCK_FILE)
        try:
            if atomic_read_text(p) == self._lock_token:
                p.unlink()
            else:
                # Taken over by another instance (stale-lock handover); leave it.
                logger.info('Fetch lock taken over by another instance, keep it')
        except FileNotFoundError:
            pass  # lock already gone
        except OSError as e:
            logger.warning(f'Failed to release fetch lock: {e}')

    def _try_reuse_fresh_cache(self, reason: str) -> Optional[dict]:
        """
        Load cache and check if fresh.

        Returns:
            Optional[dict]: Fresh cache payload, or None when missing/stale.
        """
        cache = self.load_cache()
        if cache is None:
            return None
        age = self._file_age(Path(self.CACHE_FILE))
        if age >= self.CACHE_FRESH_SECONDS:
            return None
        logger.info(
            f'{reason}: '
            f'cache_age={age:.1f}s window={cache.get("window") or "<none>"}'
        )
        return cache

    def run(self, force_refresh=False) -> dict:
        """
        Lock-guarded entry for the downtime pipeline.

        Fast path: fresh cache, return directly without lock (read-only).
        Common path: acquire cross-process lock, re-check cache in case
        another instance fetched while waiting, then fetch.
        Fallback: lock acquisition failed, reuse cache only if another process
        has freshly updated it; otherwise raise FetchError.

        Returns:
            dict: payload with window, title, collection metadata, and latest article info.
        """
        # Fast path: fresh cache, read-only, no lock needed
        if not force_refresh:
            cache = self._try_reuse_fresh_cache('Cache is fresh, reuse without fetch')
            if cache is not None:
                return cache
        if not self.acquire_lock():
            # Lock failed: reuse fresh cache if another process updated it
            cache = self._try_reuse_fresh_cache('Cache updated by another process, reuse')
            if cache is not None:
                return cache
            raise FetchError('Failed to acquire fetch lock')
        try:
            # Re-check cache after acquire_lock: another instance may have finished fetching.
            # Trades a potentially redundant cache check to prevent duplicate API calls.
            if not force_refresh:
                cache = self._try_reuse_fresh_cache('Cache updated by another process, reuse')
                if cache is not None:
                    return cache
            return self.get_downtime_info(force_refresh=force_refresh)
        finally:
            self.release_lock()