pre-commit

main.py

@@ -6,7 +6,6 @@ QQ群日常分析插件
 """
 
 import asyncio
-from typing import Optional
 
 from astrbot.api.event import filter
 from astrbot.api.star import Context, Star
@@ -79,8 +78,10 @@ class QQGroupDailyAnalysis(Star):
                 platform_count = len(discovered)
                 logger.info(f"Bot管理器初始化成功，发现 {platform_count} 个适配器")
                 for platform_id, bot_instance in discovered.items():
-                    logger.info(f"  - 平台 {platform_id}: {type(bot_instance).__name__}")
-                
+                    logger.info(
+                        f"  - 平台 {platform_id}: {type(bot_instance).__name__}"
+                    )
+
                 # 启动调度器
                 await auto_scheduler.start_scheduler()
             else:
@@ -139,7 +140,7 @@ class QQGroupDailyAnalysis(Star):
     @filter.command("群分析")
     @filter.permission_type(PermissionType.ADMIN)
     async def analyze_group_daily(
-        self, event: AiocqhttpMessageEvent, days: Optional[int] = None
+        self, event: AiocqhttpMessageEvent, days: int | None = None
     ):
         """
         分析群聊日常活动
@@ -177,9 +178,11 @@ class QQGroupDailyAnalysis(Star):
             # 获取该群对应的平台ID和bot实例
             platform_id = auto_scheduler._get_platform_id_for_group(group_id)
             bot_instance = bot_manager.get_bot_instance(platform_id)
-            
+
             if not bot_instance:
-                yield event.plain_result(f"❌ 未找到群 {group_id} 对应的bot实例（平台: {platform_id}）")
+                yield event.plain_result(
+                    f"❌ 未找到群 {group_id} 对应的bot实例（平台: {platform_id}）"
+                )
                 return
 
             # 获取群聊消息
@@ -431,7 +434,6 @@ class QQGroupDailyAnalysis(Star):
 • 输出格式: {output_format}
 • PDF 功能: {pdf_status}
 • 最小消息数: {min_threshold}
-• 最大查询轮数: {max_rounds}
 
 💡 可用命令: enable, disable, status, reload, test
 💡 支持的输出格式: image, text, pdf (图片和PDF包含活跃度可视化)

src/analysis/analyzers/base_analyzer.py

@@ -4,7 +4,7 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import List, Dict, Tuple, Any
+from typing import Any
 from astrbot.api import logger
 from ...models.data_models import TokenUsage
 from ..utils.json_utils import parse_json_response
@@ -76,7 +76,7 @@ class BaseAnalyzer(ABC):
         pass
 
     @abstractmethod
-    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+    def extract_with_regex(self, result_text: str, max_count: int) -> list[dict]:
         """
         使用正则表达式提取数据
 
@@ -90,7 +90,7 @@ class BaseAnalyzer(ABC):
         pass
 
     @abstractmethod
-    def create_data_objects(self, data_list: List[Dict]) -> List[Any]:
+    def create_data_objects(self, data_list: list[dict]) -> list[Any]:
         """
         创建数据对象列表
 
@@ -102,7 +102,7 @@ class BaseAnalyzer(ABC):
         """
         pass
 
-    async def analyze(self, data: Any, umo: str = None) -> Tuple[List[Any], TokenUsage]:
+    async def analyze(self, data: Any, umo: str = None) -> tuple[list[Any], TokenUsage]:
         """
         统一的分析流程
 

src/analysis/analyzers/golden_quote_analyzer.py

@@ -3,7 +3,6 @@
 专门处理群聊金句提取和分析
 """
 
-from typing import List, Dict, Tuple
 from datetime import datetime
 from astrbot.api import logger
 from ...models.data_models import GoldenQuote, TokenUsage
@@ -38,7 +37,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
         """获取温度参数"""
         return 0.7
 
-    def build_prompt(self, messages: List[Dict]) -> str:
+    def build_prompt(self, messages: list[dict]) -> str:
         """
         构建金句分析提示词
 
@@ -77,7 +76,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
         logger.warning("未找到有效的金句分析提示词配置，请检查配置文件")
         return ""
 
-    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+    def extract_with_regex(self, result_text: str, max_count: int) -> list[dict]:
         """
         使用正则表达式提取金句信息
 
@@ -90,7 +89,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
         """
         return extract_golden_quotes_with_regex(result_text, max_count)
 
-    def create_data_objects(self, quotes_data: List[Dict]) -> List[GoldenQuote]:
+    def create_data_objects(self, quotes_data: list[dict]) -> list[GoldenQuote]:
         """
         创建金句对象列表
 
@@ -125,7 +124,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
             logger.error(f"创建金句对象失败: {e}")
             return []
 
-    def extract_interesting_messages(self, messages: List[Dict]) -> List[Dict]:
+    def extract_interesting_messages(self, messages: list[dict]) -> list[dict]:
         """
         提取圣经的文本消息
 
@@ -161,8 +160,8 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
             return []
 
     async def analyze_golden_quotes(
-        self, messages: List[Dict], umo: str = None
-    ) -> Tuple[List[GoldenQuote], TokenUsage]:
+        self, messages: list[dict], umo: str = None
+    ) -> tuple[list[GoldenQuote], TokenUsage]:
         """
         分析群聊金句
 

src/analysis/analyzers/topic_analyzer.py

@@ -3,7 +3,6 @@
 专门处理群聊话题分析
 """
 
-from typing import List, Dict, Tuple
 from datetime import datetime
 import re
 from astrbot.api import logger
@@ -39,7 +38,7 @@ class TopicAnalyzer(BaseAnalyzer):
         """获取温度参数"""
         return 0.6
 
