diff --git a/module/smart_mgmt/downtime_fetch.py b/module/smart_mgmt/downtime_fetch.py index 516d6525e..1dc354abb 100644 --- a/module/smart_mgmt/downtime_fetch.py +++ b/module/smart_mgmt/downtime_fetch.py @@ -22,8 +22,7 @@ def parse_downtime(window: str) -> Optional[datetime]: 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. + Optional[datetime]: End datetime parsed from the window, or None if no pattern matched. """ match = DOWNTIME_FORMAT_RE.search(window) if not match: @@ -38,8 +37,9 @@ def parse_downtime(window: str) -> Optional[datetime]: def update_downtime(config: AzurLaneConfig, force_refresh=False) -> bool: """Fetch downtime and write config.""" + logger.info('Fetch downtime') try: - payload = DowntimeFetcher().run(force_refresh=force_refresh) + payload = DowntimeFetcher(instance_name=config.config_name).run(force_refresh=force_refresh) except FetchError as e: # Expected operational failure degrade logger.warning(f'Failed to fetch downtime: {e}') @@ -75,6 +75,7 @@ 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') + logger.info(f'Downtime fetch start: server={config.SERVER} test_mode={config.DowntimeFetch_FetcherTest}') if config.DowntimeFetch_FetcherTest: DowntimeFetcherTest().run() else: diff --git a/module/smart_mgmt/downtime_fetcher.py b/module/smart_mgmt/downtime_fetcher.py index 6f8c3b99c..47a01d5fb 100644 --- a/module/smart_mgmt/downtime_fetcher.py +++ b/module/smart_mgmt/downtime_fetcher.py @@ -24,19 +24,24 @@ 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' @@ -44,6 +49,14 @@ class DowntimeFetcher: 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 + @@ -55,17 +68,14 @@ class DowntimeFetcher: 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})') - 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 - def __init__(self): + 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': ( @@ -78,12 +88,27 @@ class DowntimeFetcher: def _request_api(self, url, params) -> dict: """Request Bilibili API and return inner data field (code == 0).""" - http_resp = self.session.get(url, params=params, timeout=self.REQUEST_TIMEOUT) - http_resp.raise_for_status() - body = http_resp.json() + 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') - return body['data'] + 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: """ @@ -92,31 +117,43 @@ class DowntimeFetcher: Returns: CollectionMeta: Metadata of the matched collection. """ - lists = self._request_api(self.ARTICLE_LISTS_API, {'mid': str(self.DEFAULT_MID), 'sort': 0})['lists'] + 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 item['name'] == self.DEFAULT_COLLECTION_NAME: + 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']), ) - logger.info(f'Found collection: id={collection_meta.collection_id} ' - f'update_time={collection_meta.update_time}') - return collection_meta - raise FetchError(f'collection not found: mid={self.DEFAULT_MID} name={self.DEFAULT_COLLECTION_NAME!r}') - - @staticmethod - def _normalize_article(article) -> Article: - return Article( - pub_ts=int(article['publish_time']), - title=article['title'], - summary=article['summary'], + 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 @@ -124,12 +161,6 @@ class DowntimeFetcher: year += 1 return year - def _build_window_dt(self, published, month, day, sh, smin, eh, emin): - year = self._infer_year(published, month, day) - start = datetime(year, month, day, sh, smin, tzinfo=self.CN_TZ) - end = datetime(year, month, day, eh, emin, tzinfo=self.CN_TZ) - return start, end - def parse_window_from_article(self, article: Article) -> Optional[DowntimeWindow]: """ Returns: @@ -137,82 +168,168 @@ class DowntimeFetcher: 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) - duration_match = self.DURATION_RE.search(summary) + window_match = self.WINDOW_RE.search(summary) if window_match: month = int(window_match.group(1)) day = int(window_match.group(2)) - sh = int(window_match.group(3)) - smin = int(window_match.group(4)) - eh = int(window_match.group(5)) - emin = int(window_match.group(6)) - start, end = self._build_window_dt(published, month, day, sh, smin, eh, emin) - return DowntimeWindow(window=f'{start:%Y-%m-%d} {start:%H:%M}~{end:%H:%M}') + 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)) - sh = int(title_match.group(3)) - smin = int(title_match.group(4)) + start_hour = int(title_match.group(3)) + start_minute = int(title_match.group(4)) hours = int(duration_match.group(1)) - start, _ = self._build_window_dt(published, month, day, sh, smin, sh, smin) + 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=f'{start:%Y-%m-%d} {start:%H:%M}~{end:%H:%M}') + return DowntimeWindow(window=self._format_window(start, end)) return None @staticmethod - def _latest_article(articles: list[Article]) -> Article: - return max(articles, key=lambda a: a.pub_ts) + 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, local_update_time=None) -> tuple[Optional[list[Article]], CollectionMeta]: + def fetch_articles( + self, + cache_update_time=None, + cache_article_count=None, + ) -> tuple[CollectionMeta, Optional[list[Article]]]: """ - find_collection -> incremental check -> fetch articles + find_collection -> incremental check -> fetch articles -> consistency check Returns: - tuple[Optional[list[Article]], CollectionMeta]: - articles is None when collection not updated (incremental check hit). + 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() - if local_update_time is not None and local_update_time == collection_meta.update_time: - logger.info('Collection not updated, reuse local cache') - return None, collection_meta + # 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(f'Collection updated, fetching articles (id={collection_meta.collection_id})') - articles_data = self._request_api(self.ARTICLE_COLLECTION_API, {'id': str(collection_meta.collection_id)}) - articles = [self._normalize_article(article) for article in articles_data['articles']] - logger.info(f'Fetched {len(articles)} articles') - return articles, collection_meta + 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') - def collect_info(self, local_update_time=None) -> tuple[Optional[DowntimeWindow], CollectionMeta, Optional[Article]]: + 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[Optional[DowntimeWindow], CollectionMeta, Optional[Article]]: - (downtime_window, collection_meta, latest_article). - downtime_window is None if parse failed. latest_article is None if collection not updated. + 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. """ - articles, collection_meta = self.fetch_articles(local_update_time) - if articles is None: - return None, collection_meta, None - latest_article = self._latest_article(articles) + 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 downtime_window, collection_meta, 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]: - if not Path(cls.CACHE_FILE).exists(): + try: + text = atomic_read_text(cls.CACHE_FILE) + except FileNotFoundError: return None - text = atomic_read_text(cls.CACHE_FILE) if not text: return None try: @@ -227,40 +344,47 @@ class DowntimeFetcher: atomic_write(cls.CACHE_FILE, json.dumps(payload, ensure_ascii=False, indent=2)) logger.info(f'Saved downtime cache to {cls.CACHE_FILE}') - @classmethod - def is_cache_fresh(cls): - """Check if cache file was modified within CACHE_FRESH_SECONDS.""" - p = Path(cls.CACHE_FILE) - if not p.exists(): - return False - age = datetime.now().timestamp() - p.stat().st_mtime - return age < cls.CACHE_FRESH_SECONDS + @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. + force_refresh (bool): Skip incremental check, always fetch from API and overwrite cache. Returns: - dict: payload with window, title, collection_id, update_time. + dict: payload with window, title, collection metadata, and latest article info. """ cache = self.load_cache() - local_update_time = None if force_refresh else (cache.get('update_time') if cache else None) + 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') - downtime_window, collection_meta, latest_article = self.collect_info(local_update_time) + 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') + 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('Failed to parse, reuse cached 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') @@ -270,82 +394,142 @@ class DowntimeFetcher: 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 - @classmethod - def acquire_lock(cls) -> bool: + 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(cls.FETCH_LOCK_FILE) + p = Path(self.FETCH_LOCK_FILE) p.parent.mkdir(parents=True, exist_ok=True) - deadline = time.monotonic() + cls.LOCK_WAIT_SECONDS + deadline = time.monotonic() + self.LOCK_WAIT_SECONDS while time.monotonic() < deadline: - if p.exists(): - try: - age = datetime.now().timestamp() - p.stat().st_mtime - if age < cls.LOCK_WAIT_SECONDS: - time.sleep(1) - continue - logger.warning(f'Stale lock assumed (age={age:.0f}s), overriding') - p.unlink() - except (OSError, FileNotFoundError): - pass - # Atomic create via O_CREAT|O_EXCL to avoid race between instances try: fd = os.open(str(p), os.O_CREAT | os.O_EXCL | os.O_WRONLY) - os.write(fd, str(datetime.now().timestamp()).encode()) - os.close(fd) - return True except FileExistsError: - time.sleep(1) + # 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 - @classmethod - def release_lock(cls): + def release_lock(self): + p = Path(self.FETCH_LOCK_FILE) try: - Path(cls.FETCH_LOCK_FILE).unlink() + 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 + 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, then fetch. + 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_id, update_time. + 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.load_cache() - if cache and self.is_cache_fresh(): - logger.info('Cache is fresh, reuse without fetch') + 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.load_cache() - if cache and self.is_cache_fresh(): - logger.info('Cache updated by another process, reuse') + 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() diff --git a/module/submodule/utils.py b/module/submodule/utils.py index 86330ba2b..d1c40e3a7 100644 --- a/module/submodule/utils.py +++ b/module/submodule/utils.py @@ -22,8 +22,8 @@ def get_available_func(): 'IslandPearl', 'AzurLaneUncensored', 'Benchmark', - 'GameManager', 'DowntimeFetch', + 'GameManager', ) def get_available_mod():