fix: 正确的导入各重构模块

2026-09-22 13:38:43 +00:00 · 2026-02-09 13:08:23 +08:00
parent 2005f0633f
commit e1ad6c9fc9
18 changed files with 66 additions and 1461 deletions
@@ -15,5 +15,3 @@ QQ群日常分析插件 - 源代码包
 - utils: 工具函数
 - visualization: 可视化组件
 """
 __version__ = "2.0.0"
@@ -1,9 +1 @@
-from .message_converter import MessageConverter
+# 应用层 - 编排和用例
 from .reporting_service import ReportingService
 from .scheduling_service import SchedulingService
 __all__ = [
    "MessageConverter",
    "SchedulingService",
    "ReportingService",
 ]
@@ -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)
@@ -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],
        }
@@ -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)} 个计划任务")
@@ -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:
@@ -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",
 ]
@@ -6,8 +6,8 @@ LLM API请求处理工具模块
 import asyncio
 from typing import Any
-from ...utils.logger import logger
+from ....utils.logger import logger
-from ...utils.resilience import CircuitBreaker, global_llm_rate_limiter
+from ....utils.resilience import CircuitBreaker, global_llm_rate_limiter
 _circuit_breakers = {}
@@ -1,7 +0,0 @@
 """
 LLM Module - LLM client implementations
 """
 from .llm_client import LLMClient
 __all__ = ["LLMClient"]
@@ -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 []
@@ -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()]
@@ -1,8 +1,8 @@
 from collections.abc import Callable
 from typing import Any
-from ..utils.logger import logger
+from ...utils.logger import logger
-from ..utils.trace_context import TraceContext
+from ...utils.trace_context import TraceContext
 class ReportDispatcher:
@@ -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",
 ]
@@ -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
@@ -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()
@@ -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
@@ -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:
@@ -3,7 +3,6 @@
 包含PDF处理和通用工具函数
 """
 from .helpers import MessageAnalyzer
 from .pdf_utils import PDFInstaller
-__all__ = ["PDFInstaller", "MessageAnalyzer"]
+__all__ = ["PDFInstaller"]