-    def build_prompt(self, messages: List[Dict]) -> str:
+    def build_prompt(self, messages: list[dict]) -> str:
         """
         构建话题分析提示词
 
@@ -83,7 +82,7 @@ class TopicAnalyzer(BaseAnalyzer):
                 # 获取发送者ID并过滤机器人消息
                 user_id = str(sender.get("user_id", ""))
                 bot_qq_ids = self.config_manager.get_bot_qq_ids()
-                
+
                 # 跳过机器人自己的消息
                 if bot_qq_ids and user_id in [str(qq) for qq in bot_qq_ids]:
                     logger.debug(f"build_prompt 过滤掉机器人QQ号: {user_id}")
@@ -200,7 +199,7 @@ class TopicAnalyzer(BaseAnalyzer):
         logger.warning("未找到有效的话题分析提示词配置，请检查配置文件")
         return ""
 
-    def extract_with_regex(self, result_text: str, max_topics: int) -> List[Dict]:
+    def extract_with_regex(self, result_text: str, max_topics: int) -> list[dict]:
         """
         使用正则表达式提取话题信息
 
@@ -213,7 +212,7 @@ class TopicAnalyzer(BaseAnalyzer):
         """
         return extract_topics_with_regex(result_text, max_topics)
 
-    def create_data_objects(self, topics_data: List[Dict]) -> List[SummaryTopic]:
+    def create_data_objects(self, topics_data: list[dict]) -> list[SummaryTopic]:
         """
         创建话题对象列表
 
@@ -286,7 +285,7 @@ class TopicAnalyzer(BaseAnalyzer):
             logger.error(f"创建话题对象失败: {e}", exc_info=True)
             return []
 
-    def extract_text_messages(self, messages: List[Dict]) -> List[Dict]:
+    def extract_text_messages(self, messages: list[dict]) -> list[dict]:
         """
         从群聊消息中提取文本消息
 
@@ -326,7 +325,7 @@ class TopicAnalyzer(BaseAnalyzer):
                 # 获取发送者ID并过滤机器人消息
                 user_id = str(sender.get("user_id", ""))
                 bot_qq_ids = self.config_manager.get_bot_qq_ids()
-                
+
                 # 跳过机器人自己的消息
                 if bot_qq_ids and user_id in [str(qq) for qq in bot_qq_ids]:
                     logger.debug(f"extract_text_messages 过滤掉机器人QQ号: {user_id}")
@@ -364,8 +363,8 @@ class TopicAnalyzer(BaseAnalyzer):
         return text_messages
 
     async def analyze_topics(
-        self, messages: List[Dict], umo: str = None
-    ) -> Tuple[List[SummaryTopic], TokenUsage]:
+        self, messages: list[dict], umo: str = None
+    ) -> tuple[list[SummaryTopic], TokenUsage]:
         """
         分析群聊话题
 

src/analysis/analyzers/user_title_analyzer.py

@@ -3,7 +3,6 @@
 专门处理用户称号和MBTI类型分析
 """
 
-from typing import List, Dict, Tuple
 from astrbot.api import logger
 from ...models.data_models import UserTitle, TokenUsage
 from .base_analyzer import BaseAnalyzer
@@ -36,7 +35,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
         """获取温度参数"""
         return 0.5
 
-    def build_prompt(self, user_data: Dict) -> str:
+    def build_prompt(self, user_data: dict) -> str:
         """
         构建用户称号分析提示词
 
@@ -79,7 +78,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
         logger.warning("未找到有效的用户称号分析提示词配置，请检查配置文件")
         return ""
 
-    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+    def extract_with_regex(self, result_text: str, max_count: int) -> list[dict]:
         """
         使用正则表达式提取用户称号信息
 
@@ -92,7 +91,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
         """
         return extract_user_titles_with_regex(result_text, max_count)
 
-    def create_data_objects(self, titles_data: List[Dict]) -> List[UserTitle]:
+    def create_data_objects(self, titles_data: list[dict]) -> list[UserTitle]:
         """
         创建用户称号对象列表
 
@@ -137,8 +136,8 @@ class UserTitleAnalyzer(BaseAnalyzer):
             return []
 
     def prepare_user_data(
-        self, messages: List[Dict], user_analysis: Dict, top_users: List[Dict] = None
-    ) -> Dict:
+        self, messages: list[dict], user_analysis: dict, top_users: list[dict] = None
+    ) -> dict:
         """
         准备用户数据
 
@@ -225,11 +224,11 @@ class UserTitleAnalyzer(BaseAnalyzer):
 
     async def analyze_user_titles(
         self,
-        messages: List[Dict],
-        user_analysis: Dict,
+        messages: list[dict],
+        user_analysis: dict,
         umo: str = None,
-        top_users: List[Dict] = None,
-    ) -> Tuple[List[UserTitle], TokenUsage]:
+        top_users: list[dict] = None,
+    ) -> tuple[list[UserTitle], TokenUsage]:
         """
         分析用户称号
 

src/analysis/llm_analyzer.py

@@ -4,7 +4,6 @@ LLM分析器模块
 """
 
 import asyncio
-from typing import List, Dict, Tuple
 from astrbot.api import logger
 from ..models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage
 from .analyzers.topic_analyzer import TopicAnalyzer
@@ -38,8 +37,8 @@ class LLMAnalyzer:
         self.golden_quote_analyzer = GoldenQuoteAnalyzer(context, config_manager)
 
     async def analyze_topics(
-        self, messages: List[Dict], umo: str = None
-    ) -> Tuple[List[SummaryTopic], TokenUsage]:
+        self, messages: list[dict], umo: str = None
+    ) -> tuple[list[SummaryTopic], TokenUsage]:
         """
         使用LLM分析话题
         保持原有接口，委托给专门的TopicAnalyzer处理
@@ -60,11 +59,11 @@ class LLMAnalyzer:
 
     async def analyze_user_titles(
         self,
-        messages: List[Dict],
-        user_analysis: Dict,
+        messages: list[dict],
+        user_analysis: dict,
         umo: str = None,
-        top_users: List[Dict] = None,
-    ) -> Tuple[List[UserTitle], TokenUsage]:
+        top_users: list[dict] = None,
+    ) -> tuple[list[UserTitle], TokenUsage]:
         """
         使用LLM分析用户称号
         保持原有接口，委托给专门的UserTitleAnalyzer处理
