feat: add proactive login expiry monitoring

Add login_reminder module to check credential expiration every 6 hours and send webhook alerts (24h/6h before expiry). Fix expire_time from 30 days to 4 days to align with WeChat actual validity.

Made-with: Cursor

README.md

@@ -27,7 +27,7 @@
 - **公众号搜索** — 按名称搜索公众号，获取 FakeID
 - **扫码登录** — 微信公众平台扫码登录，凭证自动保存，4 天有效期
 - **图片代理** — 代理微信 CDN 图片，解决防盗链问题
-- **Webhook 通知** — 登录过期、触发验证等事件自动推送（支持企业微信机器人）
+- **Webhook 通知** — 登录过期提醒（提前24h/6h预警+已过期通知）、触发验证等事件自动推送（支持企业微信机器人）
 - **API 文档** — 自动生成 Swagger UI / ReDoc，在线调试所有接口
 
 <div align="center">
@@ -361,7 +361,9 @@ cp env.example .env
 | `WECHAT_TOKEN` | 微信 Token（登录后自动填充） | - |
 | `WECHAT_COOKIE` | 微信 Cookie（登录后自动填充） | - |
 | `WECHAT_FAKEID` | 公众号 FakeID（登录后自动填充） | - |
-| `WEBHOOK_URL` | Webhook 通知地址（可选） | 空 |
+| `WECHAT_EXPIRE_TIME` | 凭证过期时间（登录后自动填充） | - |
+| `WEBHOOK_URL` | Webhook 通知地址（支持企业微信机器人） | 空 |
+| `WEBHOOK_NOTIFICATION_INTERVAL` | 同一事件通知最小间隔（秒） | 300 |
 | `RATE_LIMIT_GLOBAL` | 全局每分钟请求上限 | 10 |
 | `RATE_LIMIT_PER_IP` | 单 IP 每分钟请求上限 | 5 |
 | `RATE_LIMIT_ARTICLE_INTERVAL` | 文章请求最小间隔（秒） | 3 |
@@ -493,6 +495,7 @@ PROXY_URLS=socks5://myuser:mypass@vps1-ip:1080,socks5://myuser:mypass@vps2-ip:10
 │   ├── rate_limiter.py   # 限频器
 │   ├── rss_store.py      # RSS 数据存储（SQLite）
 │   ├── rss_poller.py     # RSS 后台轮询器
+│   ├── login_reminder.py # 登录过期提醒（主动检测）
 │   ├── content_processor.py  # 内容处理与图片代理
 │   ├── image_proxy.py    # 图片URL代理工具
 │   ├── article_fetcher.py    # 批量并发获取文章
@@ -540,9 +543,14 @@ PROXY_URLS=socks5://myuser:mypass@vps1-ip:1080,socks5://myuser:mypass@vps2-ip:10
 </details>
 
 <details>
-<summary><b>Token 多久过期</b></summary>
+<summary><b>Token 多久过期？如何提前知道？</b></summary>
 
-Cookie 登录有效期约 4 天，过期后需重新扫码登录。配置 `WEBHOOK_URL` 可以在过期时收到通知。
+Cookie 登录有效期约 4 天，系统会：
+1. 前端显示到期时间（`/api/admin/status` 接口返回 `expireTime` 和 `isExpired` 字段）
+2. **后台每 6 小时主动检测**，提前 24h / 6h 通过 Webhook 预警
+3. 过期后立即通过 Webhook 通知
+
+配置 `WEBHOOK_URL`（支持企业微信群机器人）可收到实时提醒，避免因凭证过期导致 RSS 轮询失败或搜索功能不可用。
 </details>
 
 <details>

app.py

@@ -59,7 +59,14 @@ async def lifespan(app: FastAPI):
 
     init_db()
     await rss_poller.start()
+    
+    # 启动登录过期提醒器（自动检测凭证有效期并 webhook 通知）
+    from utils.login_reminder import login_reminder
+    await login_reminder.start()
+    
     yield
+    
+    await login_reminder.stop()
     await rss_poller.stop()
 
 

routes/login.py

@@ -497,8 +497,8 @@ async def biz_login(request: Request):
                 import traceback
                 traceback.print_exc()
         
