diff --git a/main.py b/main.py index 83401d9..9ba34f2 100644 --- a/main.py +++ b/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包含活跃度可视化) diff --git a/src/analysis/analyzers/base_analyzer.py b/src/analysis/analyzers/base_analyzer.py index 407e115..a75faa5 100644 --- a/src/analysis/analyzers/base_analyzer.py +++ b/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]: """ 统一的分析流程 diff --git a/src/analysis/analyzers/golden_quote_analyzer.py b/src/analysis/analyzers/golden_quote_analyzer.py index 9f832d4..a772595 100644 --- a/src/analysis/analyzers/golden_quote_analyzer.py +++ b/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]: """ 分析群聊金句 diff --git a/src/analysis/analyzers/topic_analyzer.py b/src/analysis/analyzers/topic_analyzer.py index 586cb81..5cc5873 100644 --- a/src/analysis/analyzers/topic_analyzer.py +++ b/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]: """ 分析群聊话题 diff --git a/src/analysis/analyzers/user_title_analyzer.py b/src/analysis/analyzers/user_title_analyzer.py index 8c12b3e..c654e25 100644 --- a/src/analysis/analyzers/user_title_analyzer.py +++ b/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]: """ 分析用户称号 diff --git a/src/analysis/llm_analyzer.py b/src/analysis/llm_analyzer.py index abe528a..59eccb3 100644 --- a/src/analysis/llm_analyzer.py +++ b/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]: """ 并发执行所有分析任务(话题、用户称号、金句) diff --git a/src/analysis/statistics.py b/src/analysis/statistics.py index 2db5d54..8080b54 100644 --- a/src/analysis/statistics.py +++ b/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 {} diff --git a/src/analysis/utils/json_utils.py b/src/analysis/utils/json_utils.py index d17fcd5..41b3d96 100644 --- a/src/analysis/utils/json_utils.py +++ b/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]: """ 使用正则表达式提取金句信息 diff --git a/src/analysis/utils/llm_utils.py b/src/analysis/utils/llm_utils.py index 08b6e12..01f515d 100644 --- a/src/analysis/utils/llm_utils.py +++ b/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使用统计 diff --git a/src/core/bot_manager.py b/src/core/bot_manager.py index 344cff8..fd83f84 100644 --- a/src/core/bot_manager.py +++ b/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 diff --git a/src/core/config.py b/src/core/config.py index b3a6d37..1826173 100644 --- a/src/core/config.py +++ b/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 diff --git a/src/core/message_handler.py b/src/core/message_handler.py index b142ff1..cd330d5 100644 --- a/src/core/message_handler.py +++ b/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() diff --git a/src/models/data_models.py b/src/models/data_models.py index 688c25d..d39240b 100644 --- a/src/models/data_models.py +++ b/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( diff --git a/src/reports/generators.py b/src/reports/generators.py index df90aa2..b1db42a 100644 --- a/src/reports/generators.py +++ b/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" diff --git a/src/scheduler/auto_scheduler.py b/src/scheduler/auto_scheduler.py index 8dc91b4..5455119 100644 --- a/src/scheduler/auto_scheduler.py +++ b/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}}, ], ) diff --git a/src/utils/helpers.py b/src/utils/helpers.py index cfc2e81..3f701af 100644 --- a/src/utils/helpers.py +++ b/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: # 基础统计 diff --git a/src/visualization/activity_charts.py b/src/visualization/activity_charts.py index dcd3977..5c1a44e 100644 --- a/src/visualization/activity_charts.py +++ b/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)