@@ -88,8 +87,8 @@ class LLMAnalyzer:
             return [], TokenUsage()
 
     async def analyze_golden_quotes(
-        self, messages: List[Dict], umo: str = None
-    ) -> Tuple[List[GoldenQuote], TokenUsage]:
+        self, messages: list[dict], umo: str = None
+    ) -> tuple[list[GoldenQuote], TokenUsage]:
         """
         使用LLM分析群聊金句
         保持原有接口，委托给专门的GoldenQuoteAnalyzer处理
@@ -110,11 +109,11 @@ class LLMAnalyzer:
 
     async def analyze_all_concurrent(
         self,
-        messages: List[Dict],
-        user_analysis: Dict,
+        messages: list[dict],
+        user_analysis: dict,
         umo: str = None,
-        top_users: List[Dict] = None,
-    ) -> Tuple[List[SummaryTopic], List[UserTitle], List[GoldenQuote], TokenUsage]:
+        top_users: list[dict] = None,
+    ) -> tuple[list[SummaryTopic], list[UserTitle], list[GoldenQuote], TokenUsage]:
         """
         并发执行所有分析任务（话题、用户称号、金句）
 

src/analysis/statistics.py

@@ -4,7 +4,6 @@
 """
 
 from datetime import datetime
-from typing import List, Dict
 from collections import defaultdict
 from .utils import InfoUtils
 
@@ -15,7 +14,7 @@ class UserAnalyzer:
     def __init__(self, config_manager):
         self.config_manager = config_manager
 
-    def analyze_users(self, messages: List[Dict]) -> Dict[str, Dict]:
+    def analyze_users(self, messages: list[dict]) -> dict[str, dict]:
         """分析用户活跃度"""
         # 获取机器人QQ号列表用于过滤
         bot_qq_ids = self.config_manager.get_bot_qq_ids()
@@ -78,8 +77,8 @@ class UserAnalyzer:
         return dict(user_stats)
 
     def get_top_users(
-        self, user_analysis: Dict[str, Dict], limit: int = 10
-    ) -> List[Dict]:
+        self, user_analysis: dict[str, dict], limit: int = 10
+    ) -> list[dict]:
         """获取最活跃的用户"""
         # 获取机器人QQ号列表用于过滤
         bot_qq_ids = self.config_manager.get_bot_qq_ids()
@@ -106,8 +105,8 @@ class UserAnalyzer:
         return users[:limit]
 
     def get_user_activity_pattern(
-        self, user_analysis: Dict[str, Dict], user_id: str
-    ) -> Dict:
+        self, user_analysis: dict[str, dict], user_id: str
+    ) -> dict:
         """获取用户活动模式"""
         if user_id not in user_analysis:
             return {}

src/analysis/utils/json_utils.py

@@ -5,7 +5,6 @@ JSON处理工具模块
 
 import json
 import re
-from typing import List, Dict, Tuple, Optional
 from astrbot.api import logger
 
 
@@ -83,7 +82,7 @@ def fix_json(text: str) -> str:
 
 def parse_json_response(
     result_text: str, data_type: str
-) -> Tuple[bool, Optional[List[Dict]], Optional[str]]:
+) -> tuple[bool, list[dict] | None, str | None]:
     """
     统一的JSON解析方法
 
@@ -125,7 +124,7 @@ def parse_json_response(
         return False, None, error_msg
 
 
-def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
+def extract_topics_with_regex(result_text: str, max_topics: int) -> list[dict]:
     """
     使用正则表达式提取话题信息
 
@@ -178,7 +177,7 @@ def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
         return []
 
 
-def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dict]:
+def extract_user_titles_with_regex(result_text: str, max_count: int) -> list[dict]:
     """
     使用正则表达式提取用户称号信息
 
@@ -223,7 +222,7 @@ def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dic
         return []
 
 
-def extract_golden_quotes_with_regex(result_text: str, max_count: int) -> List[Dict]:
+def extract_golden_quotes_with_regex(result_text: str, max_count: int) -> list[dict]:
     """
     使用正则表达式提取金句信息
 

src/analysis/utils/llm_utils.py

@@ -4,14 +4,11 @@ LLM API请求处理工具模块
 """
 
 import asyncio
-from typing import Optional, Any
+from typing import Any
 from astrbot.api import logger
-import aiohttp
 
 
-def _try_get_provider_by_id(
-    context, provider_id: str, description: str
-) -> Optional[Any]:
+def _try_get_provider_by_id(context, provider_id: str, description: str) -> Any | None:
     """
     尝试通过 ID 获取 Provider 的辅助函数
 
@@ -38,7 +35,7 @@ def _try_get_provider_by_id(
     return None
 
 
-def _try_get_session_provider(context, umo: str) -> Optional[Any]:
+def _try_get_session_provider(context, umo: str) -> Any | None:
     """
     尝试获取会话 Provider 的辅助函数
 
@@ -64,7 +61,7 @@ def _try_get_session_provider(context, umo: str) -> Optional[Any]:
     return None
 
 
-def _try_get_first_available_provider(context) -> Optional[Any]:
+def _try_get_first_available_provider(context) -> Any | None:
     """
     尝试获取第一个可用 Provider 的辅助函数
 
@@ -87,7 +84,7 @@ def _try_get_first_available_provider(context) -> Optional[Any]:
 
 def get_provider_with_fallback(
     context, config_manager, provider_id_key: str, umo: str = None
-) -> Optional[Any]:
+) -> Any | None:
     """
     根据配置键获取 Provider，支持多级回退
 
@@ -182,7 +179,7 @@ async def call_provider_with_retry(
     temperature: float,
     umo: str = None,
     provider_id_key: str = None,
-) -> Optional[Any]:
+) -> Any | None:
     """
     调用LLM提供者，带超时、重试与退避。支持自定义服务商和配置化 Provider 选择。
 
@@ -258,7 +255,7 @@ async def call_provider_with_retry(
     return None
 
 
-def extract_token_usage(response) -> Optional[dict]:
+def extract_token_usage(response) -> dict | None:
     """
     从LLM响应中提取token使用统计
 

