Feat: add module smart_mgmt

module/smart_mgmt/downtime_fetch.py

@@ -0,0 +1,47 @@
+import random
+from datetime import timedelta
+
+from module.config.utils import get_server_next_update
+from module.logger import logger
+from module.smart_mgmt.downtime_fetcher import DowntimeFetcher, FetchError
+
+
+class DowntimeFetch:
+    def __init__(self, config):
+        self.config = config
+
+    def _set_downtime(self, window, title):
+        self.config.modified['DowntimeMgmt.Downtime.Window'] = window
+        self.config.modified['DowntimeMgmt.Downtime.Title'] = title
+        self.config.update()
+
+    def _update_downtime(self) -> bool:
+        """Fetch downtime via fetcher and write cross-task config."""
+        try:
+            payload = DowntimeFetcher().run()
+        except FetchError as e:
+            # Expected operational failure (API error, collection missing, no cache);
+            # degrade gracefully and let the caller reschedule
+            logger.warning(f'Fetch downtime failed: {e}')
+            return False
+        except Exception:
+            # Unexpected programming error; full traceback for diagnosis
+            logger.exception('Unexpected error during downtime fetch')
+            return False
+        self._set_downtime(payload.get('window', ''), payload.get('title', ''))
+        return True
+
+    def run(self):
+        success = False
+        if self.config.SERVER != 'cn':
+            # Downtime fetch targets CN Bilibili notices only; skip on other servers
+            logger.warning('Downtime fetch is only available for CN server, skip')
+        else:
+            success = self._update_downtime()
+        if success:
+            # Stagger execution across instances within 5 minutes after server update
+            next_update = get_server_next_update(self.config.Scheduler_ServerUpdate)
+            target = next_update + timedelta(seconds=random.randint(0, 300))
+            self.config.task_delay(target=target)
+        else:
+            self.config.task_delay(success=False)

module/smart_mgmt/downtime_fetcher.py

