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 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)) REQUEST_TIMEOUT = 30.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 = 600 # A bit larger than REQUEST_TIMEOUT LOCK_WAIT_SECONDS = 35 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.""" 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', }) 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 ""} ' 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 ""}' ) 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()