src/core/bot_manager.py

@@ -3,9 +3,10 @@ Bot实例管理模块
 统一管理bot实例的获取、设置和使用
 """
 
-from typing import Dict, Any
+from typing import Any
 from astrbot.api import logger
 
+
 class BotManager:
     """Bot实例管理器 - 统一管理所有bot相关操作"""
 
@@ -25,7 +26,7 @@ class BotManager:
         """设置bot实例，支持指定平台ID"""
         if not platform_id:
             platform_id = self._get_platform_id_from_instance(bot_instance)
-        
+
         if bot_instance and platform_id:
             self._bot_instances[platform_id] = bot_instance
             # 自动提取QQ号
@@ -50,7 +51,7 @@ class BotManager:
             instance = self._bot_instances.get(platform_id)
             if instance:
                 return instance
-            
+
             # 如果指定的平台不存在，记录警告并尝试回退
             if self._bot_instances:
                 first_platform = list(self._bot_instances.keys())[0]
@@ -58,11 +59,11 @@ class BotManager:
                     f"平台 '{platform_id}' 不存在，回退到第一个可用平台 '{first_platform}'"
                 )
                 return self._bot_instances[first_platform]
-            
+
             # 没有任何平台可用
             logger.error(f"平台 '{platform_id}' 不存在，且没有任何可用的bot实例")
             return None
-        
+
         # 没有指定平台ID，返回第一个可用的实例
         if self._bot_instances:
             first_platform = list(self._bot_instances.keys())[0]
@@ -72,7 +73,7 @@ class BotManager:
                     f"(共有 {len(self._bot_instances)} 个平台: {list(self._bot_instances.keys())})"
                 )
             return self._bot_instances[first_platform]
-        
+
         # 没有任何平台可用
         logger.error("没有任何可用的bot实例")
         return None
@@ -102,7 +103,7 @@ class BotManager:
 
         platforms = getattr(self._context.platform_manager, "platform_insts", [])
         discovered = {}
-        
+
         for platform in platforms:
             # 获取bot实例
             bot_client = None
@@ -110,12 +111,16 @@ class BotManager:
                 bot_client = platform.get_client()
             elif hasattr(platform, "bot"):
                 bot_client = platform.bot
-            
-            if bot_client and hasattr(platform, "metadata") and hasattr(platform.metadata, "id"):
+
+            if (
+                bot_client
+                and hasattr(platform, "metadata")
+                and hasattr(platform.metadata, "id")
+            ):
                 platform_id = platform.metadata.id
                 self.set_bot_instance(bot_client, platform_id)
                 discovered[platform_id] = bot_client
-        
+
         return discovered
 
     async def initialize_from_config(self):
@@ -132,7 +137,7 @@ class BotManager:
         # 返回发现的实例字典
         return discovered
 
-    def get_status_info(self) -> Dict[str, Any]:
+    def get_status_info(self) -> dict[str, Any]:
         """获取bot管理器状态信息"""
         return {
             "has_bot_instance": self.has_bot_instance(),
@@ -152,7 +157,7 @@ class BotManager:
                 platform_id = event.platform
             elif hasattr(event, "metadata") and hasattr(event.metadata, "id"):
                 platform_id = event.metadata.id
-            
+
             self.set_bot_instance(event.bot, platform_id)
             # 每次都尝试从bot实例提取QQ号
             bot_qq_id = self._extract_bot_qq_id(event.bot)
@@ -186,7 +191,7 @@ class BotManager:
         """判断是否应该过滤bot自己的消息（支持多个QQ号）"""
         if not self._bot_qq_ids:
             return False
-        
+
         sender_id_str = str(sender_id)
         # 检查是否在QQ号列表中
         return sender_id_str in self._bot_qq_ids

src/core/config.py

@@ -4,7 +4,6 @@
 """
 
 import sys
-from typing import Optional, List
 from astrbot.api import logger, AstrBotConfig
 
 
@@ -17,7 +16,7 @@ class ConfigManager:
         self._pyppeteer_version = None
         self._check_pyppeteer_availability()
 
-    def get_enabled_groups(self) -> List[str]:
+    def get_enabled_groups(self) -> list[str]:
         """获取启用的群组列表"""
         return self.config.get("enabled_groups", [])
 
@@ -204,7 +203,7 @@ class ConfigManager:
         self.config["output_format"] = format_type
         self.config.save_config()
 
-    def set_enabled_groups(self, groups: List[str]):
+    def set_enabled_groups(self, groups: list[str]):
         """设置启用的群组列表"""
         self.config["enabled_groups"] = groups
         self.config.save_config()
@@ -300,7 +299,7 @@ class ConfigManager:
         return self._pyppeteer_available
 
     @property
-    def pyppeteer_version(self) -> Optional[str]:
+    def pyppeteer_version(self) -> str | None:
         """获取pyppeteer版本"""
         return self._pyppeteer_version
 

src/core/message_handler.py