-        # 计算过期时间（30天后）
+        # 计算过期时间（4天后，与微信实际有效期一致）
-        expire_time = int((time.time() + 30 * 24 * 3600) * 1000)
+        expire_time = int((time.time() + 4 * 24 * 3600) * 1000)
         
         # 保存凭证
         auth_manager.save_credentials(

utils/login_reminder.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# Copyright (C) 2026 tmwgsicp
+# Licensed under the GNU Affero General Public License v3.0
+# See LICENSE file in the project root for full license text.
+# SPDX-License-Identifier: AGPL-3.0-only
+"""
+登录过期提醒（开源版）
+定期检查本地微信登录凭证过期状态，提前 webhook 通知。
+"""
+
+import asyncio
+import logging
+import time
+from typing import Optional
+from utils.webhook import webhook
+
+logger = logging.getLogger(__name__)
+
+
+class LoginReminder:
+    """登录过期提醒管理器（开源版单账号架构）"""
+    
+    def __init__(self):
+        self.check_interval = 6 * 3600  # 每 6 小时检查一次
+        self.warning_threshold = 24 * 3600  # 提前 24 小时预警
+        self.critical_threshold = 6 * 3600  # 提前 6 小时严重警告
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+        self._last_warning_level = None  # 记录最后一次警告级别，避免重复
+
+    async def start(self):
+        """启动提醒服务"""
+        if self._running:
+            logger.warning("登录提醒服务已在运行")
+            return
+        
+        self._running = True
+        self._task = asyncio.create_task(self._run())
+        logger.info("登录提醒服务已启动，检查间隔: %d 秒", self.check_interval)
+
+    async def stop(self):
+        """停止提醒服务"""
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        logger.info("登录提醒服务已停止")
+
+    async def _run(self):
+        """后台任务循环"""
+        while self._running:
+            try:
+                await self._check_login_status()
+            except Exception as e:
+                logger.error("检查登录状态失败: %s", e, exc_info=True)
+            
+            await asyncio.sleep(self.check_interval)
+
+    async def _check_login_status(self):
+        """检查本地登录凭证的过期状态"""
+        from utils.auth_manager import auth_manager
+        
+        # 获取凭证信息
+        creds = auth_manager.get_credentials()
+        if not creds or not creds.get("token"):
+            logger.debug("无登录凭证，跳过检查")
+            return
+        
+        expire_time = creds.get("expire_time", 0)
+        if expire_time <= 0:
+            logger.debug("凭证无过期时间，跳过检查")
+            return
+        
+        nickname = creds.get("nickname", "未知账号")
+        now = int(time.time() * 1000)  # 毫秒时间戳
+        time_left_ms = expire_time - now
+        time_left_sec = time_left_ms / 1000
+        
+        # 已过期
+        if time_left_sec <= 0:
+            if self._last_warning_level != 'expired':
+                await self._notify_expired(nickname)
+                self._last_warning_level = 'expired'
+            return
+        
+        # 严重警告（6 小时内过期）
+        if time_left_sec <= self.critical_threshold:
+            if self._last_warning_level not in ['critical', 'expired']:
+                await self._notify_critical(nickname, time_left_sec)
+                self._last_warning_level = 'critical'
+            return
+        
+        # 一般警告（24 小时内过期）
+        if time_left_sec <= self.warning_threshold:
+            if self._last_warning_level not in ['warning', 'critical', 'expired']:
+                await self._notify_warning(nickname, time_left_sec)
+                self._last_warning_level = 'warning'
+            return
+        
+        # 状态正常，重置警告级别
+        if self._last_warning_level is not None:
+            self._last_warning_level = None
+            logger.info("登录状态已恢复正常: %s", nickname)
+
+    async def _notify_warning(self, nickname: str, time_left: float):
+        """发送一般警告通知"""
+        hours = time_left / 3600
+        logger.warning(
+            "登录凭证即将过期 [%s] - 剩余 %.1f 小时",
+            nickname, hours
+        )
+        
+        await webhook.notify('login_expiring_soon', {
+            'nickname': nickname,
+            'hours_left': round(hours, 1),
+            'level': 'warning',
+            'message': f'登录凭证将在 {round(hours, 1)} 小时后过期，请及时重新登录',
+        })
+
+    async def _notify_critical(self, nickname: str, time_left: float):
+        """发送严重警告通知"""
+        hours = time_left / 3600
+        logger.error(
+            "登录凭证即将过期（紧急）[%s] - 剩余 %.1f 小时",
+            nickname, hours
+        )
+        
+        await webhook.notify('login_expiring_critical', {
+            'nickname': nickname,
+            'hours_left': round(hours, 1),
+            'level': 'critical',
+            'message': f'登录凭证将在 {round(hours, 1)} 小时后过期（紧急），请立即重新登录',
+        })
+
+    async def _notify_expired(self, nickname: str):
+        """发送已过期通知"""
+        logger.error("登录凭证已过期 [%s]", nickname)
+        
+        await webhook.notify('login_expired', {
+            'nickname': nickname,
+            'message': '登录凭证已过期，API 功能将受限，请重新登录',
+        })
+
+
+# 全局单例
+login_reminder = LoginReminder()

utils/webhook.py

@@ -21,6 +21,8 @@ logger = logging.getLogger("webhook")
 EVENT_LABELS = {
     "login_success": "登录成功",
     "login_expired": "登录过期",
+    "login_expiring_soon": "登录即将过期",
+    "login_expiring_critical": "登录即将过期（紧急）",
     "verification_required": "触发验证",
     "content_fetch_failed": "文章内容获取失败",
 }