ruff ci

src/chat/message_manager/message_manager.py

@@ -7,7 +7,7 @@ import asyncio
 import random
 import time
 from collections import defaultdict, deque
-from typing import TYPE_CHECKING, Any, Dict
+from typing import TYPE_CHECKING, Any
 
 from src.chat.chatter_manager import ChatterManager
 from src.chat.message_receive.chat_stream import ChatStream
@@ -21,7 +21,6 @@ from src.plugin_system.apis.chat_api import get_chat_manager
 from .distribution_manager import stream_loop_manager
 from .sleep_system.state_manager import SleepState, sleep_state_manager
 
-
 if TYPE_CHECKING:
     pass
 
@@ -44,8 +43,8 @@ class MessageManager:
         self.chatter_manager = ChatterManager(self.action_manager)
 
         # 消息缓存系统 - 直接集成到消息管理器
-        self.message_caches: Dict[str, deque] = defaultdict(deque)  # 每个流的消息缓存
-        self.stream_processing_status: Dict[str, bool] = defaultdict(bool)  # 流的处理状态
+        self.message_caches: dict[str, deque] = defaultdict(deque)  # 每个流的消息缓存
+        self.stream_processing_status: dict[str, bool] = defaultdict(bool)  # 流的处理状态
         self.cache_stats = {
             "total_cached_messages": 0,
             "total_flushed_messages": 0,

src/chat/message_manager/sleep_system/sleep_logic.py

@@ -1,10 +1,10 @@
-from datetime import datetime, time, timedelta
 import random
-from typing import Optional, Tuple
+from datetime import datetime, timedelta
 
 from src.common.logger import get_logger
 from src.config.config import global_config
 from src.schedule.schedule_manager import schedule_manager
+
 from .state_manager import SleepState, sleep_state_manager
 
 logger = get_logger("sleep_logic")
@@ -77,7 +77,7 @@ class SleepLogic:
             logger.info(f"当前时间 {now.strftime('%H:%M')} 已到达或超过预定起床时间 {wake_up_time.strftime('%H:%M')}。")
             sleep_state_manager.set_state(SleepState.AWAKE)
 
-    def _should_be_sleeping(self, now: datetime) -> Tuple[bool, Optional[datetime]]:
+    def _should_be_sleeping(self, now: datetime) -> tuple[bool, datetime | None]:
         """
         判断在当前时刻，是否应该处于睡眠时间。
 
@@ -108,10 +108,10 @@ class SleepLogic:
                  return True, wake_up_time
             # 如果当前时间大于入睡时间，说明已经进入睡眠窗口
             return True, wake_up_time
-            
+
         return False, None
 
-    def _get_fixed_sleep_times(self, now: datetime) -> Tuple[Optional[datetime], Optional[datetime]]:
+    def _get_fixed_sleep_times(self, now: datetime) -> tuple[datetime | None, datetime | None]:
         """
         当使用“固定时间”模式时，从此方法计算睡眠和起床时间。
         会加入配置中的随机偏移量，让作息更自然。
@@ -129,7 +129,7 @@ class SleepLogic:
             wake_up_t = datetime.strptime(sleep_config.fixed_wake_up_time, "%H:%M").time()
 
             sleep_time = datetime.combine(now.date(), sleep_t) + timedelta(minutes=sleep_offset)
-            
+
             # 如果起床时间比睡觉时间早，说明是第二天
             wake_up_day = now.date() + timedelta(days=1) if wake_up_t < sleep_t else now.date()
             wake_up_time = datetime.combine(wake_up_day, wake_up_t) + timedelta(minutes=wake_up_offset)
@@ -139,7 +139,7 @@ class SleepLogic:
             logger.error(f"解析固定睡眠时间失败: {e}")
             return None, None
 
-    def _get_sleep_times_from_schedule(self, now: datetime) -> Tuple[Optional[datetime], Optional[datetime]]:
+    def _get_sleep_times_from_schedule(self, now: datetime) -> tuple[datetime | None, datetime | None]:
         """
         当使用“日程表”模式时，从此方法获取睡眠时间。
         实现了核心逻辑：
@@ -164,8 +164,8 @@ class SleepLogic:
         wake_up_time = None
 
         return sleep_time, wake_up_time
-    
-    def _get_wakeup_times_from_schedule(self, now: datetime) -> Tuple[Optional[datetime], Optional[datetime]]:
+
+    def _get_wakeup_times_from_schedule(self, now: datetime) -> tuple[datetime | None, datetime | None]:
             """
             当使用“日程表”模式时，从此方法获取睡眠时间。
             实现了核心逻辑：
@@ -192,4 +192,4 @@ class SleepLogic:
 
 
 # 全局单例
-sleep_logic = SleepLogic()
+sleep_logic = SleepLogic()

src/chat/message_manager/sleep_system/state_manager.py

@@ -1,6 +1,6 @@
 import enum
 from datetime import datetime, timedelta
-from typing import Any, Dict, Optional
+from typing import Any
 
 from src.common.logger import get_logger
 from src.manager.local_store_manager import local_storage
@@ -43,7 +43,7 @@ class SleepStateManager:
         """
         初始化状态管理器，定义状态数据结构并从本地加载历史状态。
         """
-        self.state: Dict[str, Any] = {}
+        self.state: dict[str, Any] = {}
         self._default_state()
         self.load_state()
 
@@ -115,9 +115,9 @@ class SleepStateManager:
     def set_state(
         self,
         new_state: SleepState,
-        duration_seconds: Optional[float] = None,
-        sleep_start: Optional[datetime] = None,
-        wake_up: Optional[datetime] = None,
+        duration_seconds: float | None = None,
+        sleep_start: datetime | None = None,
+        wake_up: datetime | None = None,
     ):
         """
         核心函数：切换到新的睡眠状态，并更新相关的状态数据。
@@ -132,7 +132,7 @@ class SleepStateManager:
         if new_state == SleepState.AWAKE:
             self._default_state() # 醒来时重置所有状态
             self.state["state"] = SleepState.AWAKE # 确保状态正确
-        
+
         elif new_state == SleepState.SLEEPING:
             self.state["sleep_start_time"] = (sleep_start or datetime.now()).isoformat()
             self.state["wake_up_time"] = wake_up.isoformat() if wake_up else None
@@ -153,7 +153,7 @@ class SleepStateManager:
         self.state["last_checked"] = datetime.now().isoformat()
         self.save_state()
 
-    def get_wake_up_time(self) -> Optional[datetime]:
+    def get_wake_up_time(self) -> datetime | None:
         """获取预定的起床时间，如果已设置的话。"""
         wake_up_str = self.state.get("wake_up_time")
         if wake_up_str:
@@ -163,7 +163,7 @@ class SleepStateManager:
                 return None
         return None
 
-    def get_sleep_start_time(self) -> Optional[datetime]:
+    def get_sleep_start_time(self) -> datetime | None:
         """获取本次睡眠的开始时间，如果已设置的话。"""
         sleep_start_str = self.state.get("sleep_start_time")
         if sleep_start_str:
@@ -187,4 +187,4 @@ class SleepStateManager:
 
 
 # 全局单例
-sleep_state_manager = SleepStateManager()
+sleep_state_manager = SleepStateManager()

src/chat/message_manager/sleep_system/tasks.py

@@ -1,5 +1,6 @@
 from src.common.logger import get_logger
 from src.manager.async_task_manager import AsyncTask, async_task_manager
+
 from .sleep_logic import sleep_logic
 
 logger = get_logger("sleep_tasks")