@@ -4,7 +4,6 @@
 """
 
 from datetime import datetime, timedelta
-from typing import List, Dict
 from collections import defaultdict
 from astrbot.api import logger
 from ...src.models.data_models import GroupStatistics, TokenUsage, EmojiStatistics
@@ -48,7 +47,7 @@ class MessageHandler:
 
     async def fetch_group_messages(
         self, bot_instance, group_id: str, days: int, platform_id: str = None
-    ) -> List[Dict]:
+    ) -> list[dict]:
         """获取群聊消息记录"""
         try:
             # 验证参数
@@ -96,7 +95,10 @@ class MessageHandler:
                     except Exception as api_err:
                         error_msg = str(api_err)
                         # 检查是否是特定的错误码（1200表示不在该群）
-                        if "retcode=1200" in error_msg or "消息undefined不存在" in error_msg:
+                        if (
+                            "retcode=1200" in error_msg
+                            or "消息undefined不存在" in error_msg
+                        ):
                             logger.warning(f"群 {group_id} 机器人不在此群中: {api_err}")
                             return []
                         else:
@@ -179,7 +181,7 @@ class MessageHandler:
             logger.error(f"群 {group_id} 获取群聊消息记录失败: {e}", exc_info=True)
             return []
 
-    def calculate_statistics(self, messages: List[Dict]) -> GroupStatistics:
+    def calculate_statistics(self, messages: list[dict]) -> GroupStatistics:
         """计算基础统计数据"""
         total_chars = 0
         participants = set()

src/models/data_models.py

@@ -4,7 +4,6 @@
 """
 
 from dataclasses import dataclass, field
-from typing import List
 
 
 @dataclass
@@ -12,7 +11,7 @@ class SummaryTopic:
     """话题总结数据结构"""
 
     topic: str
-    contributors: List[str]
+    contributors: list[str]
     detail: str
 
 
@@ -87,7 +86,7 @@ class GroupStatistics:
     total_characters: int
     participant_count: int
     most_active_period: str
-    golden_quotes: List[GoldenQuote]
+    golden_quotes: list[GoldenQuote]
     emoji_count: int  # 保持向后兼容
     emoji_statistics: EmojiStatistics = field(default_factory=EmojiStatistics)
     activity_visualization: ActivityVisualization = field(

src/reports/generators.py

@@ -6,7 +6,6 @@
 import base64
 import aiohttp
 from datetime import datetime
-from typing import Dict, Optional
 from pathlib import Path
 from astrbot.api import logger
 from .templates import HTMLTemplates
@@ -22,8 +21,8 @@ class ReportGenerator:
         self.activity_visualizer = ActivityVisualizer()
 
     async def generate_image_report(
-        self, analysis_result: Dict, group_id: str, html_render_func
-    ) -> Optional[str]:
+        self, analysis_result: dict, group_id: str, html_render_func
+    ) -> str | None:
         """生成图片格式的分析报告"""
         try:
             # 准备渲染数据
@@ -68,8 +67,8 @@ class ReportGenerator:
                 return None
 
     async def generate_pdf_report(
-        self, analysis_result: Dict, group_id: str
-    ) -> Optional[str]:
+        self, analysis_result: dict, group_id: str
+    ) -> str | None:
         """生成PDF格式的分析报告"""
         try:
             # 确保输出目录存在
@@ -105,7 +104,7 @@ class ReportGenerator:
             logger.error(f"生成 PDF 报告失败: {e}")
             return None
 
-    def generate_text_report(self, analysis_result: Dict) -> str:
+    def generate_text_report(self, analysis_result: dict) -> str:
         """生成文本格式的分析报告"""
         stats = analysis_result["statistics"]
         topics = analysis_result["topics"]
@@ -146,7 +145,7 @@ class ReportGenerator:
 
         return report
 
-    async def _prepare_render_data(self, analysis_result: Dict) -> Dict:
+    async def _prepare_render_data(self, analysis_result: dict) -> dict:
         """准备渲染数据"""
         stats = analysis_result["statistics"]
         topics = analysis_result["topics"]
