From e1ad6c9fc938ed9dff775bca3294ce7f3249be6d Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Mon, 9 Feb 2026 13:08:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=AD=A3=E7=A1=AE=E7=9A=84=E5=AF=BC?= =?UTF-8?q?=E5=85=A5=E5=90=84=E9=87=8D=E6=9E=84=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/__init__.py | 2 - src/application/__init__.py | 10 +- src/application/message_converter.py | 222 --------------- src/application/reporting_service.py | 263 ------------------ src/application/scheduling_service.py | 263 ------------------ .../services/analysis_application_service.py | 10 +- src/infrastructure/__init__.py | 28 -- .../analysis/utils/llm_utils.py | 4 +- src/infrastructure/llm/__init__.py | 7 - src/infrastructure/llm/llm_client.py | 187 ------------- .../messaging/message_sender.py | 53 ++++ src/infrastructure/reporting/dispatcher.py | 4 +- src/infrastructure/resilience/__init__.py | 15 - .../resilience/circuit_breaker.py | 137 --------- src/infrastructure/resilience/rate_limiter.py | 139 --------- src/infrastructure/resilience/retry.py | 176 ------------ .../scheduler/auto_scheduler.py | 4 +- src/utils/__init__.py | 3 +- 18 files changed, 66 insertions(+), 1461 deletions(-) delete mode 100644 src/application/message_converter.py delete mode 100644 src/application/reporting_service.py delete mode 100644 src/application/scheduling_service.py delete mode 100644 src/infrastructure/llm/__init__.py delete mode 100644 src/infrastructure/llm/llm_client.py create mode 100644 src/infrastructure/messaging/message_sender.py delete mode 100644 src/infrastructure/resilience/__init__.py delete mode 100644 src/infrastructure/resilience/circuit_breaker.py delete mode 100644 src/infrastructure/resilience/rate_limiter.py delete mode 100644 src/infrastructure/resilience/retry.py diff --git a/src/__init__.py b/src/__init__.py index dee30f0..e8f259c 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -15,5 +15,3 @@ QQ群日常分析插件 - 源代码包 - utils: 工具函数 - visualization: 可视化组件 """ - -__version__ = "2.0.0" diff --git a/src/application/__init__.py b/src/application/__init__.py index 55c8c18..e7c6d1e 100644 --- a/src/application/__init__.py +++ b/src/application/__init__.py @@ -1,9 +1 @@ -from .message_converter import MessageConverter -from .reporting_service import ReportingService -from .scheduling_service import SchedulingService - -__all__ = [ - "MessageConverter", - "SchedulingService", - "ReportingService", -] +# 应用层 - 编排和用例 diff --git a/src/application/message_converter.py b/src/application/message_converter.py deleted file mode 100644 index a4d4f36..0000000 --- a/src/application/message_converter.py +++ /dev/null @@ -1,222 +0,0 @@ -""" -消息转换器 - 连接原始平台消息和 UnifiedMessage - -该模块通过在原始平台消息格式和新的 UnifiedMessage 格式之间进行转换， -提供向后兼容性。 -""" - -from ..domain.value_objects.unified_message import ( - MessageContent, - MessageContentType, - UnifiedMessage, -) - - -class MessageConverter: - """ - 在原始平台消息和 UnifiedMessage 格式之间进行转换。 - - 这提供了一个迁移路径：现有代码可以继续使用原始字典， - 而新代码使用 UnifiedMessage。 - """ - - @staticmethod - def from_onebot_message(raw_msg: dict, group_id: str) -> UnifiedMessage | None: - """ - 将 OneBot v11 原始消息转换为 UnifiedMessage。 - - Args: - raw_msg: 来自 OneBot API 的原始消息字典 - group_id: 群组 ID - - Returns: - UnifiedMessage 或 None（如果转换失败） - """ - try: - sender = raw_msg.get("sender", {}) - message_chain = raw_msg.get("message", []) - - # 处理字符串消息格式 - if isinstance(message_chain, str): - message_chain = [{"type": "text", "data": {"text": message_chain}}] - - contents = [] - text_parts = [] - - for seg in message_chain: - seg_type = seg.get("type", "") - seg_data = seg.get("data", {}) - - if seg_type == "text": - text = seg_data.get("text", "") - text_parts.append(text) - contents.append( - MessageContent(type=MessageContentType.TEXT, text=text) - ) - - elif seg_type == "image": - contents.append( - MessageContent( - type=MessageContentType.IMAGE, - url=seg_data.get("url", seg_data.get("file", "")), - ) - ) - - elif seg_type == "at": - contents.append( - MessageContent( - type=MessageContentType.AT, - at_user_id=str(seg_data.get("qq", "")), - ) - ) - - elif seg_type in ("face", "mface", "bface", "sface"): - contents.append( - MessageContent( - type=MessageContentType.EMOJI, - emoji_id=str(seg_data.get("id", "")), - raw_data={"face_type": seg_type}, - ) - ) - - elif seg_type == "reply": - contents.append( - MessageContent( - type=MessageContentType.REPLY, - raw_data={"reply_id": seg_data.get("id", "")}, - ) - ) - - elif seg_type == "forward": - contents.append( - MessageContent( - type=MessageContentType.FORWARD, raw_data=seg_data - ) - ) - - elif seg_type == "record": - contents.append( - MessageContent( - type=MessageContentType.VOICE, - url=seg_data.get("url", seg_data.get("file", "")), - ) - ) - - elif seg_type == "video": - contents.append( - MessageContent( - type=MessageContentType.VIDEO, - url=seg_data.get("url", seg_data.get("file", "")), - ) - ) - - else: - contents.append( - MessageContent(type=MessageContentType.UNKNOWN, raw_data=seg) - ) - - # 从内容中提取 reply_to - reply_to = None - for c in contents: - if c.type == MessageContentType.REPLY and c.raw_data: - reply_to = str(c.raw_data.get("reply_id", "")) - break - - return UnifiedMessage( - message_id=str(raw_msg.get("message_id", "")), - sender_id=str(sender.get("user_id", "")), - sender_name=sender.get("nickname", ""), - sender_card=sender.get("card", "") or None, - group_id=group_id, - text_content="".join(text_parts), - contents=tuple(contents), - timestamp=raw_msg.get("time", 0), - platform="onebot", - reply_to_id=reply_to, - ) - - except Exception: - return None - - @staticmethod - def to_onebot_message(unified: UnifiedMessage) -> dict: - """ - 将 UnifiedMessage 转换回 OneBot v11 原始格式。 - - 用于与期望原始字典的现有代码向后兼容。 - """ - message_chain = [] - - for content in unified.contents: - if content.type == MessageContentType.TEXT: - message_chain.append({"type": "text", "data": {"text": content.text}}) - elif content.type == MessageContentType.IMAGE: - message_chain.append({"type": "image", "data": {"url": content.url}}) - elif content.type == MessageContentType.AT: - message_chain.append({"type": "at", "data": {"qq": content.at_user_id}}) - elif content.type == MessageContentType.EMOJI: - face_type = ( - content.raw_data.get("face_type", "face") - if content.raw_data - else "face" - ) - message_chain.append( - {"type": face_type, "data": {"id": content.emoji_id}} - ) - elif content.type == MessageContentType.REPLY: - reply_id = ( - content.raw_data.get("reply_id", "") if content.raw_data else "" - ) - message_chain.append({"type": "reply", "data": {"id": reply_id}}) - elif content.type == MessageContentType.VOICE: - message_chain.append({"type": "record", "data": {"url": content.url}}) - elif content.type == MessageContentType.VIDEO: - message_chain.append({"type": "video", "data": {"url": content.url}}) - - # 确保填充发送者字段，即使原始数据中缺失 - sender_data = { - "user_id": unified.sender_id, - "nickname": unified.sender_name, - "card": unified.sender_card or "", - } - - return { - "message_id": unified.message_id, - "sender": sender_data, - "group_id": unified.group_id, - "message": message_chain, - "time": unified.timestamp, - # 添加这些辅助字段，以便旧分析器可以直接使用 - "raw_message": unified.text_content, - "user_id": unified.sender_id, - } - - @staticmethod - def batch_from_onebot( - raw_messages: list[dict], group_id: str - ) -> list[UnifiedMessage]: - """将一批 OneBot 消息转换为 UnifiedMessage 列表。""" - result = [] - for raw_msg in raw_messages: - unified = MessageConverter.from_onebot_message(raw_msg, group_id) - if unified: - result.append(unified) - return result - - @staticmethod - def batch_to_onebot(unified_messages: list[UnifiedMessage]) -> list[dict]: - """将一批 UnifiedMessage 转换为 OneBot 原始格式。""" - return [MessageConverter.to_onebot_message(msg) for msg in unified_messages] - - @staticmethod - def unified_to_analysis_text(messages: list[UnifiedMessage]) -> str: - """ - 将 UnifiedMessage 列表转换为 LLM 分析文本格式。 - - 这是现有 LLM 分析器期望的格式。 - """ - lines = [] - for msg in messages: - if msg.has_text(): - lines.append(msg.to_analysis_format()) - return "\n".join(lines) diff --git a/src/application/reporting_service.py b/src/application/reporting_service.py deleted file mode 100644 index 37589c0..0000000 --- a/src/application/reporting_service.py +++ /dev/null @@ -1,263 +0,0 @@ -""" -报告服务 - 生成和发送报告的应用服务 - -该服务协调报告的生成并将其发送到群组。 -""" - -from datetime import datetime -from typing import Any - -from ..domain.services import ReportGenerator -from ..domain.value_objects.golden_quote import GoldenQuote -from ..domain.value_objects.statistics import GroupStatistics -from ..domain.value_objects.topic import Topic -from ..domain.value_objects.user_title import UserTitle -from ..infrastructure.config import ConfigManager -from ..infrastructure.persistence import HistoryRepository - - -class ReportingService: - """ - 生成和管理报告的应用服务。 - - 该服务协调领域服务和基础设施 - 以生成和发送分析报告。 - """ - - def __init__( - self, - config: ConfigManager, - history_repository: HistoryRepository, - ): - """ - 初始化报告服务。 - - Args: - config: 配置管理器 - history_repository: 用于存储报告的仓库 - """ - self.config = config - self.history = history_repository - - def generate_report( - self, - group_id: str, - group_name: str, - statistics: GroupStatistics, - topics: list[Topic], - user_titles: list[UserTitle], - golden_quotes: list[GoldenQuote], - date_str: str | None = None, - ) -> str: - """ - 生成完整的分析报告。 - - Args: - group_id: 群组标识符 - group_name: 群组显示名称 - statistics: 群组统计 - topics: 讨论话题列表 - user_titles: 用户称号列表 - golden_quotes: 金句列表 - date_str: 报告日期（默认为今天） - - Returns: - 格式化的报告字符串 - """ - date_str = date_str or datetime.now().strftime("%Y-%m-%d") - - generator = ReportGenerator( - group_name=group_name, - date_str=date_str, - ) - - # 根据配置生成报告 - report = generator.generate_full_report( - statistics=statistics, - topics=topics if self.config.get_include_topics() else [], - user_titles=user_titles if self.config.get_include_user_titles() else [], - golden_quotes=golden_quotes - if self.config.get_include_golden_quotes() - else [], - include_header=True, - include_footer=True, - ) - - return report - - def generate_summary( - self, - group_id: str, - statistics: GroupStatistics, - top_topic: Topic | None = None, - top_quote: GoldenQuote | None = None, - date_str: str | None = None, - ) -> str: - """ - 生成简要摘要报告。 - - Args: - group_id: 群组标识符 - statistics: 群组统计 - top_topic: 最重要的话题 - top_quote: 最佳金句 - date_str: 报告日期 - - Returns: - 简要摘要字符串 - """ - date_str = date_str or datetime.now().strftime("%Y-%m-%d") - - generator = ReportGenerator(date_str=date_str) - return generator.generate_summary_report( - statistics=statistics, - top_topic=top_topic, - top_quote=top_quote, - ) - - def save_report( - self, - group_id: str, - report_data: dict[str, Any], - date_str: str | None = None, - ) -> bool: - """ - 保存报告到历史记录。 - - Args: - group_id: 群组标识符 - report_data: 报告数据字典 - date_str: 报告日期 - - Returns: - 如果保存成功则返回 True - """ - date_str = date_str or datetime.now().strftime("%Y-%m-%d") - - return self.history.save_analysis_result( - group_id=group_id, - result=report_data, - date_str=date_str, - ) - - def get_report( - self, - group_id: str, - date_str: str, - ) -> dict[str, Any] | None: - """ - 获取已保存的报告。 - - Args: - group_id: 群组标识符 - date_str: 报告日期 - - Returns: - 报告数据或 None - """ - return self.history.get_analysis_result(group_id, date_str) - - def get_recent_reports( - self, - group_id: str, - limit: int = 7, - ) -> list[dict[str, Any]]: - """ - 获取群组的最近报告。 - - Args: - group_id: 群组标识符 - limit: 最大报告数 - - Returns: - 报告数据字典列表 - """ - return self.history.get_recent_results(group_id, limit) - - def has_report_for_today(self, group_id: str) -> bool: - """ - 检查今天是否已存在报告。 - - Args: - group_id: 群组标识符 - - Returns: - 如果报告存在则返回 True - """ - today = datetime.now().strftime("%Y-%m-%d") - return self.history.has_analysis_for_date(group_id, today) - - def format_for_platform( - self, - report: str, - platform: str, - format_type: str | None = None, - ) -> str: - """ - 为特定平台格式化报告。 - - Args: - report: 原始报告文本 - platform: 目标平台 - format_type: 覆盖格式类型 - - Returns: - 平台格式化的报告 - """ - format_type = format_type or self.config.get_report_format() - - # 目前保持原样返回。可以扩展为平台特定的格式化 - if format_type == "markdown": - return report - elif format_type == "text": - # 去除 markdown 格式 - return self._strip_markdown(report) - else: - return report - - def _strip_markdown(self, text: str) -> str: - """从文本中去除 markdown 格式。""" - # 简单的 markdown 去除 - import re - - # 去除加粗 - text = re.sub(r"\*\*(.*?)\*\*", r"\1", text) - # 去除斜体 - text = re.sub(r"\*(.*?)\*", r"\1", text) - # 去除标题 - text = re.sub(r"^#+\s*", "", text, flags=re.MULTILINE) - - return text - - def create_report_data( - self, - group_id: str, - group_name: str, - statistics: GroupStatistics, - topics: list[Topic], - user_titles: list[UserTitle], - golden_quotes: list[GoldenQuote], - ) -> dict[str, Any]: - """ - 创建用于存储的报告数据字典。 - - Args: - group_id: 群组标识符 - group_name: 群组显示名称 - statistics: 群组统计 - topics: 话题列表 - user_titles: 用户称号列表 - golden_quotes: 金句列表 - - Returns: - 报告数据字典 - """ - return { - "group_id": group_id, - "group_name": group_name, - "timestamp": datetime.now().isoformat(), - "statistics": statistics.to_dict(), - "topics": [t.to_dict() for t in topics], - "user_titles": [u.to_dict() for u in user_titles], - "golden_quotes": [q.to_dict() for q in golden_quotes], - } diff --git a/src/application/scheduling_service.py b/src/application/scheduling_service.py deleted file mode 100644 index 03fb217..0000000 --- a/src/application/scheduling_service.py +++ /dev/null @@ -1,263 +0,0 @@ -""" -调度服务 - 计划分析的应用服务 - -该服务管理计划的分析任务并与 -分析编排器协调。 -""" - -import asyncio -from collections.abc import Callable -from datetime import datetime, timedelta -from typing import Any - -from ..infrastructure.config import ConfigManager -from ..utils.logger import logger - - -class ScheduledTask: - """表示一个计划的分析任务。""" - - def __init__( - self, - task_id: str, - group_id: str, - scheduled_time: str, # HH:MM 格式 - callback: Callable, - enabled: bool = True, - ): - self.task_id = task_id - self.group_id = group_id - self.scheduled_time = scheduled_time - self.callback = callback - self.enabled = enabled - self.last_run: datetime | None = None - self.next_run: datetime | None = None - self._calculate_next_run() - - def _calculate_next_run(self) -> None: - """计算下一次运行时间。""" - if not self.enabled: - self.next_run = None - return - - try: - hours, minutes = map(int, self.scheduled_time.split(":")) - now = datetime.now() - next_run = now.replace(hour=hours, minute=minutes, second=0, microsecond=0) - - # 如果今天的时间已过，计划明天运行 - if next_run <= now: - next_run += timedelta(days=1) - - self.next_run = next_run - except ValueError: - logger.error(f"无效的计划时间格式: {self.scheduled_time}") - self.next_run = None - - def should_run(self) -> bool: - """检查任务现在是否应该运行。""" - if not self.enabled or not self.next_run: - return False - - now = datetime.now() - - # 检查我们是否在执行窗口内（5分钟容差） - if self.next_run <= now <= self.next_run + timedelta(minutes=5): - # 检查我们今天是否还没有运行 - if self.last_run is None or self.last_run.date() != now.date(): - return True - - return False - - def mark_completed(self) -> None: - """将任务标记为完成并计划下一次运行。""" - self.last_run = datetime.now() - self._calculate_next_run() - - -class SchedulingService: - """ - 管理计划分析任务的应用服务。 - - 该服务运行一个后台循环，检查并 - 执行计划的任务。 - """ - - def __init__(self, config: ConfigManager): - """ - 初始化调度服务。 - - Args: - config: 配置管理器 - """ - self.config = config - self._tasks: dict[str, ScheduledTask] = {} - self._running = False - self._task: asyncio.Task | None = None - self._callbacks: dict[str, Callable] = {} - - def register_callback(self, name: str, callback: Callable) -> None: - """ - 为计划任务注册回调。 - - Args: - name: 回调名称 - callback: 异步回调函数 - """ - self._callbacks[name] = callback - - def add_task( - self, - group_id: str, - scheduled_time: str | None = None, - callback_name: str = "analyze", - ) -> str: - """ - 为群组添加计划任务。 - - Args: - group_id: 群组标识符 - scheduled_time: HH:MM 格式的时间（如果未提供，则使用配置默认值） - callback_name: 要使用的注册回调的名称 - - Returns: - 任务 ID - """ - scheduled_time = scheduled_time or self.config.get_analysis_time() - task_id = f"task_{group_id}" - - callback = self._callbacks.get(callback_name) - if not callback: - logger.warning(f"回调 '{callback_name}' 未注册") - return task_id - - task = ScheduledTask( - task_id=task_id, - group_id=group_id, - scheduled_time=scheduled_time, - callback=callback, - enabled=True, - ) - - self._tasks[task_id] = task - logger.info(f"为 {scheduled_time} 添加了计划任务 {task_id}") - - return task_id - - def remove_task(self, task_id: str) -> bool: - """ - 移除计划任务。 - - Args: - task_id: 任务标识符 - - Returns: - 如果任务被移除则返回 True - """ - if task_id in self._tasks: - del self._tasks[task_id] - logger.info(f"移除了计划任务 {task_id}") - return True - return False - - def enable_task(self, task_id: str) -> bool: - """启用计划任务。""" - if task_id in self._tasks: - self._tasks[task_id].enabled = True - self._tasks[task_id]._calculate_next_run() - return True - return False - - def disable_task(self, task_id: str) -> bool: - """禁用计划任务。""" - if task_id in self._tasks: - self._tasks[task_id].enabled = False - self._tasks[task_id].next_run = None - return True - return False - - def get_task_status(self, task_id: str) -> dict[str, Any] | None: - """ - 获取计划任务的状态。 - - Args: - task_id: 任务标识符 - - Returns: - 任务状态字典或 None - """ - task = self._tasks.get(task_id) - if not task: - return None - - return { - "task_id": task.task_id, - "group_id": task.group_id, - "scheduled_time": task.scheduled_time, - "enabled": task.enabled, - "last_run": task.last_run.isoformat() if task.last_run else None, - "next_run": task.next_run.isoformat() if task.next_run else None, - } - - def list_tasks(self) -> list[dict[str, Any]]: - """列出所有计划任务。""" - return [self.get_task_status(tid) for tid in self._tasks.keys()] - - async def start(self) -> None: - """启动调度服务。""" - if self._running: - return - - self._running = True - self._task = asyncio.create_task(self._run_loop()) - logger.info("调度服务已启动") - - async def stop(self) -> None: - """停止调度服务。""" - self._running = False - if self._task: - self._task.cancel() - try: - await self._task - except asyncio.CancelledError: - pass - logger.info("调度服务已停止") - - async def _run_loop(self) -> None: - """主调度循环。""" - while self._running: - try: - await self._check_and_run_tasks() - # 每分钟检查一次 - await asyncio.sleep(60) - except asyncio.CancelledError: - break - except Exception as e: - logger.error(f"调度循环出错: {e}") - await asyncio.sleep(60) - - async def _check_and_run_tasks(self) -> None: - """检查并执行到期任务。""" - for task in list(self._tasks.values()): - if task.should_run(): - try: - logger.info(f"正在执行计划任务 {task.task_id}") - await task.callback(task.group_id) - task.mark_completed() - logger.info(f"计划任务 {task.task_id} 已完成") - except Exception as e: - logger.error(f"执行任务 {task.task_id} 失败: {e}") - - def setup_from_config(self) -> None: - """根据配置设置计划任务。""" - if not self.config.get_auto_analysis_enabled(): - logger.info("自动分析已禁用") - return - - enabled_groups = self.config.get_enabled_groups() - analysis_time = self.config.get_analysis_time() - - for group_id in enabled_groups: - self.add_task(group_id, analysis_time) - - logger.info(f"设置了 {len(enabled_groups)} 个计划任务") diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index fa264d9..f512faf 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -7,12 +7,12 @@ import asyncio from typing import Any +from ...domain.models.data_models import TokenUsage +from ...domain.repositories.analysis_repository import IAnalysisProvider +from ...domain.repositories.report_repository import IReportGenerator +from ...domain.services.analysis_domain_service import AnalysisDomainService +from ...domain.services.statistics_service import StatisticsService from ...utils.logger import logger -from ..domain.models.data_models import TokenUsage -from ..domain.repositories.analysis_repository import IAnalysisProvider -from ..domain.repositories.report_repository import IReportGenerator -from ..domain.services.analysis_domain_service import AnalysisDomainService -from ..domain.services.statistics_service import StatisticsService class AnalysisApplicationService: diff --git a/src/infrastructure/__init__.py b/src/infrastructure/__init__.py index e49437c..2af8900 100644 --- a/src/infrastructure/__init__.py +++ b/src/infrastructure/__init__.py @@ -1,29 +1 @@ # 基础设施层 -# 持久化 -# LLM -# 配置 -# 弹性/容错 -from . import config, llm, persistence, platform, resilience - -__all__ = [ - "config", - "llm", - "persistence", - "platform", - "resilience", - # 平台 - "PlatformAdapter", - "PlatformAdapterFactory", - "OneBotAdapter", - # 持久化 - "HistoryRepository", - # LLM - "LLMClient", - # 配置 - "ConfigManager", - # 弹性 - "CircuitBreaker", - "RateLimiter", - "retry_async", - "RetryConfig", -] diff --git a/src/infrastructure/analysis/utils/llm_utils.py b/src/infrastructure/analysis/utils/llm_utils.py index 83901c8..cd9d66f 100644 --- a/src/infrastructure/analysis/utils/llm_utils.py +++ b/src/infrastructure/analysis/utils/llm_utils.py @@ -6,8 +6,8 @@ LLM API请求处理工具模块 import asyncio from typing import Any -from ...utils.logger import logger -from ...utils.resilience import CircuitBreaker, global_llm_rate_limiter +from ....utils.logger import logger +from ....utils.resilience import CircuitBreaker, global_llm_rate_limiter _circuit_breakers = {} diff --git a/src/infrastructure/llm/__init__.py b/src/infrastructure/llm/__init__.py deleted file mode 100644 index 845fc7a..0000000 --- a/src/infrastructure/llm/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -""" -LLM Module - LLM client implementations -""" - -from .llm_client import LLMClient - -__all__ = ["LLMClient"] diff --git a/src/infrastructure/llm/llm_client.py b/src/infrastructure/llm/llm_client.py deleted file mode 100644 index 857c01f..0000000 --- a/src/infrastructure/llm/llm_client.py +++ /dev/null @@ -1,187 +0,0 @@ -""" -LLM 客户端 - 包装 AstrBot 的 LLM 提供商系统 - -该模块提供了一个访问 AstrBot LLM 功能的清晰接口， -抽象了提供商管理的细节。 -""" - -from typing import Any - -from ...domain.exceptions import LLMException, LLMRateLimitException -from ...domain.value_objects.statistics import TokenUsage -from ...utils.logger import logger - - -class LLMClient: - """ - 用于与 LLM 提供商交互的客户端。 - - 该类包装了 AstrBot 的提供商系统，并提供了一个 - 清晰的接口来进行 LLM 调用。 - """ - - def __init__(self, context: Any): - """ - 初始化 LLM 客户端。 - - Args: - context: 具有提供商访问权限的 AstrBot 插件上下文 - """ - self.context = context - self._provider_cache: dict[str, Any] = {} - - def get_provider(self, provider_id: str | None = None) -> Any: - """ - 通过 ID 获取 LLM 提供商。 - - Args: - provider_id: 特定的提供商 ID，None 表示默认 - - Returns: - 提供商实例 - - Raises: - LLMException: 如果未找到提供商 - """ - try: - if provider_id and provider_id in self._provider_cache: - return self._provider_cache[provider_id] - - if provider_id: - provider = self.context.get_provider_by_id(provider_id) - else: - # 获取默认提供商 - providers = self.context.get_all_providers() - if not providers: - raise LLMException("无可用 LLM 提供商") - provider = providers[0] - - if provider: - self._provider_cache[provider_id or "default"] = provider - - return provider - - except Exception as e: - raise LLMException(f"获取提供商失败: {e}") - - async def chat_completion( - self, - prompt: str, - provider_id: str | None = None, - max_tokens: int = 2000, - temperature: float = 0.7, - system_prompt: str | None = None, - ) -> tuple[str, TokenUsage]: - """ - 发起聊天完成请求。 - - Args: - prompt: 用户提示词 - provider_id: 特定的提供商 ID (可选) - max_tokens: 响应中的最大 token 数 - temperature: 采样温度 - system_prompt: 可选的系统提示词 - - Returns: - (response_text, token_usage) 元组 - - Raises: - LLMException: 如果请求失败 - """ - try: - provider = self.get_provider(provider_id) - if not provider: - raise LLMException("无可用提供商", provider_id or "default") - - # 构建消息 - messages = [] - if system_prompt: - messages.append({"role": "system", "content": system_prompt}) - messages.append({"role": "user", "content": prompt}) - - # 发起请求 - response = await provider.text_chat( - messages=messages, - session_id=None, # 无状态 - ) - - # 提取响应文本 - if hasattr(response, "completion_text"): - response_text = response.completion_text - elif isinstance(response, dict): - response_text = response.get( - "completion_text", response.get("text", "") - ) - else: - response_text = str(response) - - # 提取 token 使用情况 - token_usage = TokenUsage() - if hasattr(response, "usage"): - usage = response.usage - if hasattr(usage, "prompt_tokens"): - token_usage = TokenUsage( - prompt_tokens=usage.prompt_tokens or 0, - completion_tokens=usage.completion_tokens or 0, - total_tokens=usage.total_tokens or 0, - ) - - return response_text, token_usage - - except Exception as e: - error_msg = str(e).lower() - if "rate limit" in error_msg or "429" in error_msg: - raise LLMRateLimitException(str(e), provider_id or "default") - raise LLMException(f"聊天完成请求失败: {e}", provider_id or "default") - - async def analyze_with_json_output( - self, - prompt: str, - provider_id: str | None = None, - max_tokens: int = 2000, - temperature: float = 0.7, - ) -> tuple[str, TokenUsage]: - """ - 发起期望 JSON 输出的完成请求。 - - Args: - prompt: 分析提示词 - provider_id: 特定的提供商 ID (可选) - max_tokens: 响应中的最大 token 数 - temperature: 采样温度 - - Returns: - (response_text, token_usage) 元组 - """ - # 如果提示词中没有 JSON 指令，则添加 - json_instruction = "\nRespond with valid JSON only." - if "json" not in prompt.lower(): - prompt = prompt + json_instruction - - return await self.chat_completion( - prompt=prompt, - provider_id=provider_id, - max_tokens=max_tokens, - temperature=temperature, - ) - - def list_available_providers(self) -> list[dict[str, str]]: - """ - 列出所有可用的 LLM 提供商。 - - Returns: - 提供商信息字典列表 - """ - try: - providers = self.context.get_all_providers() - return [ - { - "id": getattr(p, "id", str(i)), - "name": getattr(p, "name", f"Provider {i}"), - "type": getattr(p, "type", "unknown"), - } - for i, p in enumerate(providers) - ] - except Exception as e: - logger.error(f"列出提供商失败: {e}") - return [] diff --git a/src/infrastructure/messaging/message_sender.py b/src/infrastructure/messaging/message_sender.py new file mode 100644 index 0000000..d1fa793 --- /dev/null +++ b/src/infrastructure/messaging/message_sender.py @@ -0,0 +1,53 @@ +""" +消息发送器 - 基础设施层 +提供高层消息发送接口，支持跨平台智能路由。 +""" + +from ...utils.logger import logger + + +class MessageSender: + """ + 消息发送器 + 封装了 PlatformAdapter 的底层调用，提供更高层的发送接口 + """ + + def __init__(self, bot_manager, config_manager, retry_manager): + self.bot_manager = bot_manager + self.config_manager = config_manager + self.retry_manager = retry_manager + + async def send_text( + self, group_id: str, text: str, platform_id: str = None + ) -> bool: + """发送文本消息""" + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.error(f"[MessageSender] 未找到平台 {platform_id} 的适配器") + return False + return await adapter.send_text(group_id, text) + + async def send_image_smart( + self, group_id: str, image_url: str, caption: str = "", platform_id: str = None + ) -> bool: + """智能发送图片，支持自动选择适配器""" + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.error(f"[MessageSender] 未找到平台 {platform_id} 的适配器") + return False + return await adapter.send_image(group_id, image_url, caption) + + async def send_pdf( + self, group_id: str, pdf_path: str, caption: str = "", platform_id: str = None + ) -> bool: + """发送 PDF 文件""" + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.error(f"[MessageSender] 未找到平台 {platform_id} 的适配器") + return False + return await adapter.send_file(group_id, pdf_path) + + def _get_available_platforms(self, group_id: str): + """获取可用的平台列表 (Helper for Dispatcher)""" + # 简单实现：返回所有已加载的平台 + return [(pid, None) for pid in self.bot_manager.get_platform_ids()] diff --git a/src/infrastructure/reporting/dispatcher.py b/src/infrastructure/reporting/dispatcher.py index 53cf114..b44948f 100644 --- a/src/infrastructure/reporting/dispatcher.py +++ b/src/infrastructure/reporting/dispatcher.py @@ -1,8 +1,8 @@ from collections.abc import Callable from typing import Any -from ..utils.logger import logger -from ..utils.trace_context import TraceContext +from ...utils.logger import logger +from ...utils.trace_context import TraceContext class ReportDispatcher: diff --git a/src/infrastructure/resilience/__init__.py b/src/infrastructure/resilience/__init__.py deleted file mode 100644 index fc18a69..0000000 --- a/src/infrastructure/resilience/__init__.py +++ /dev/null @@ -1,15 +0,0 @@ -""" -弹性模块 - 断路器、速率限制器和重试工具 -""" - -from .circuit_breaker import CircuitBreaker, CircuitState -from .rate_limiter import RateLimiter -from .retry import RetryConfig, retry_async - -__all__ = [ - "CircuitBreaker", - "CircuitState", - "RateLimiter", - "retry_async", - "RetryConfig", -] diff --git a/src/infrastructure/resilience/circuit_breaker.py b/src/infrastructure/resilience/circuit_breaker.py deleted file mode 100644 index ca19599..0000000 --- a/src/infrastructure/resilience/circuit_breaker.py +++ /dev/null @@ -1,137 +0,0 @@ -""" -断路器 - 防止级联故障 - -实现断路器模式，防止对失败服务的重复调用。 -""" - -import time -from collections.abc import Callable -from dataclasses import dataclass, field -from enum import Enum - -from ...utils.logger import logger - - -class CircuitState(Enum): - """断路器状态。""" - - CLOSED = "closed" # 正常运行 - OPEN = "open" # 故障中，拒绝调用 - HALF_OPEN = "half_open" # 测试服务是否恢复 - - -@dataclass -class CircuitBreaker: - """ - 断路器实现。 - - 通过跟踪故障率并临时阻止对故障服务的调用 - 来防止级联故障。 - """ - - name: str - failure_threshold: int = 5 - recovery_timeout: float = 30.0 - half_open_max_calls: int = 3 - - # 内部状态 - _state: CircuitState = field(default=CircuitState.CLOSED, init=False) - _failure_count: int = field(default=0, init=False) - _success_count: int = field(default=0, init=False) - _last_failure_time: float = field(default=0, init=False) - _half_open_calls: int = field(default=0, init=False) - - @property - def state(self) -> CircuitState: - """获取当前断路器状态，检查是否恢复。""" - if self._state == CircuitState.OPEN: - if time.time() - self._last_failure_time >= self.recovery_timeout: - self._transition_to(CircuitState.HALF_OPEN) - return self._state - - def _transition_to(self, new_state: CircuitState) -> None: - """转换到新状态。""" - old_state = self._state - self._state = new_state - - if new_state == CircuitState.CLOSED: - self._failure_count = 0 - self._success_count = 0 - elif new_state == CircuitState.HALF_OPEN: - self._half_open_calls = 0 - - logger.debug(f"断路器 {self.name}: {old_state.value} -> {new_state.value}") - - def record_success(self) -> None: - """记录成功调用。""" - if self._state == CircuitState.HALF_OPEN: - self._success_count += 1 - if self._success_count >= self.half_open_max_calls: - self._transition_to(CircuitState.CLOSED) - elif self._state == CircuitState.CLOSED: - # 成功时重置故障计数 - self._failure_count = 0 - - def record_failure(self) -> None: - """记录失败调用。""" - self._failure_count += 1 - self._last_failure_time = time.time() - - if self._state == CircuitState.HALF_OPEN: - self._transition_to(CircuitState.OPEN) - elif self._state == CircuitState.CLOSED: - if self._failure_count >= self.failure_threshold: - self._transition_to(CircuitState.OPEN) - - def can_execute(self) -> bool: - """检查是否可以执行调用。""" - state = self.state # 这可能触发状态转换 - - if state == CircuitState.CLOSED: - return True - elif state == CircuitState.OPEN: - return False - elif state == CircuitState.HALF_OPEN: - self._half_open_calls += 1 - return self._half_open_calls <= self.half_open_max_calls - - return False - - def reset(self) -> None: - """重置断路器到关闭状态。""" - self._transition_to(CircuitState.CLOSED) - - async def execute( - self, - func: Callable, - *args, - fallback: Callable | None = None, - **kwargs, - ): - """ - 使用断路器保护执行函数。 - - 参数: - func: 要执行的异步函数 - *args: 函数参数 - fallback: 断路器打开时的可选降级函数 - **kwargs: 函数关键字参数 - - 返回: - 函数结果或降级结果 - - 异常: - Exception: 如果断路器打开且没有提供降级函数 - """ - if not self.can_execute(): - if fallback: - return await fallback(*args, **kwargs) - raise Exception(f"断路器 {self.name} 已打开") - - try: - result = await func(*args, **kwargs) - self.record_success() - return result - except Exception: - self.record_failure() - raise diff --git a/src/infrastructure/resilience/rate_limiter.py b/src/infrastructure/resilience/rate_limiter.py deleted file mode 100644 index f56e93b..0000000 --- a/src/infrastructure/resilience/rate_limiter.py +++ /dev/null @@ -1,139 +0,0 @@ -""" -速率限制器 - 控制请求速率 - -实现令牌桶速率限制，防止服务过载。 -""" - -import asyncio -import time -from dataclasses import dataclass, field - - -@dataclass -class RateLimiter: - """ - 令牌桶速率限制器。 - - 使用令牌桶算法控制操作速率。 - """ - - name: str - rate: float # 每秒令牌数 - burst: int # 最大突发大小（桶容量） - - # 内部状态 - _tokens: float = field(default=0, init=False) - _last_update: float = field(default=0, init=False) - _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) - - def __post_init__(self): - """初始化令牌桶。""" - self._tokens = float(self.burst) - self._last_update = time.time() - - def _refill(self) -> None: - """根据经过的时间补充令牌。""" - now = time.time() - elapsed = now - self._last_update - self._tokens = min(self.burst, self._tokens + elapsed * self.rate) - self._last_update = now - - async def acquire(self, tokens: int = 1, timeout: float | None = None) -> bool: - """ - 从桶中获取令牌。 - - 参数: - tokens: 要获取的令牌数 - timeout: 最大等待时间（None = 无限等待） - - 返回: - 如果获取到令牌返回 True，超时返回 False - """ - start_time = time.time() - - async with self._lock: - while True: - self._refill() - - if self._tokens >= tokens: - self._tokens -= tokens - return True - - if timeout is not None: - elapsed = time.time() - start_time - if elapsed >= timeout: - return False - - # 计算获取足够令牌的等待时间 - tokens_needed = tokens - self._tokens - wait_time = tokens_needed / self.rate - - if timeout is not None: - remaining = timeout - (time.time() - start_time) - wait_time = min(wait_time, remaining) - - if wait_time > 0: - await asyncio.sleep(wait_time) - - def try_acquire(self, tokens: int = 1) -> bool: - """ - 尝试获取令牌而不等待。 - - 参数: - tokens: 要获取的令牌数 - - 返回: - 如果获取到令牌返回 True，否则返回 False - """ - self._refill() - - if self._tokens >= tokens: - self._tokens -= tokens - return True - return False - - @property - def available_tokens(self) -> float: - """获取当前可用令牌数。""" - self._refill() - return self._tokens - - def reset(self) -> None: - """重置速率限制器到满容量。""" - self._tokens = float(self.burst) - self._last_update = time.time() - - -class RateLimiterGroup: - """ - 不同操作的速率限制器组。 - """ - - def __init__(self): - self._limiters: dict[str, RateLimiter] = {} - - def get_or_create( - self, - name: str, - rate: float = 1.0, - burst: int = 5, - ) -> RateLimiter: - """ - 获取或创建速率限制器。 - - 参数: - name: 限制器名称 - rate: 每秒令牌数 - burst: 最大突发大小 - - 返回: - RateLimiter 实例 - """ - if name not in self._limiters: - self._limiters[name] = RateLimiter(name=name, rate=rate, burst=burst) - return self._limiters[name] - - def reset_all(self) -> None: - """重置所有速率限制器。""" - for limiter in self._limiters.values(): - limiter.reset() diff --git a/src/infrastructure/resilience/retry.py b/src/infrastructure/resilience/retry.py deleted file mode 100644 index c1c7933..0000000 --- a/src/infrastructure/resilience/retry.py +++ /dev/null @@ -1,176 +0,0 @@ -""" -重试 - 带指数退避的重试工具 - -提供用于处理瞬态故障的重试装饰器和工具。 -""" - -import asyncio -import random -from collections.abc import Callable -from dataclasses import dataclass -from functools import wraps - -from ...utils.logger import logger - - -@dataclass -class RetryConfig: - """重试行为配置。""" - - max_attempts: int = 3 - base_delay: float = 1.0 - max_delay: float = 60.0 - exponential_base: float = 2.0 - jitter: bool = True - retry_exceptions: tuple[type[Exception], ...] = (Exception,) - - -def calculate_delay( - attempt: int, - base_delay: float, - max_delay: float, - exponential_base: float, - jitter: bool, -) -> float: - """ - 计算重试尝试的延迟。 - - 参数: - attempt: 当前尝试次数（从 0 开始） - base_delay: 基础延迟（秒） - max_delay: 最大延迟（秒） - exponential_base: 指数退避的基数 - jitter: 是否添加随机抖动 - - 返回: - 延迟时间（秒） - """ - delay = base_delay * (exponential_base**attempt) - delay = min(delay, max_delay) - - if jitter: - delay = delay * (0.5 + random.random()) - - return delay - - -def retry_async( - max_attempts: int = 3, - base_delay: float = 1.0, - max_delay: float = 60.0, - exponential_base: float = 2.0, - jitter: bool = True, - retry_exceptions: tuple[type[Exception], ...] = (Exception,), - on_retry: Callable[[Exception, int], None] | None = None, -): - """ - 带指数退避的异步函数重试装饰器。 - - 参数: - max_attempts: 最大尝试次数 - base_delay: 重试之间的基础延迟 - max_delay: 重试之间的最大延迟 - exponential_base: 指数退避的基数 - jitter: 是否添加随机抖动 - retry_exceptions: 要重试的异常元组 - on_retry: 重试时的可选回调（异常，尝试次数） - - 返回: - 装饰后的函数 - """ - - def decorator(func: Callable): - @wraps(func) - async def wrapper(*args, **kwargs): - last_exception = None - - for attempt in range(max_attempts): - try: - return await func(*args, **kwargs) - except retry_exceptions as e: - last_exception = e - - if attempt < max_attempts - 1: - delay = calculate_delay( - attempt, base_delay, max_delay, exponential_base, jitter - ) - - if on_retry: - on_retry(e, attempt + 1) - - logger.debug( - f"重试 {attempt + 1}/{max_attempts} {func.__name__} " - f"延迟 {delay:.2f}s: {e}" - ) - await asyncio.sleep(delay) - else: - logger.warning( - f"{func.__name__} 的所有 {max_attempts} 次尝试均失败: {e}" - ) - - raise last_exception - - return wrapper - - return decorator - - -class RetryExecutor: - """ - 带重试逻辑的函数执行器。 - """ - - def __init__(self, config: RetryConfig | None = None): - """ - 初始化重试执行器。 - - 参数: - config: 重试配置 - """ - self.config = config or RetryConfig() - - async def execute( - self, - func: Callable, - *args, - config: RetryConfig | None = None, - **kwargs, - ): - """ - 使用重试逻辑执行函数。 - - 参数: - func: 要执行的异步函数 - *args: 函数参数 - config: 可选的覆盖配置 - **kwargs: 函数关键字参数 - - 返回: - 函数结果 - - 异常: - Exception: 如果所有重试都失败 - """ - cfg = config or self.config - last_exception = None - - for attempt in range(cfg.max_attempts): - try: - return await func(*args, **kwargs) - except cfg.retry_exceptions as e: - last_exception = e - - if attempt < cfg.max_attempts - 1: - delay = calculate_delay( - attempt, - cfg.base_delay, - cfg.max_delay, - cfg.exponential_base, - cfg.jitter, - ) - logger.debug( - f"重试 {attempt + 1}/{cfg.max_attempts} 延迟 {delay:.2f}s: {e}" - ) - await asyncio.sleep(delay) - - raise last_exception diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index c10514a..3b6c317 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -9,10 +9,10 @@ import weakref from apscheduler.triggers.cron import CronTrigger from ...utils.logger import logger -from ..core.message_sender import MessageSender +from ...utils.trace_context import TraceContext +from ..messaging.message_sender import MessageSender from ..platform.factory import PlatformAdapterFactory from ..reporting.dispatcher import ReportDispatcher -from ..utils.trace_context import TraceContext class AutoScheduler: diff --git a/src/utils/__init__.py b/src/utils/__init__.py index 9c47efd..58c2f69 100644 --- a/src/utils/__init__.py +++ b/src/utils/__init__.py @@ -3,7 +3,6 @@ 包含PDF处理和通用工具函数 """ -from .helpers import MessageAnalyzer from .pdf_utils import PDFInstaller -__all__ = ["PDFInstaller", "MessageAnalyzer"] +__all__ = ["PDFInstaller"]