""" 后台定时任务管理 """ import asyncio from typing import List, Optional from contextlib import asynccontextmanager from app.database import get_db from app.core.logger import logger from app.services.key_rotation_service import KeyRotationService from app.services.session_service import SessionService from app.services.heartbeat_monitor import HeartbeatMonitor from app.models.system_config import SystemConfig from app.config import settings class BackgroundTaskManager: """后台任务管理器""" def __init__(self): self.tasks: List[asyncio.Task] = [] self.running = False async def start_all_tasks(self): """启动所有后台任务""" if self.running: logger.warning("后台任务已在运行") return self.running = True logger.info("启动后台定时任务...") # 启动各个任务 self.tasks = [ asyncio.create_task(self._key_rotation_task()), asyncio.create_task(self._session_cleanup_task()), asyncio.create_task(self._heartbeat_cleanup_task()), ] logger.info(f"已启动 {len(self.tasks)} 个后台任务") async def stop_all_tasks(self): """停止所有后台任务""" if not self.running: logger.warning("后台任务未运行") return logger.info("停止后台定时任务...") self.running = False # 取消所有任务 for task in self.tasks: if not task.done(): task.cancel() # 等待任务完成 try: await asyncio.gather(*self.tasks, return_exceptions=True) except asyncio.CancelledError: pass self.tasks.clear() logger.info("后台定时任务已停止") async def _key_rotation_task(self): """密钥轮换任务 - 根据配置间隔执行""" # 获取检查间隔配置 check_interval_minutes = SystemConfig.get_config_value('key_rotation_check_interval_minutes', 5) check_interval_seconds = check_interval_minutes * 60 logger.info(f"启动密钥轮换定时任务 (间隔: {check_interval_minutes}分钟)") while self.running: try: logger.debug("执行密钥轮换检查...") db = next(get_db()) try: # 检查是否需要轮换密钥 if KeyRotationService.should_rotate_key(db): logger.info("检测到需要轮换密钥,开始执行...") success, public_key, error_msg = KeyRotationService.rotate_key(db) if success: logger.info("密钥轮换成功") else: logger.error(f"密钥轮换失败: {error_msg}") else: logger.debug("当前无需轮换密钥") # 清理过期密钥 cleaned_count = KeyRotationService.cleanup_expired_keys(db) if cleaned_count > 0: logger.info(f"清理了 {cleaned_count} 个过期密钥") finally: db.close() except Exception as e: logger.exception("密钥轮换任务异常") # 根据配置等待指定时间 await asyncio.sleep(check_interval_seconds) async def _session_cleanup_task(self): """Session清理任务 - 根据配置间隔执行""" # 获取清理间隔配置 cleanup_interval_hours = SystemConfig.get_config_value('session_cleanup_interval_hours', 2) cleanup_interval_seconds = cleanup_interval_hours * 3600 logger.info(f"启动Session清理定时任务 (间隔: {cleanup_interval_hours}小时)") while self.running: try: logger.debug("执行Session清理...") db = next(get_db()) try: cleaned_count = SessionService.cleanup_expired_sessions(db) if cleaned_count > 0: logger.info(f"清理了 {cleaned_count} 个30天前的非活跃Session记录") else: logger.debug("没有需要清理的旧Session记录") finally: db.close() except Exception as e: logger.exception("Session清理任务异常") # 根据配置等待指定时间 await asyncio.sleep(cleanup_interval_seconds) async def _heartbeat_cleanup_task(self): """心跳监控缓存清理任务 - 根据配置间隔执行""" # 获取清理间隔配置 cleanup_interval_hours = SystemConfig.get_config_value('heartbeat_cleanup_interval_hours', 1) cleanup_interval_seconds = cleanup_interval_hours * 3600 logger.info(f"启动心跳缓存清理定时任务 (间隔: {cleanup_interval_hours}小时)") while self.running: try: logger.debug("执行心跳缓存清理...") cleaned_count = HeartbeatMonitor.cleanup_cache() if cleaned_count > 0: logger.info(f"清理了 {cleaned_count} 个过期的缓存条目") else: logger.debug("心跳缓存状态正常,无需清理") # 输出监控统计信息 stats = HeartbeatMonitor.get_monitoring_stats() logger.info(f"心跳监控统计: 监控用户={stats.get('total_monitored_users', 0)}, " f"受保护用户={stats.get('protected_users', 0)}, " f"失败用户={stats.get('failing_users', 0)}, " f"缓存大小={stats.get('cache_size', 0)}") except Exception as e: logger.exception("心跳缓存清理任务异常") # 根据配置等待指定时间 await asyncio.sleep(cleanup_interval_seconds) # 全局后台任务管理器实例 background_task_manager = BackgroundTaskManager() @asynccontextmanager async def lifespan_manager(app): """FastAPI lifespan 事件管理器""" # 启动时执行 logger.info("应用启动,开始初始化后台任务...") await background_task_manager.start_all_tasks() yield # 关闭时执行 logger.info("应用关闭,开始清理后台任务...") await background_task_manager.stop_all_tasks() # 便捷函数 async def start_background_tasks(): """启动后台任务(手动调用)""" await background_task_manager.start_all_tasks() async def stop_background_tasks(): """停止后台任务(手动调用)""" await background_task_manager.stop_all_tasks() def get_task_manager() -> BackgroundTaskManager: """获取后台任务管理器实例""" return background_task_manager