@@ -0,0 +1,345 @@
+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
+
+@dataclass
+class Article:
+    pub_ts: int
+    title: str
+    summary: str
+
+@dataclass
+class DowntimeWindow:
+    window: str
+
+
+class DowntimeFetcher:
+    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
+
+    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})')
+
+    CACHE_FILE = './log/downtime_cache.json'
+    FETCH_LOCK_FILE = './log/downtime_fetch.lock'
+    # Multi-instance duplicate fetch avoidance window (seconds)
+    CACHE_FRESH_SECONDS = 300
+    # A bit larger than REQUEST_TIMEOUT
+    LOCK_WAIT_SECONDS = 35
+
+    def __init__(self):
+        self.session = self._session_for()
+
+    def _session_for(self):
+        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)."""
+        http_resp = self.session.get(url, params=params, timeout=self.REQUEST_TIMEOUT)
+        http_resp.raise_for_status()
+        body = http_resp.json()
+        if body['code'] != 0:
+            raise FetchError(body.get('message') or body.get('msg') or 'unknown error')
+        return body['data']
+
+    def find_collection(self) -> CollectionMeta:
+        """
+        Find target collection by name from UP's article lists.
+
+        Returns:
+            CollectionMeta: Metadata of the matched collection.
+        """
+        lists = self._request_api(self.ARTICLE_LISTS_API, {'mid': str(self.DEFAULT_MID), 'sort': 0})['lists']
+        for item in lists:
+            if item['name'] == self.DEFAULT_COLLECTION_NAME:
+                collection_meta = CollectionMeta(
+                    collection_id=int(item['id']),
+                    collection_name=item['name'],
+                    update_time=int(item['update_time']),
+                )
+                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'],
+        )
+
+    @staticmethod
+    def _clean_text(text: str):
+        return text.replace('[图片]', '').replace('\u3000', ' ').strip()
+
+    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 _build_window_dt(self, published, sm, sd, sh, smin, eh, emin):
+        year = self._infer_year(published, sm, sd)
+        start = datetime(year, sm, sd, sh, smin, tzinfo=self.CN_TZ)
+        end = datetime(year, sm, sd, eh, emin, tzinfo=self.CN_TZ)
+        return start, end
+
+    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.
+        """
+        published = datetime.fromtimestamp(article.pub_ts, tz=self.CN_TZ)
+        body = self._clean_text(article.summary)
+        window_match = self.WINDOW_RE.search(body)
+        duration_match = self.DURATION_RE.search(body)
+
+        if window_match:
+            sm = int(window_match.group(1))
+            sd = 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, sm, sd, sh, smin, eh, emin)
+            return DowntimeWindow(window=f'{start:%Y-%m-%d} {start:%H:%M}~{end:%H:%M}')
+
+        if duration_match:
+            title_match = self.TITLE_TIME_RE.search(article.title)
+            if title_match:
+                sm = int(title_match.group(1))
+                sd = int(title_match.group(2))
+                sh = int(title_match.group(3))
+                smin = int(title_match.group(4))
+                hours = int(duration_match.group(1))
+                start, _ = self._build_window_dt(published, sm, sd, sh, smin, sh, smin)
+                end = start + timedelta(hours=hours)
+                return DowntimeWindow(window=f'{start:%Y-%m-%d} {start:%H:%M}~{end:%H:%M}')
+
+        return None
+
+    @staticmethod
+    def _latest_article(articles: list[Article]) -> Article:
+        return max(articles, key=lambda a: a.pub_ts)
+
+    def fetch_articles(self, local_update_time=None) -> tuple[Optional[list[Article]], CollectionMeta]:
+        """
+        Full fetch pipeline: find_collection → incremental check → fetch articles.
+
+        Returns:
+            tuple[Optional[list[Article]], CollectionMeta]:
+                articles is None when collection not updated (incremental check hit).
+        """
+        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
+
+        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
+
+    def collect_info(self, local_update_time=None) -> tuple[Optional[DowntimeWindow], CollectionMeta, Optional[Article]]:
+        """
+        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.
+        """
+        articles, collection_meta = self.fetch_articles(local_update_time)
+        if articles is None:
+            return None, collection_meta, None
+        latest_article = self._latest_article(articles)
+        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
+
+    @classmethod
+    def load_cache(cls) -> Optional[dict]:
+        if not Path(cls.CACHE_FILE).exists():
+            return None
+        text = atomic_read_text(cls.CACHE_FILE)
+        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}')
+
+    @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
+
+    def get_downtime_info(self) -> dict:
+        """
+        Full pipeline: load cache → collect_info → save cache.
+
+        Returns:
+            dict: payload with window, title, collection_id, update_time.
+        """
+        cache = self.load_cache()
+        # Avoid duplicate fetch across instances: reuse fresh cache
+        if cache and self.is_cache_fresh():
+            logger.info('Cache is fresh, reuse without fetch')
+            return cache
+        local_update_time = cache.get('update_time') if cache else None
+
+        downtime_window, collection_meta, latest_article = self.collect_info(local_update_time)
+
+        if latest_article is None:
+            if cache is None:
+                raise FetchError('No available cache and collection not updated')
+            return cache
+
+        # Degrade on parse failure: reuse cached window
+        if downtime_window is None:
+            if cache and cache.get('window'):
+                logger.warning('Parse failed, reuse cached window')
+                window = cache['window']
+            else:
+                window = ''
+        else:
+            window = downtime_window.window
+
+        cache_payload = {
+            'collection_id': collection_meta.collection_id,
+            'update_time': collection_meta.update_time,
+            'window': window,
+            'title': latest_article.title,
+        }
+        self.save_cache(cache_payload)
+        return cache_payload
+
+    @classmethod
+    def acquire_lock(cls) -> bool:
+        """
+        Acquire cross-process fetch lock.
+
+        Returns:
+            bool: True if acquired, False on timeout or OSError.
+        """
+        p = Path(cls.FETCH_LOCK_FILE)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        deadline = time.monotonic() + cls.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)
+            except OSError as e:
+                logger.warning(f'Failed to acquire fetch lock: {e}')
+                return False
+        return False
+
+    @classmethod
+    def release_lock(cls):
+        try:
+            Path(cls.FETCH_LOCK_FILE).unlink()
+        except FileNotFoundError:
+            pass
+        except OSError as e:
+            logger.warning(f'Failed to release fetch lock: {e}')
+
+    def run(self) -> dict:
+        """
+        Lock-guarded entry for the downtime pipeline.
+
+        Fast path: cache fresh, invoke get_downtime_info without lock (read-only, no contention).
+        Common path: acquire cross-process lock, then get_downtime_info (re-checks freshness).
+        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.
+        """
+        if self.is_cache_fresh():
+            return self.get_downtime_info()
+        if not self.acquire_lock():
+            if self.is_cache_fresh():
+                logger.info('Cache updated by another process, reuse')
+                return self.get_downtime_info()
+            raise FetchError('Failed to acquire fetch lock')
+        try:
+            return self.get_downtime_info()
+        finally:
+            self.release_lock()

module/smart_mgmt/utils.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from typing import Optional
+
+from module.logger import logger
+
+
+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(downtime: str) -> Optional[datetime]:
+    """
+    Parse downtime window string into end datetime.
+
+    Args:
+        downtime (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(downtime)
+    if not match:
+        logger.warning(f'Failed to parse downtime window: {downtime}')
+        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