@@ -239,7 +238,7 @@ class ReportGenerator:
         }
 
     def _render_html_template(
-        self, template: str, data: Dict, use_jinja_style: bool = False
+        self, template: str, data: dict, use_jinja_style: bool = False
     ) -> str:
         """HTML模板渲染，支持两种占位符格式
 
@@ -281,7 +280,7 @@ class ReportGenerator:
 
         return result
 
-    async def _get_user_avatar(self, user_id: str) -> Optional[str]:
+    async def _get_user_avatar(self, user_id: str) -> str | None:
         """获取用户头像的base64编码"""
         try:
             avatar_url = f"https://q4.qlogo.cn/headimg_dl?dst_uin={user_id}&spec=640"

src/scheduler/auto_scheduler.py

@@ -45,16 +45,22 @@ class AutoScheduler:
         """根据群ID获取对应的平台ID"""
         try:
             # 首先检查已注册的bot实例
-            if hasattr(self.bot_manager, "_bot_instances") and self.bot_manager._bot_instances:
+            if (
+                hasattr(self.bot_manager, "_bot_instances")
+                and self.bot_manager._bot_instances
+            ):
                 # 如果只有一个实例，直接返回
                 if len(self.bot_manager._bot_instances) == 1:
                     platform_id = list(self.bot_manager._bot_instances.keys())[0]
                     logger.debug(f"只有一个适配器，使用平台: {platform_id}")
                     return platform_id
-                
+
                 # 如果有多个实例，尝试通过API检查群属于哪个适配器
                 logger.info(f"检测到多个适配器，正在验证群 {group_id} 属于哪个平台...")
-                for platform_id, bot_instance in self.bot_manager._bot_instances.items():
+                for (
+                    platform_id,
+                    bot_instance,
+                ) in self.bot_manager._bot_instances.items():
                     try:
                         # 尝试调用 get_group_info 来验证群是否存在
                         if hasattr(bot_instance, "call_action"):
@@ -65,26 +71,39 @@ class AutoScheduler:
                                 logger.info(f"✅ 群 {group_id} 属于平台 {platform_id}")
                                 return platform_id
                             else:
-                                logger.debug(f"平台 {platform_id} 返回了无效结果: {result}")
+                                logger.debug(
+                                    f"平台 {platform_id} 返回了无效结果: {result}"
+                                )
                         else:
-                            logger.debug(f"平台 {platform_id} 的 bot 实例没有 call_action 方法")
+                            logger.debug(
+                                f"平台 {platform_id} 的 bot 实例没有 call_action 方法"
+                            )
                     except Exception as e:
                         # 检查是否是特定的错误码（1200表示不在该群）
                         error_msg = str(e)
-                        if "retcode=1200" in error_msg or "消息undefined不存在" in error_msg:
-                            logger.debug(f"平台 {platform_id} 确认群 {group_id} 不存在: {e}")
+                        if (
+                            "retcode=1200" in error_msg
+                            or "消息undefined不存在" in error_msg
+                        ):
+                            logger.debug(
+                                f"平台 {platform_id} 确认群 {group_id} 不存在: {e}"
+                            )
                         else:
-                            logger.debug(f"平台 {platform_id} 无法获取群 {group_id} 信息: {e}")
+                            logger.debug(
+                                f"平台 {platform_id} 无法获取群 {group_id} 信息: {e}"
+                            )
                         continue
-                
+
                 # 如果所有适配器都尝试失败，记录警告并返回第一个
-                logger.warning(f"⚠️ 无法确定群 {group_id} 属于哪个平台，使用第一个适配器")
+                logger.warning(
+                    f"⚠️ 无法确定群 {group_id} 属于哪个平台，使用第一个适配器"
+                )
                 first_platform = list(self.bot_manager._bot_instances.keys())[0]
                 logger.warning(f"使用默认平台: {first_platform}")
                 return first_platform
-            
+
             # 没有任何bot实例，返回None
-            logger.error(f"❌ 没有注册的bot实例")
+            logger.error("❌ 没有注册的bot实例")
             return None
         except Exception as e:
             logger.error(f"❌ 获取平台ID失败: {e}")
@@ -250,49 +269,71 @@ class AutoScheduler:
                 messages = None
                 platform_id = None
                 bot_instance = None
-                
+
                 # 获取所有可用的平台ID和bot实例
-                if hasattr(self.bot_manager, "_bot_instances") and self.bot_manager._bot_instances:
+                if (
+                    hasattr(self.bot_manager, "_bot_instances")
+                    and self.bot_manager._bot_instances
+                ):
                     available_platforms = list(self.bot_manager._bot_instances.items())
-                    logger.info(f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试...")
-                    
+                    logger.info(
+                        f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试..."
+                    )
+
                     for test_platform_id, test_bot_instance in available_platforms:
                         try:
-                            logger.info(f"尝试使用平台 {test_platform_id} 获取群 {group_id} 的消息...")
-                            analysis_days = self.config_manager.get_analysis_days()
-                            test_messages = await self.message_handler.fetch_group_messages(
-                                test_bot_instance, group_id, analysis_days, test_platform_id
+                            logger.info(
+                                f"尝试使用平台 {test_platform_id} 获取群 {group_id} 的消息..."
                             )
-                            
+                            analysis_days = self.config_manager.get_analysis_days()
+                            test_messages = (
+                                await self.message_handler.fetch_group_messages(
+                                    test_bot_instance,
+                                    group_id,
+                                    analysis_days,
+                                    test_platform_id,
+                                )
+                            )
+
                             if test_messages and len(test_messages) > 0:
                                 # 成功获取到消息，使用这个平台
                                 messages = test_messages
                                 platform_id = test_platform_id
                                 bot_instance = test_bot_instance
-                                logger.info(f"✅ 群 {group_id} 成功通过平台 {platform_id} 获取到 {len(messages)} 条消息")
+                                logger.info(
+                                    f"✅ 群 {group_id} 成功通过平台 {platform_id} 获取到 {len(messages)} 条消息"
+                                )
                                 break
                             else:
-                                logger.debug(f"平台 {test_platform_id} 未获取到消息，继续尝试下一个平台")
+                                logger.debug(
+                                    f"平台 {test_platform_id} 未获取到消息，继续尝试下一个平台"
+                                )
                         except Exception as e:
-                            logger.debug(f"平台 {test_platform_id} 获取消息失败: {e}，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 获取消息失败: {e}，继续尝试下一个平台"
+                            )
                             continue
-                    
+
                     if not messages:
-                        logger.warning(f"群 {group_id} 所有平台都尝试失败，未获取到足够的消息记录")
+                        logger.warning(
+                            f"群 {group_id} 所有平台都尝试失败，未获取到足够的消息记录"
+                        )
                         return
                 else:
                     # 回退到原来的逻辑（单个平台）
                     logger.warning(f"群 {group_id} 没有多个平台可用，使用回退逻辑")
                     platform_id = await self._get_platform_id_for_group(group_id)
-                    
+
                     if not platform_id:
                         logger.error(f"❌ 群 {group_id} 无法获取平台ID，跳过分析")
                         return
-                    
+
                     bot_instance = self.bot_manager.get_bot_instance(platform_id)
-                    
+
                     if not bot_instance:
-                        logger.error(f"❌ 群 {group_id} 未找到对应的bot实例（平台: {platform_id}）")
+                        logger.error(
+                            f"❌ 群 {group_id} 未找到对应的bot实例（平台: {platform_id}）"
+                        )
                         return
 
                     # 获取群聊消息
@@ -448,38 +489,56 @@ class AutoScheduler:
         """发送图片消息到群 - 依次尝试所有可用平台"""
         try:
             # 获取所有可用的平台，依次尝试发送
-            if hasattr(self.bot_manager, "_bot_instances") and self.bot_manager._bot_instances:
+            if (
+                hasattr(self.bot_manager, "_bot_instances")
+                and self.bot_manager._bot_instances
+            ):
                 available_platforms = list(self.bot_manager._bot_instances.items())
-                logger.info(f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送图片...")
-                
+                logger.info(
+                    f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送图片..."
+                )
+
                 for test_platform_id, test_bot_instance in available_platforms:
                     try:
-                        logger.info(f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送图片...")
-                        
+                        logger.info(
+                            f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送图片..."
+                        )
+
                         # 发送图片消息到群
                         await test_bot_instance.api.call_action(
                             "send_group_msg",
                             group_id=group_id,
                             message=[
-                                {"type": "text", "data": {"text": "📊 每日群聊分析报告已生成："}},
+                                {
+                                    "type": "text",
+                                    "data": {"text": "📊 每日群聊分析报告已生成："},
+                                },
                                 {"type": "image", "data": {"url": image_url}},
                             ],
                         )
-                        logger.info(f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送图片")
+                        logger.info(
+                            f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送图片"
+                        )
                         return True  # 成功发送，返回
-                        
+
                     except Exception as e:
                         error_msg = str(e)
                         # 检查是否是特定的错误码
                         if "retcode=1200" in error_msg:
                             if "rich media transfer failed" in error_msg:
-                                logger.debug(f"平台 {test_platform_id} 图片发送失败：媒体传输失败，继续尝试下一个平台")
+                                logger.debug(
+                                    f"平台 {test_platform_id} 图片发送失败：媒体传输失败，继续尝试下一个平台"
+                                )
                             else:
-                                logger.debug(f"平台 {test_platform_id} 图片发送失败：机器人可能不在此群中，继续尝试下一个平台")
+                                logger.debug(
+                                    f"平台 {test_platform_id} 图片发送失败：机器人可能不在此群中，继续尝试下一个平台"
+                                )
                         else:
-                            logger.debug(f"平台 {test_platform_id} 图片发送失败: {e}，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 图片发送失败: {e}，继续尝试下一个平台"
+                            )
                         continue
-                
+
                 # 所有平台都尝试失败
                 logger.error(f"❌ 群 {group_id} 所有平台都尝试发送图片失败")
                 return False
@@ -487,15 +546,17 @@ class AutoScheduler:
                 # 回退到原来的逻辑（单个平台）
                 logger.warning(f"群 {group_id} 没有多个平台可用，使用回退逻辑")
                 platform_id = await self._get_platform_id_for_group(group_id)
-                
+
                 if not platform_id:
                     logger.error(f"❌ 群 {group_id} 无法获取平台ID，无法发送图片")
                     return False
-                
+
                 bot_instance = self.bot_manager.get_bot_instance(platform_id)
-                
+
                 if not bot_instance:
-                    logger.error(f"❌ 群 {group_id} 发送图片失败：缺少bot实例（平台: {platform_id}）")
+                    logger.error(
+                        f"❌ 群 {group_id} 发送图片失败：缺少bot实例（平台: {platform_id}）"
+                    )
                     return False
 
                 # 发送图片消息到群
@@ -503,7 +564,10 @@ class AutoScheduler:
                     "send_group_msg",
                     group_id=group_id,
                     message=[
-                        {"type": "text", "data": {"text": "📊 每日群聊分析报告已生成："}},
+                        {
+                            "type": "text",
+                            "data": {"text": "📊 每日群聊分析报告已生成："},
+                        },
                         {"type": "image", "data": {"url": image_url}},
                     ],
                 )
@@ -518,30 +582,43 @@ class AutoScheduler:
         """发送文本消息到群 - 依次尝试所有可用平台"""
         try:
             # 获取所有可用的平台，依次尝试发送
-            if hasattr(self.bot_manager, "_bot_instances") and self.bot_manager._bot_instances:
+            if (
+                hasattr(self.bot_manager, "_bot_instances")
+                and self.bot_manager._bot_instances
+            ):
                 available_platforms = list(self.bot_manager._bot_instances.items())
-                logger.info(f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送文本...")
-                
+                logger.info(
+                    f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送文本..."
+                )
+
                 for test_platform_id, test_bot_instance in available_platforms:
                     try:
-                        logger.info(f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送文本...")
-                        
+                        logger.info(
+                            f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送文本..."
+                        )
+
                         # 发送文本消息到群
                         await test_bot_instance.api.call_action(
                             "send_group_msg", group_id=group_id, message=text_content
                         )
-                        logger.info(f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送文本")
+                        logger.info(
+                            f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送文本"
+                        )
                         return True  # 成功发送，返回
-                        
+
                     except Exception as e:
                         error_msg = str(e)
                         # 检查是否是特定的错误码
                         if "retcode=1200" in error_msg:
-                            logger.debug(f"平台 {test_platform_id} 发送文本失败：机器人可能不在此群中，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 发送文本失败：机器人可能不在此群中，继续尝试下一个平台"
+                            )
                         else:
-                            logger.debug(f"平台 {test_platform_id} 发送文本失败: {e}，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 发送文本失败: {e}，继续尝试下一个平台"
+                            )
                         continue
-                
+
                 # 所有平台都尝试失败
                 logger.error(f"❌ 群 {group_id} 所有平台都尝试发送文本失败")
                 return False
@@ -549,15 +626,17 @@ class AutoScheduler:
                 # 回退到原来的逻辑（单个平台）
                 logger.warning(f"群 {group_id} 没有多个平台可用，使用回退逻辑")
                 platform_id = await self._get_platform_id_for_group(group_id)
-                
+
                 if not platform_id:
                     logger.error(f"❌ 群 {group_id} 无法获取平台ID，无法发送文本")
                     return False
-                
+
                 bot_instance = self.bot_manager.get_bot_instance(platform_id)
-                
+
                 if not bot_instance:
-                    logger.error(f"❌ 群 {group_id} 发送文本失败：缺少bot实例（平台: {platform_id}）")
+                    logger.error(
+                        f"❌ 群 {group_id} 发送文本失败：缺少bot实例（平台: {platform_id}）"
+                    )
                     return False
 
                 # 发送文本消息到群
@@ -575,35 +654,51 @@ class AutoScheduler:
         """发送PDF文件到群 - 依次尝试所有可用平台"""
         try:
             # 获取所有可用的平台，依次尝试发送
-            if hasattr(self.bot_manager, "_bot_instances") and self.bot_manager._bot_instances:
+            if (
+                hasattr(self.bot_manager, "_bot_instances")
+                and self.bot_manager._bot_instances
+            ):
                 available_platforms = list(self.bot_manager._bot_instances.items())
-                logger.info(f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送PDF...")
-                
+                logger.info(
+                    f"群 {group_id} 检测到 {len(available_platforms)} 个可用平台，开始依次尝试发送PDF..."
+                )
+
                 for test_platform_id, test_bot_instance in available_platforms:
                     try:
-                        logger.info(f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送PDF...")
-                        
+                        logger.info(
+                            f"尝试使用平台 {test_platform_id} 向群 {group_id} 发送PDF..."
+                        )
+
                         # 发送PDF文件到群
                         await test_bot_instance.api.call_action(
                             "send_group_msg",
                             group_id=group_id,
                             message=[
-                                {"type": "text", "data": {"text": "📊 每日群聊分析报告已生成："}},
+                                {
+                                    "type": "text",
+                                    "data": {"text": "📊 每日群聊分析报告已生成："},
+                                },
                                 {"type": "file", "data": {"file": pdf_path}},
                             ],
                         )
-                        logger.info(f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送PDF")
+                        logger.info(
+                            f"✅ 群 {group_id} 成功通过平台 {test_platform_id} 发送PDF"
+                        )
                         return True  # 成功发送，返回
-                        
+
                     except Exception as e:
                         error_msg = str(e)
                         # 检查是否是特定的错误码
                         if "retcode=1200" in error_msg:
-                            logger.debug(f"平台 {test_platform_id} 发送PDF失败：机器人可能不在此群中，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 发送PDF失败：机器人可能不在此群中，继续尝试下一个平台"
+                            )
                         else:
-                            logger.debug(f"平台 {test_platform_id} 发送PDF失败: {e}，继续尝试下一个平台")
+                            logger.debug(
+                                f"平台 {test_platform_id} 发送PDF失败: {e}，继续尝试下一个平台"
+                            )
                         continue
-                
+
                 # 所有平台都尝试失败
                 logger.error(f"❌ 群 {group_id} 所有平台都尝试发送PDF失败")
                 return False
@@ -611,15 +706,17 @@ class AutoScheduler:
                 # 回退到原来的逻辑（单个平台）
                 logger.warning(f"群 {group_id} 没有多个平台可用，使用回退逻辑")
                 platform_id = await self._get_platform_id_for_group(group_id)
-                
+
                 if not platform_id:
                     logger.error(f"❌ 群 {group_id} 无法获取平台ID，无法发送PDF")
                     return False
-                
+
                 bot_instance = self.bot_manager.get_bot_instance(platform_id)
-                
+
                 if not bot_instance:
-                    logger.error(f"❌ 群 {group_id} 发送PDF失败：缺少bot实例（平台: {platform_id}）")
+                    logger.error(
+                        f"❌ 群 {group_id} 发送PDF失败：缺少bot实例（平台: {platform_id}）"
+                    )
                     return False
 
                 # 发送PDF文件到群
@@ -627,7 +724,10 @@ class AutoScheduler:
                     "send_group_msg",
                     group_id=group_id,
                     message=[
-                        {"type": "text", "data": {"text": "📊 每日群聊分析报告已生成："}},
+                        {
+                            "type": "text",
+                            "data": {"text": "📊 每日群聊分析报告已生成："},
+                        },
                         {"type": "file", "data": {"file": pdf_path}},
                     ],
                 )

src/utils/helpers.py

@@ -3,7 +3,6 @@
 包含消息分析和其他通用功能
 """
 
-from typing import List, Dict
 from ...src.models.data_models import TokenUsage
 from ...src.core.message_handler import MessageHandler
 from ...src.analysis.llm_analyzer import LLMAnalyzer
@@ -44,8 +43,8 @@ class MessageAnalyzer:
                 await self.message_handler.set_bot_qq_ids([bot_qq_id])
 
     async def analyze_messages(
-        self, messages: List[Dict], group_id: str, unified_msg_origin: str = None
-    ) -> Dict:
+        self, messages: list[dict], group_id: str, unified_msg_origin: str = None
+    ) -> dict:
         """完整的消息分析流程"""
         try:
             # 基础统计

src/visualization/activity_charts.py

@@ -5,7 +5,6 @@
 
 from datetime import datetime
 from collections import defaultdict
-from typing import Dict, List
 from ..models.data_models import ActivityVisualization
 
 
@@ -16,7 +15,7 @@ class ActivityVisualizer:
         pass
 
     def generate_activity_visualization(
-        self, messages: List[Dict]
+        self, messages: list[dict]
     ) -> ActivityVisualization:
         """生成活跃度可视化数据 - 专注于小时级别分析"""
         hourly_activity = defaultdict(int)