nezumi/astrbot_plugin_qq_group_daily_analysis
1
0
Fork 0
mirror of https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git synced 2026-09-22 13:38:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: code structure for improved readability and scalability

Browse Source
This commit is contained in:
SXP-Simon
2026-02-08 14:01:18 +08:00
parent 9534d4ea68
commit a3c6ea737a
5 changed files with 4507 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/07_ddd_refactoring_analysis.md
+698
View File
@@ -0,0 +1,698 @@
# 07. DDD 重构分析报告 (Domain-Driven Design Refactoring Analysis)
> **分析日期**: 2026-02-08
> **分析范围**: 插件完整代码库 + 现有文档 (01-06)
> **分析目的**: 基于 DDD 范式，结合现有代码现状和 06_review.md 的务实建议，制定可落地的重构方案
---
## 1. 执行摘要 (Executive Summary)
### 1.1 当前状态评估
经过对代码库的深度分析，本插件已经完成了一次显著的重构（基于 06_review.md 的建议），当前架构状态：
| 维度 | 状态 | 说明 |
|------|------|------|
| **模块拆分** | ✅ 已完成 | AutoScheduler 已从 1000+ 行精简至 517 行 |
| **职责分离** | ✅ 已完成 | MessageSender、ReportDispatcher、BotManager 已独立 |
| **TraceID** | ✅ 已实现 | 使用 contextvars 实现零侵入链路追踪 |
| **熔断器** | ✅ 已实现 | CircuitBreaker + GlobalRateLimiter 已就绪 |
| **框架对齐** | ✅ 已完成 | 使用 Context.cron_manager + OnPlatformLoaded 钩子 |
| **DDD 分层** | ⚠️ 部分 | 有模块划分但未严格遵循 DDD 分层架构 |
### 1.2 核心结论
**本插件已完成 06_review.md 建议的务实重构，代码质量良好。**
进一步的 DDD 重构应聚焦于：
1. **领域边界明确化** - 定义清晰的限界上下文
2. **领域模型增强** - 引入轻量级 DDD 战术模式
3. **依赖方向规范** - 确保依赖指向领域层
4. **可测试性提升** - 通过接口抽象支持单元测试
---
## 2. 现有架构分析
### 2.1 当前目录结构
```
astrbot_plugin_qq_group_daily_analysis/
├── main.py # 插件入口 (Application Layer - Controller)
├── src/
│ ├── core/ # 核心模块
│ │ ├── bot_manager.py # Bot 实例管理 (Infrastructure)
│ │ ├── config.py # 配置管理 (Infrastructure)
│ │ ├── history_manager.py # 历史记录管理 (Infrastructure)
│ │ ├── message_handler.py # 消息处理 (Application Service)
│ │ └── message_sender.py # 消息发送 (Infrastructure)
│ ├── scheduler/ # 调度模块
│ │ ├── auto_scheduler.py # 自动调度器 (Application Service)
│ │ └── retry.py # 重试管理器 (Infrastructure)
│ ├── analysis/ # 分析模块
│ │ ├── llm_analyzer.py # LLM 分析协调器 (Application Service)
│ │ ├── analyzers/ # 具体分析器 (Domain Services)
│ │ │ ├── base_analyzer.py
│ │ │ ├── topic_analyzer.py
│ │ │ ├── user_title_analyzer.py
│ │ │ └── golden_quote_analyzer.py
│ │ └── utils/ # 分析工具
│ │ ├── json_utils.py
│ │ └── llm_utils.py
│ ├── models/ # 数据模型
│ │ └── data_models.py # 数据结构定义 (Domain Models)
│ ├── reports/ # 报告模块
│ │ ├── dispatcher.py # 报告分发器 (Application Service)
│ │ ├── generators.py # 报告生成器 (Domain Service)
│ │ └── templates/ # 报告模板 (Infrastructure)
│ ├── visualization/ # 可视化模块
│ │ └── activity_charts.py # 活跃度图表 (Domain Service)
│ └── utils/ # 工具模块
│ ├── helpers.py # 辅助函数
│ ├── pdf_utils.py # PDF 工具 (Infrastructure)
│ ├── resilience.py # 熔断器/限流器 (Infrastructure)
│ └── trace_context.py # 链路追踪 (Infrastructure)
└── docs/ # 文档
```
### 2.2 当前架构优点
| 优点 | 体现 |
|------|------|
| **模块化清晰** | 按功能划分目录：core、scheduler、analysis、reports、visualization |
| **职责单一** | MessageSender 只负责发送，ReportDispatcher 只负责分发 |
| **可观测性** | TraceContext + TraceLogFilter 实现全链路追踪 |
| **弹性设计** | CircuitBreaker + GlobalRateLimiter + RetryManager |
| **框架对齐** | 使用 AstrBot 的 cron_manager、platform_manager、put_kv_data |
| **并发控制** | asyncio.Semaphore 控制并发，asyncio.gather 并行执行 |
### 2.3 当前架构不足 (DDD 视角)
| 不足 | 说明 | 影响 |
|------|------|------|
| **领域边界模糊** | 没有明确的限界上下文定义 | 模块间耦合度难以评估 |
| **贫血模型** | data_models.py 只有数据结构，无行为 | 业务逻辑分散在 Service 中 |
| **依赖方向混乱** | Application 层直接依赖 Infrastructure | 测试困难，替换实现困难 |
| **缺少聚合根** | 没有定义实体的一致性边界 | 状态管理分散 |
| **接口抽象不足** | 直接依赖具体实现类 | Mock 测试困难 |
---
## 3. DDD 重构方案
### 3.1 限界上下文识别 (Bounded Contexts)
基于业务能力分析，本插件包含以下限界上下文：
```
┌─────────────────────────────────────────────────────────────────┐
│ 群聊日常分析插件 (Plugin) │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 调度上下文 │ │ 分析上下文 │ │ 报告上下文 │ │
│ │ (Scheduling) │ │ (Analysis) │ │ (Reporting) │ │
│ │ │ │ │ │ │ │
│ │ - 定时任务 │ │ - 话题分析 │ │ - 报告生成 │ │
│ │ - 并发控制 │ │ - 用户画像 │ │ - 格式转换 │ │
│ │ - 任务编排 │ │ - 金句提取 │ │ - 消息发送 │ │
│ │ │ │ - 统计计算 │ │ - 重试机制 │ │
│ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │ │
│ └────────────────────┼────────────────────┘ │
│ │ │
│ ┌─────────────────────────────┴─────────────────────────────┐ │
│ │ 共享内核 (Shared Kernel) │ │
│ │ - 配置管理 (ConfigManager) │ │
│ │ - Bot 管理 (BotManager) │ │
│ │ - 链路追踪 (TraceContext) │ │
│ │ - 弹性组件 (CircuitBreaker, RateLimiter) │ │
│ └─────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
### 3.2 DDD 分层架构设计
```
┌─────────────────────────────────────────────────────────────────┐
│ Interface Layer │
│ (接口层 / 用户界面) │
│ main.py - 命令处理、事件响应、框架集成 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Application Layer │
│ (应用层 / 用例) │
│ - AnalysisOrchestrator: 分析流程编排 │
│ - SchedulingService: 定时任务管理 │
│ - ReportingService: 报告生成与分发 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Domain Layer │
│ (领域层 / 核心业务) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Entities (实体) │ │
│ │ - AnalysisTask: 分析任务实体 │ │
│ │ - GroupAnalysisResult: 群分析结果实体 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Value Objects (值对象) │ │
│ │ - SummaryTopic, UserTitle, GoldenQuote │ │
│ │ - GroupStatistics, TokenUsage │ │
│ │ - AnalysisContext, RetryPolicy │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Domain Services (领域服务) │ │
│ │ - TopicAnalyzer, UserTitleAnalyzer, GoldenQuoteAnalyzer│ │
│ │ - StatisticsCalculator, ActivityVisualizer │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Repository Interfaces (仓储接口) │ │
│ │ - IAnalysisHistoryRepository │ │
│ │ - IMessageRepository │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Infrastructure Layer │
│ (基础设施层 / 实现) │
│ - KVAnalysisHistoryRepository: 使用 AstrBot KV 存储 │
│ - OneBotMessageRepository: 通过 OneBot API 获取消息 │
│ - MessageSender: 消息发送实现 │
│ - LLMClient: LLM API 调用实现 │
│ - ConfigManager: 配置读写实现 │
│ - BotManager: Bot 实例管理实现 │
└─────────────────────────────────────────────────────────────────┘
```
### 3.3 目标目录结构
```
astrbot_plugin_qq_group_daily_analysis/
├── main.py # Interface Layer
├── src/
│ ├── application/ # Application Layer
│ │ ├── __init__.py
│ │ ├── analysis_orchestrator.py # 分析流程编排
│ │ ├── scheduling_service.py # 定时任务服务
│ │ └── reporting_service.py # 报告服务
│ │
│ ├── domain/ # Domain Layer
│ │ ├── __init__.py
│ │ ├── entities/ # 实体
│ │ │ ├── __init__.py
│ │ │ ├── analysis_task.py # 分析任务实体
│ │ │ └── analysis_result.py # 分析结果实体
│ │ ├── value_objects/ # 值对象
│ │ │ ├── __init__.py
│ │ │ ├── topic.py # SummaryTopic
│ │ │ ├── user_title.py # UserTitle
│ │ │ ├── golden_quote.py # GoldenQuote
│ │ │ ├── statistics.py # GroupStatistics, TokenUsage
│ │ │ └── analysis_context.py # AnalysisContext
│ │ ├── services/ # 领域服务
│ │ │ ├── __init__.py
│ │ │ ├── topic_analyzer.py
│ │ │ ├── user_title_analyzer.py
│ │ │ ├── golden_quote_analyzer.py
│ │ │ ├── statistics_calculator.py
│ │ │ └── report_generator.py
│ │ └── repositories/ # 仓储接口
│ │ ├── __init__.py
│ │ ├── analysis_history_repository.py
│ │ └── message_repository.py
│ │
│ ├── infrastructure/ # Infrastructure Layer
│ │ ├── __init__.py
│ │ ├── persistence/ # 持久化实现
│ │ │ ├── __init__.py
│ │ │ └── kv_analysis_history_repository.py
│ │ ├── messaging/ # 消息通信
│ │ │ ├── __init__.py
│ │ │ ├── onebot_message_repository.py
│ │ │ ├── message_sender.py
│ │ │ └── retry_manager.py
│ │ ├── llm/ # LLM 集成
│ │ │ ├── __init__.py
│ │ │ ├── llm_client.py
│ │ │ └── llm_utils.py
│ │ ├── bot/ # Bot 管理
│ │ │ ├── __init__.py
│ │ │ └── bot_manager.py
│ │ ├── config/ # 配置管理
│ │ │ ├── __init__.py
│ │ │ └── config_manager.py
│ │ └── resilience/ # 弹性组件
│ │ ├── __init__.py
│ │ ├── circuit_breaker.py
│ │ ├── rate_limiter.py
│ │ └── trace_context.py
│ │
│ └── shared/ # 共享内核
│ ├── __init__.py
│ └── exceptions.py # 自定义异常
│
├── tests/ # 测试
│ ├── unit/
│ │ ├── domain/
│ │ └── application/
│ └── integration/
│
└── docs/ # 文档
```
---
## 4. 领域模型设计
### 4.1 实体 (Entities)
#### 4.1.1 AnalysisTask (分析任务)
```python
# src/domain/entities/analysis_task.py
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time
import uuid
class TaskStatus(Enum):
PENDING = "pending"
FETCHING_MESSAGES = "fetching_messages"
ANALYZING = "analyzing"
GENERATING_REPORT = "generating_report"
SENDING = "sending"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class AnalysisTask:
"""
分析任务实体 - 聚合根
封装单次群聊分析的完整生命周期
"""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
group_id: str = ""
platform_id: str = ""
trace_id: str = ""
status: TaskStatus = TaskStatus.PENDING
is_manual: bool = False
# 时间戳
created_at: float = field(default_factory=time.time)
started_at: Optional[float] = None
completed_at: Optional[float] = None
# 结果引用
result_id: Optional[str] = None
error_message: Optional[str] = None
# 业务方法
def start(self):
"""开始执行任务"""
if self.status != TaskStatus.PENDING:
raise ValueError(f"Cannot start task in {self.status} status")
self.status = TaskStatus.FETCHING_MESSAGES
self.started_at = time.time()
def advance_to(self, status: TaskStatus):
"""推进任务状态"""
self.status = status
def complete(self, result_id: str):
"""完成任务"""
self.status = TaskStatus.COMPLETED
self.result_id = result_id
self.completed_at = time.time()
def fail(self, error: str):
"""标记失败"""
self.status = TaskStatus.FAILED
self.error_message = error
self.completed_at = time.time()
@property
def duration(self) -> Optional[float]:
"""任务耗时"""
if self.started_at and self.completed_at:
return self.completed_at - self.started_at
return None
```
#### 4.1.2 GroupAnalysisResult (分析结果)
```python
# src/domain/entities/analysis_result.py
from dataclasses import dataclass, field
from typing import List, Optional
import time
@dataclass
class GroupAnalysisResult:
"""
群分析结果实体
聚合所有分析产出
"""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
group_id: str = ""
trace_id: str = ""
# 分析结果
statistics: Optional['GroupStatistics'] = None
topics: List['SummaryTopic'] = field(default_factory=list)
user_titles: List['UserTitle'] = field(default_factory=list)
golden_quotes: List['GoldenQuote'] = field(default_factory=list)
# 元数据
message_count: int = 0
analysis_days: int = 1
created_at: float = field(default_factory=time.time)
# 部分失败记录
partial_failures: List[str] = field(default_factory=list)
def add_partial_failure(self, module: str):
"""记录部分失败"""
if module not in self.partial_failures:
self.partial_failures.append(module)
def is_complete(self) -> bool:
"""检查是否完整"""
return len(self.partial_failures) == 0
def to_dict(self) -> dict:
"""转换为字典（用于报告生成）"""
return {
"statistics": self.statistics,
"topics": self.topics,
"user_titles": self.user_titles,
"golden_quotes": self.golden_quotes,
}
```
### 4.2 值对象 (Value Objects)
当前 `data_models.py` 中的类已经是良好的值对象设计，保持不变：
- `SummaryTopic` - 话题摘要
- `UserTitle` - 用户称号
- `GoldenQuote` - 金句
- `GroupStatistics` - 群统计
- `TokenUsage` - Token 使用量
- `EmojiStatistics` - 表情统计
- `ActivityVisualization` - 活跃度可视化
新增值对象：
```python
# src/domain/value_objects/analysis_context.py
from dataclasses import dataclass
@dataclass(frozen=True)
class AnalysisContext:
"""
分析上下文值对象
封装单次分析的元信息
"""
trace_id: str
group_id: str
platform_id: str
analysis_days: int
is_manual: bool
unified_msg_origin: str = ""
@dataclass(frozen=True)
class RetryPolicy:
"""
重试策略值对象
"""
max_retries: int = 3
base_delay: float = 5.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter_range: tuple = (1.0, 5.0)
```
### 4.3 仓储接口 (Repository Interfaces)
```python
# src/domain/repositories/analysis_history_repository.py
from abc import ABC, abstractmethod
from typing import Optional, List
class IAnalysisHistoryRepository(ABC):
"""分析历史仓储接口"""
@abstractmethod
async def save(self, group_id: str, result: 'GroupAnalysisResult') -> bool:
"""保存分析结果"""
pass
@abstractmethod
async def get(self, group_id: str, date_str: str, time_str: str) -> Optional[dict]:
"""获取指定时间的分析结果"""
pass
@abstractmethod
async def exists(self, group_id: str, date_str: str, time_str: str) -> bool:
"""检查是否存在分析记录"""
pass
@abstractmethod
async def list_by_group(self, group_id: str, limit: int = 10) -> List[dict]:
"""列出群的历史分析"""
pass
# src/domain/repositories/message_repository.py
from abc import ABC, abstractmethod
from typing import List
class IMessageRepository(ABC):
"""消息仓储接口"""
@abstractmethod
async def fetch_messages(
self,
group_id: str,
days: int,
platform_id: str
) -> List[dict]:
"""获取群消息"""
pass
```
---
## 5. 重构路线图
### Phase 0: 准备工作 (1 天)
**目标**: 建立重构基础设施
1. **创建目录结构**
- 按照 3.3 节创建新的目录结构
- 保留原有文件，采用渐进式迁移
2. **定义接口**
- 创建 `IAnalysisHistoryRepository` 接口
- 创建 `IMessageRepository` 接口
- 创建 `IMessageSender` 接口
3. **添加共享异常**
```python
# src/shared/exceptions.py
class AnalysisError(Exception): pass
class MessageFetchError(AnalysisError): pass
class LLMError(AnalysisError): pass
class ReportGenerationError(AnalysisError): pass
class MessageSendError(AnalysisError): pass
```
### Phase 1: 领域层提取 (3-4 天)
**目标**: 将核心业务逻辑迁移到领域层
1. **迁移值对象**
- 将 `data_models.py` 拆分到 `domain/value_objects/`
- 添加 `AnalysisContext` 和 `RetryPolicy`
2. **创建实体**
- 实现 `AnalysisTask` 实体
- 实现 `GroupAnalysisResult` 实体
3. **迁移领域服务**
- 将 `analyzers/` 移动到 `domain/services/`
- 将 `ActivityVisualizer` 移动到 `domain/services/`
- 确保领域服务不依赖基础设施
**验收标准**:
- 领域层代码无外部依赖（除 Python 标准库和 dataclasses）
- 所有领域逻辑可独立测试
### Phase 2: 基础设施层实现 (2-3 天)
**目标**: 实现仓储和外部集成
1. **实现仓储**
- `KVAnalysisHistoryRepository` 实现 `IAnalysisHistoryRepository`
- `OneBotMessageRepository` 实现 `IMessageRepository`
2. **迁移基础设施组件**
- 将 `message_sender.py` 移动到 `infrastructure/messaging/`
- 将 `bot_manager.py` 移动到 `infrastructure/bot/`
- 将 `config.py` 移动到 `infrastructure/config/`
- 将 `resilience.py` 拆分到 `infrastructure/resilience/`
3. **LLM 客户端封装**
- 创建 `LLMClient` 封装 LLM 调用
- 集成熔断器和限流器
**验收标准**:
- 基础设施层实现所有仓储接口
- 可通过 Mock 替换任意基础设施组件
### Phase 3: 应用层重构 (2-3 天)
**目标**: 实现用例编排
1. **创建应用服务**
- `AnalysisOrchestrator`: 编排完整分析流程
- `SchedulingService`: 管理定时任务
- `ReportingService`: 处理报告生成和发送
2. **依赖注入**
- 应用服务通过构造函数注入仓储接口
- 使用接口而非具体实现
3. **重构 AutoScheduler**
- 将业务逻辑委托给 `AnalysisOrchestrator`
- AutoScheduler 仅保留调度职责
**验收标准**:
- 应用层仅依赖领域层接口
- 可为应用服务编写单元测试
### Phase 4: 接口层对接 (1-2 天)
**目标**: 更新入口点
1. **重构 main.py**
- 使用依赖注入组装各层
- 命令处理器委托给应用服务
2. **保持向后兼容**
- 确保所有命令功能不变
- 配置格式保持兼容
**验收标准**:
- 所有现有功能正常工作
- 通过 `ruff check` 和 `ruff format`
### Phase 5: 测试与文档 (2 天)
**目标**: 补充测试和更新文档
1. **单元测试**
- 领域实体测试
- 领域服务测试
- 应用服务测试（Mock 仓储）
2. **集成测试**
- 端到端分析流程测试
3. **文档更新**
- 更新 README
- 更新架构文档
---
## 6. 风险评估与缓解
| 风险 | 可能性 | 影响 | 缓解措施 |
|------|--------|------|----------|
| 重构引入回归 Bug | 中 | 高 | 渐进式迁移，保留原有代码直到验证完成 |
| 过度设计 | 中 | 中 | 遵循 YAGNI，仅实现必要的抽象 |
| 性能下降 | 低 | 中 | 保持异步架构，避免不必要的对象创建 |
| 团队学习曲线 | 中 | 低 | 提供清晰的文档和代码示例 |
| 与 AstrBot 框架冲突 | 低 | 高 | 继续遵循 06_review.md 的框架对齐原则 |
---
## 7. 不建议实施的方案
基于 06_review.md 的分析和当前代码状态，以下方案**不建议**实施：
| 方案 | 原因 |
|------|------|
| 自建 EventBus | AstrBot 已有事件系统，自建会造成重复 |
| 完整 CQRS 模式 | 插件规模不足以支撑 CQRS 复杂度 |
| 独立数据库 | 应使用 AstrBot 的 KV 存储 |
| 复杂的 Saga 模式 | 当前流程足够简单，不需要分布式事务 |
| 领域事件 + 事件溯源 | 过度设计，简单的状态机即可 |
---
## 8. 总结与建议
### 8.1 当前状态
本插件已经完成了一次成功的务实重构，代码质量显著提升：
- ✅ AutoScheduler 精简化
- ✅ MessageSender、ReportDispatcher 独立
- ✅ TraceContext 链路追踪
- ✅ CircuitBreaker + RateLimiter 弹性设计
- ✅ 框架 API 对齐
### 8.2 DDD 重构建议
**推荐程度**: ⭐⭐⭐ (可选，非必须)
当前代码已经足够好用。DDD 重构的主要收益是：
1. **更好的可测试性** - 通过接口抽象支持 Mock
2. **更清晰的边界** - 领域层与基础设施层分离
3. **更好的可扩展性** - 新增分析器更容易
**建议优先级**:
1. **P0 (推荐)**: 定义仓储接口，提升可测试性
2. **P1 (可选)**: 引入 AnalysisTask 实体，统一状态管理
3. **P2 (可选)**: 完整的目录结构重组
### 8.3 下一步行动
如果决定进行 DDD 重构：
1. 从 Phase 0 开始，先建立接口定义
2. 采用渐进式迁移，不要一次性重写
3. 每个 Phase 完成后进行功能验证
4. 保持与 AstrBot 框架的对齐
如果决定维持现状：
1. 当前架构已足够支撑业务需求
2. 可以在需要时逐步引入 DDD 元素
3. 重点关注功能迭代而非架构重构
---
## 附录 A: 术语表
| 术语 | 定义 |
|------|------|
| **限界上下文 (Bounded Context)** | 领域模型的边界，定义了模型的适用范围 |
| **实体 (Entity)** | 具有唯一标识的领域对象，生命周期内身份不变 |
| **值对象 (Value Object)** | 无唯一标识的领域对象，通过属性值定义 |
| **聚合根 (Aggregate Root)** | 聚合的入口点，保证聚合内的一致性 |
| **领域服务 (Domain Service)** | 无状态的领域逻辑，不属于任何实体 |
| **仓储 (Repository)** | 领域对象的持久化抽象 |
| **应用服务 (Application Service)** | 用例编排，协调领域对象完成业务流程 |
## 附录 B: 参考资料
1. Eric Evans - *Domain-Driven Design: Tackling Complexity in the Heart of Software*
2. Vaughn Vernon - *Implementing Domain-Driven Design*
3. 06_review.md - 本项目的务实重构审查报告
4. AstrBot 官方文档 - https://astrbot.app/
docs/08_cross_platform_decoupling_analysis.md
+968
View File
@@ -0,0 +1,968 @@
# 08. 跨平台解耦调研报告 (Cross-Platform Decoupling Analysis)
> **调研日期**: 2026-02-08
> **调研范围**: AstrBot 平台抽象层 + 插件 QQ 硬编码分析
> **调研目的**: 分析如何将插件从 QQ 专属改造为跨平台通用插件
---
## 1. 执行摘要
### 1.1 当前问题
本插件 (`astrbot_plugin_qq_group_daily_analysis`) 当前存在严重的平台耦合问题：
| 问题类型 | 数量 | 影响 |
|----------|------|------|
| **直接导入 aiocqhttp** | 1 处 | 插件无法在非 QQ 平台加载 |
| **AiocqhttpMessageEvent 类型检查** | 12+ 处 | 所有命令仅限 QQ 平台 |
| **OneBot API 调用 (call_action)** | 8+ 处 | 消息获取/发送依赖 OneBot |
| **QQ 特定数据结构** | 15+ 处 | 消息格式、表情类型等 |
| **QQ 号相关逻辑** | 10+ 处 | bot_qq_id、self_id 等 |
### 1.2 核心发现
**AstrBot 已提供完善的跨平台抽象层**，支持 12+ 个平台：
| 平台 | 适配器 | 消息历史支持 |
|------|--------|--------------|
| QQ (OneBot v11) | `aiocqhttp` | ✅ `get_group_msg_history` |
| QQ 官方 | `qqofficial` | ❌ 不支持 |
| Telegram | `telegram` | ✅ 可通过 API 获取 |
| Discord | `discord` | ✅ 可通过 API 获取 |
| Slack | `slack` | ✅ 可通过 API 获取 |
| 飞书 | `lark` | ✅ 可通过 API 获取 |
| 钉钉 | `dingtalk` | ⚠️ 有限支持 |
| 企业微信 | `wecom` | ⚠️ 有限支持 |
| Misskey | `misskey` | ✅ 可通过 API 获取 |
| Satori | `satori` | 取决于实现 |
| 微信公众号 | `weixin_offacc` | ❌ 不支持 |
| WebChat | `webchat` | ❌ 不支持 |
### 1.3 解耦可行性评估
| 维度 | 评估 | 说明 |
|------|------|------|
| **技术可行性** | ✅ 高 | AstrBot 已有完善的平台抽象 |
| **工作量** | 中等 | 约 3-5 天工作量 |
| **风险** | 低 | 渐进式重构，可保持兼容 |
| **收益** | 高 | 支持 5+ 主流平台 |
---
## 2. AstrBot 平台抽象层分析
### 2.1 核心架构
```
┌─────────────────────────────────────────────────────────────────┐
│ AstrBot Core │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Platform Abstraction │ │
│ │ │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ Platform │ │ AstrMessage │ │ MessageType │ │ │
│ │ │ (Abstract) │ │ Event │ │ (Enum) │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
│ │ │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ AstrBotMsg │ │ Group │ │MessageMember │ │ │
│ │ │ (Model) │ │ (Model) │ │ (Model) │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Platform Adapters │ │
│ │ │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │aiocqhttp│ │telegram│ │discord │ │ slack │ │ lark │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │dingtalk│ │ wecom │ │misskey │ │ satori │ │ webchat│ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
### 2.2 核心抽象类
#### 2.2.1 Platform (平台基类)
**文件**: `astrbot/core/platform/platform.py`
```python
class Platform(abc.ABC):
"""平台适配器基类"""
def __init__(self, config: dict, event_queue: Queue):
self.config = config
self._event_queue = event_queue
self.client_self_id = uuid.uuid4().hex
@abc.abstractmethod
def run(self) -> Coroutine[Any, Any, None]:
"""启动平台"""
raise NotImplementedError
@abc.abstractmethod
def meta(self) -> PlatformMetadata:
"""获取平台元数据"""
raise NotImplementedError
async def send_by_session(self, session: MessageSesion, message_chain: MessageChain):
"""通过会话发送消息（跨平台统一接口）"""
pass
def commit_event(self, event: AstrMessageEvent):
"""提交事件到事件队列"""
self._event_queue.put_nowait(event)
def get_client(self):
"""获取平台客户端对象"""
pass
```
#### 2.2.2 AstrMessageEvent (消息事件基类)
**文件**: `astrbot/core/platform/astr_message_event.py`
```python
class AstrMessageEvent(abc.ABC):
"""统一消息事件基类 - 所有平台事件的父类"""
def __init__(
self,
message_str: str,
message_obj: AstrBotMessage,
platform_meta: PlatformMetadata,
session_id: str,
):
self.message_str = message_str # 纯文本消息
self.message_obj = message_obj # 完整消息对象
self.platform_meta = platform_meta # 平台元数据
self.session = MessageSession(...) # 会话信息
# 统一的跨平台方法
def get_platform_name(self) -> str: # 获取平台类型
def get_platform_id(self) -> str: # 获取平台实例ID
def get_message_str(self) -> str: # 获取消息文本
def get_message_type(self) -> MessageType: # 获取消息类型
def get_group_id(self) -> str: # 获取群组ID
def get_self_id(self) -> str: # 获取机器人ID
def get_sender_id(self) -> str: # 获取发送者ID
def get_sender_name(self) -> str: # 获取发送者名称
# 统一的发送方法
async def send(self, message: MessageChain):
"""发送消息（由子类实现具体逻辑）"""
pass
async def get_group(self, group_id: str = None) -> Group | None:
"""获取群组信息（由支持的平台实现）"""
pass
```
#### 2.2.3 AstrBotMessage (统一消息模型)
**文件**: `astrbot/core/platform/astrbot_message.py`
```python
class AstrBotMessage:
"""AstrBot 统一消息对象"""
type: MessageType # 消息类型 (GROUP_MESSAGE, FRIEND_MESSAGE, OTHER)
self_id: str # 机器人ID
session_id: str # 会话ID
message_id: str # 消息ID
group: Group | None # 群组信息
sender: MessageMember # 发送者信息
message: list[BaseMessageComponent] # 消息链
message_str: str # 纯文本消息
raw_message: object # 原始消息对象
timestamp: int # 时间戳
class MessageMember:
user_id: str # 用户ID (平台无关)
nickname: str | None # 昵称
class Group:
group_id: str # 群组ID (平台无关)
group_name: str | None # 群名称
group_owner: str | None # 群主ID
group_admins: list[str] # 管理员ID列表
members: list[MessageMember] # 群成员列表
```
#### 2.2.4 MessageType (消息类型枚举)
**文件**: `astrbot/core/platform/message_type.py`
```python
class MessageType(Enum):
GROUP_MESSAGE = "GroupMessage" # 群组消息
FRIEND_MESSAGE = "FriendMessage" # 私聊消息
OTHER_MESSAGE = "OtherMessage" # 其他消息
```
### 2.3 各平台适配器对比
| 平台 | 事件类 | 消息获取方法 | 消息发送方法 | 群信息获取 |
|------|--------|--------------|--------------|------------|
| aiocqhttp | `AiocqhttpMessageEvent` | `call_action("get_group_msg_history")` | `send_group_msg` | `get_group_info` |
| telegram | `TelegramPlatformEvent` | `get_chat_history()` | `send_message()` | `get_chat()` |
| discord | `DiscordPlatformEvent` | `channel.history()` | `channel.send()` | `get_channel()` |
| slack | `SlackMessageEvent` | `conversations_history()` | `chat_postMessage()` | `conversations_info()` |
| lark | `LarkMessageEvent` | 飞书 API | 飞书 API | 飞书 API |
### 2.4 平台检测与适配模式
**正确的跨平台写法**:
```python
# ✅ 推荐：使用基类 AstrMessageEvent
from astrbot.api.event import AstrMessageEvent
@filter.command("分析")
async def analyze(self, event: AstrMessageEvent):
# 使用统一接口
group_id = event.get_group_id()
platform = event.get_platform_name()
# 根据平台选择策略
if platform == "aiocqhttp":
messages = await self._fetch_qq_messages(event)
elif platform == "telegram":
messages = await self._fetch_telegram_messages(event)
elif platform == "discord":
messages = await self._fetch_discord_messages(event)
else:
yield event.plain_result(f"❌ 平台 {platform} 暂不支持消息历史获取")
return
```
---
## 3. 插件 QQ 硬编码清单
### 3.1 硬编码分类汇总
| 类别 | 文件 | 行数 | 严重程度 | 解耦难度 |
|------|------|------|----------|----------|
| 直接导入 aiocqhttp | `main.py` | 14-16 | 🔴 高 | 低 |
| 类型检查 AiocqhttpMessageEvent | `main.py` | 多处 | 🔴 高 | 低 |
| OneBot API 调用 | `message_handler.py` | 90-110 | 🔴 高 | 中 |
| OneBot API 调用 | `auto_scheduler.py` | 85-90 | 🔴 高 | 中 |
| QQ 表情类型处理 | `message_handler.py` | 206-256 | 🟡 中 | 中 |
| QQ 号相关逻辑 | `bot_manager.py` | 多处 | 🟡 中 | 低 |
| 消息格式假设 | `message_handler.py` | 133-147 | 🟡 中 | 中 |
| 错误码检查 | 多文件 | 多处 | 🟢 低 | 低 |
### 3.2 详细硬编码清单
#### 3.2.1 main.py - 入口文件
```python
# 🔴 硬编码 1: 直接导入 aiocqhttp 事件类
# 行 14-16
from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import (
AiocqhttpMessageEvent,
)
# 🔴 硬编码 2-13: 所有命令都限制 QQ 平台
# 行 122, 128-129
async def analyze_group_daily(self, event: AiocqhttpMessageEvent, ...):
if not isinstance(event, AiocqhttpMessageEvent):
yield event.plain_result("❌ 此功能仅支持QQ群聊")
return
# 同样的模式在以下命令中重复:
# - set_output_format (行 330, 336-338)
# - set_report_template (行 377, 383-385)
# - view_templates (行 447, 452-454)
# - install_pdf_deps (行 533, 538-540)
# - analysis_settings (行 556, 567-569)
# 🔴 硬编码 14: 直接调用 OneBot API 发送消息
# 行 218-225
if hasattr(bot_instance, "api") and hasattr(bot_instance.api, "call_action"):
await bot_instance.api.call_action(
"send_group_msg",
group_id=int(group_id),
message=message_chain,
)
```
#### 3.2.2 message_handler.py - 消息处理
```python
# 🔴 硬编码 15: QQ 号提取逻辑
# 行 40-48
def _extract_bot_qq_id_from_instance(self, bot_instance):
"""从bot实例中提取QQ号（单个）"""
if hasattr(bot_instance, "self_id") and bot_instance.self_id:
return str(bot_instance.self_id)
elif hasattr(bot_instance, "qq") and bot_instance.qq:
return str(bot_instance.qq)
...
# 🔴 硬编码 16: OneBot API 调用获取消息历史
# 行 90-111
if hasattr(bot_instance, "call_action"):
result = await bot_instance.call_action(
"get_group_msg_history", **payloads
)
elif hasattr(bot_instance, "api"):
# QQ 官方 bot (botClient) 不支持历史消息
logger.error("检测到 QQ 官方 Bot，官方 API 不支持获取历史消息")
return []
# 🟡 硬编码 17: QQ 消息格式假设
# 行 124-147
round_messages = result.get("messages", [])
for msg in round_messages:
msg_time = datetime.fromtimestamp(msg.get("time", 0))
sender_id = str(msg.get("sender", {}).get("user_id", ""))
# 🟡 硬编码 18-22: QQ 特定表情类型处理
# 行 206-256
elif content.get("type") == "face": # QQ基础表情
emoji_statistics.face_count += 1
elif content.get("type") == "mface": # 动画表情/魔法表情
emoji_statistics.mface_count += 1
elif content.get("type") == "bface": # 超级表情
emoji_statistics.bface_count += 1
elif content.get("type") == "sface": # 小表情
emoji_statistics.sface_count += 1
```
#### 3.2.3 auto_scheduler.py - 自动调度
```python
# 🔴 硬编码 23: OneBot API 调用获取群信息
# 行 85-92
if hasattr(bot_instance, "call_action"):
result = await bot_instance.call_action(
"get_group_info", group_id=int(group_id)
)
# 🟡 硬编码 24: OneBot 错误码检查
# 行 100-107
if "retcode=1200" in error_msg or "消息undefined不存在" in error_msg:
logger.warning(f"群 {group_id} 机器人不在此群中")
# 🔴 硬编码 25: OneBot API 获取群列表
# 行 478-479
result = await call_action_func("get_group_list")
```
#### 3.2.4 bot_manager.py - Bot 管理
```python
# 🟡 硬编码 26-28: QQ 号相关属性和方法
# 行 18, 39-47, 80-82
self._bot_qq_ids = [] # 命名暗示 QQ 专属
def set_bot_qq_ids(self, bot_qq_ids):
"""设置bot QQ号（支持单个QQ号或QQ号列表）"""
def has_bot_qq_id(self) -> bool:
"""检查是否有配置的bot QQ号"""
# 🟡 硬编码 29: 平台检查硬编码
# 行 151-154
if hasattr(event, "get_platform_name") and event.get_platform_name() != "aiocqhttp":
return False
```
#### 3.2.5 message_sender.py - 消息发送
```python
# 🔴 硬编码 30-32: OneBot API 调用发送消息
# 行 38-40, 73-75, 117-119
await bot.api.call_action("send_group_msg", group_id=group_id, message=...)
```
#### 3.2.6 retry.py - 重试管理
```python
# 🔴 硬编码 33-34: OneBot API 调用
# 行 192-206
if hasattr(bot, "api") and hasattr(bot.api, "call_action"):
result = await bot.api.call_action(
"send_group_msg", group_id=int(task.group_id), message=message
)
# 行 296-304
await bot.api.call_action(
"send_group_forward_msg",
group_id=int(task.group_id),
messages=nodes,
)
```
### 3.3 硬编码影响分析
```
┌─────────────────────────────────────────────────────────────────┐
│ 硬编码影响链 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ main.py │
│ └── import AiocqhttpMessageEvent ──────────────────────────┐ │
│ │ │ │
│ ▼ │ │
│ 所有命令处理器 │ │
│ (12个命令全部限制 QQ) │ │
│ │ │ │
│ ▼ │ │
│ message_handler.py │ │
│ └── call_action("get_group_msg_history") ──────────────────┤ │
│ │ │ │
│ ▼ │ │
│ auto_scheduler.py │ │
│ └── call_action("get_group_info") ─────────────────────────┤ │
│ └── call_action("get_group_list") ─────────────────────────┤ │
│ │ │ │
│ ▼ │ │
│ message_sender.py / retry.py │ │
│ └── call_action("send_group_msg") ─────────────────────────┘ │
│ └── call_action("send_group_forward_msg") │
│ │
│ 结果: 插件完全无法在非 QQ 平台使用 │
└─────────────────────────────────────────────────────────────────┘
```
---
## 4. 跨平台解耦方案
### 4.1 设计目标
1. **平台无关的核心逻辑** - 分析、报告生成与平台解耦
2. **可插拔的平台适配器** - 每个平台独立的消息获取/发送实现
3. **渐进式迁移** - 保持 QQ 功能完整，逐步添加其他平台
4. **统一的接口抽象** - 定义清晰的平台能力接口
### 4.2 架构设计
```
┌─────────────────────────────────────────────────────────────────┐
│ Plugin Architecture (目标) │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Application Layer │ │
│ │ main.py - 使用 AstrMessageEvent 基类 │ │
│ │ - 命令处理器接受所有平台事件 │ │
│ │ - 根据平台能力选择处理策略 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Domain Layer │ │
│ │ - MessageAnalyzer (平台无关) │ │
│ │ - ReportGenerator (平台无关) │ │
│ │ - LLMAnalyzer (平台无关) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Platform Abstraction Layer │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────┐ │ │
│ │ │ IPlatformMessageRepository (Interface) │ │ │
│ │ │ - fetch_messages(group_id, days) -> List[Msg] │ │ │
│ │ │ - get_group_info(group_id) -> GroupInfo │ │ │
│ │ │ - get_group_list() -> List[str] │ │ │
│ │ └──────────────────────────────────────────────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────┐ │ │
│ │ │ IPlatformMessageSender (Interface) │ │ │
│ │ │ - send_text(group_id, text) -> bool │ │ │
│ │ │ - send_image(group_id, image) -> bool │ │ │
│ │ │ - send_file(group_id, file) -> bool │ │ │
│ │ └──────────────────────────────────────────────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────┐ │ │
│ │ │ PlatformCapabilities (Value Object) │ │ │
│ │ │ - supports_message_history: bool │ │ │
│ │ │ - supports_group_list: bool │ │ │
│ │ │ - supports_file_upload: bool │ │ │
│ │ │ - supports_forward_message: bool │ │ │
│ │ └──────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Platform Implementations │ │
│ │ │ │
│ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │
│ │ │ OneBot │ │ Telegram │ │ Discord │ ... │ │
│ │ │ Adapter │ │ Adapter │ │ Adapter │ │ │
│ │ └────────────┘ └────────────┘ └────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
### 4.3 接口定义
#### 4.3.1 平台消息仓储接口
```python
# src/platform/interfaces.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class UnifiedMessage:
"""统一消息格式"""
message_id: str
sender_id: str
sender_name: str
content: str # 纯文本内容
raw_content: list # 原始消息链
timestamp: int
message_type: str # text, image, file, etc.
@dataclass
class UnifiedGroup:
"""统一群组格式"""
group_id: str
group_name: str
member_count: int
owner_id: Optional[str] = None
@dataclass
class PlatformCapabilities:
"""平台能力描述"""
platform_name: str
supports_message_history: bool = False
supports_group_list: bool = False
supports_group_info: bool = False
supports_file_upload: bool = False
supports_forward_message: bool = False
max_message_history_days: int = 0
class IPlatformMessageRepository(ABC):
"""平台消息仓储接口"""
@abstractmethod
async def fetch_messages(
self,
group_id: str,
days: int,
max_count: int = 1000
) -> List[UnifiedMessage]:
"""获取群消息历史"""
pass
@abstractmethod
async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]:
"""获取群信息"""
pass
@abstractmethod
async def get_group_list(self) -> List[str]:
"""获取群列表"""
pass
@abstractmethod
def get_capabilities(self) -> PlatformCapabilities:
"""获取平台能力"""
pass
class IPlatformMessageSender(ABC):
"""平台消息发送接口"""
@abstractmethod
async def send_text(self, group_id: str, text: str) -> bool:
"""发送文本消息"""
pass
@abstractmethod
async def send_image(self, group_id: str, image_url: str) -> bool:
"""发送图片消息"""
pass
@abstractmethod
async def send_file(self, group_id: str, file_path: str) -> bool:
"""发送文件"""
pass
```
#### 4.3.2 OneBot 实现示例
```python
# src/platform/adapters/onebot_adapter.py
from ..interfaces import (
IPlatformMessageRepository,
IPlatformMessageSender,
UnifiedMessage,
UnifiedGroup,
PlatformCapabilities,
)
class OneBotMessageRepository(IPlatformMessageRepository):
"""OneBot v11 消息仓储实现"""
def __init__(self, bot_instance):
self.bot = bot_instance
async def fetch_messages(
self, group_id: str, days: int, max_count: int = 1000
) -> List[UnifiedMessage]:
"""通过 get_group_msg_history 获取消息"""
if not hasattr(self.bot, "call_action"):
return []
try:
result = await self.bot.call_action(
"get_group_msg_history",
group_id=int(group_id),
count=max_count,
)
messages = []
for msg in result.get("messages", []):
# 转换为统一格式
unified = self._convert_message(msg)
if unified:
messages.append(unified)
return messages
except Exception as e:
logger.error(f"OneBot fetch_messages failed: {e}")
return []
def _convert_message(self, raw_msg: dict) -> UnifiedMessage:
"""将 OneBot 消息转换为统一格式"""
sender = raw_msg.get("sender", {})
# 提取纯文本内容
text_parts = []
for seg in raw_msg.get("message", []):
if seg.get("type") == "text":
text_parts.append(seg.get("data", {}).get("text", ""))
return UnifiedMessage(
message_id=str(raw_msg.get("message_id", "")),
sender_id=str(sender.get("user_id", "")),
sender_name=sender.get("nickname", "") or sender.get("card", ""),
content="".join(text_parts),
raw_content=raw_msg.get("message", []),
timestamp=raw_msg.get("time", 0),
message_type="mixed",
)
def get_capabilities(self) -> PlatformCapabilities:
return PlatformCapabilities(
platform_name="onebot",
supports_message_history=True,
supports_group_list=True,
supports_group_info=True,
supports_file_upload=True,
supports_forward_message=True,
max_message_history_days=7,
)
```
#### 4.3.3 Telegram 实现示例
```python
# src/platform/adapters/telegram_adapter.py
class TelegramMessageRepository(IPlatformMessageRepository):
"""Telegram 消息仓储实现"""
def __init__(self, bot_client):
self.bot = bot_client
async def fetch_messages(
self, group_id: str, days: int, max_count: int = 1000
) -> List[UnifiedMessage]:
"""通过 Telegram API 获取消息历史"""
try:
from datetime import datetime, timedelta
# Telegram 使用 chat_id
chat_id = int(group_id)
# 获取消息历史 (需要 bot 有读取历史的权限)
messages = []
async for message in self.bot.get_chat_history(
chat_id=chat_id,
limit=max_count,
):
# 过滤时间范围
msg_time = message.date
if msg_time < datetime.now() - timedelta(days=days):
break
unified = self._convert_message(message)
if unified:
messages.append(unified)
return messages
except Exception as e:
logger.error(f"Telegram fetch_messages failed: {e}")
return []
def get_capabilities(self) -> PlatformCapabilities:
return PlatformCapabilities(
platform_name="telegram",
supports_message_history=True,
supports_group_list=True,
supports_group_info=True,
supports_file_upload=True,
supports_forward_message=True,
max_message_history_days=30,
)
```
### 4.4 平台适配器工厂
```python
# src/platform/factory.py
from typing import Optional
from .interfaces import IPlatformMessageRepository, IPlatformMessageSender
from .adapters.onebot_adapter import OneBotMessageRepository, OneBotMessageSender
from .adapters.telegram_adapter import TelegramMessageRepository, TelegramMessageSender
from .adapters.discord_adapter import DiscordMessageRepository, DiscordMessageSender
class PlatformAdapterFactory:
"""平台适配器工厂"""
@staticmethod
def create_repository(
platform_name: str,
bot_instance
) -> Optional[IPlatformMessageRepository]:
"""根据平台类型创建消息仓储"""
adapters = {
"aiocqhttp": OneBotMessageRepository,
"telegram": TelegramMessageRepository,
"discord": DiscordMessageRepository,
"slack": SlackMessageRepository,
"lark": LarkMessageRepository,
}
adapter_class = adapters.get(platform_name)
if adapter_class:
return adapter_class(bot_instance)
return None
@staticmethod
def create_sender(
platform_name: str,
bot_instance
) -> Optional[IPlatformMessageSender]:
"""根据平台类型创建消息发送器"""
senders = {
"aiocqhttp": OneBotMessageSender,
"telegram": TelegramMessageSender,
"discord": DiscordMessageSender,
"slack": SlackMessageSender,
"lark": LarkMessageSender,
}
sender_class = senders.get(platform_name)
if sender_class:
return sender_class(bot_instance)
return None
@staticmethod
def get_supported_platforms() -> list[str]:
"""获取支持的平台列表"""
return ["aiocqhttp", "telegram", "discord", "slack", "lark"]
```
### 4.5 重构后的命令处理器
```python
# main.py (重构后)
from astrbot.api.event import AstrMessageEvent # 使用基类
class GroupDailyAnalysis(Star): # 改名，去掉 QQ 前缀
@filter.command("群分析")
@filter.permission_type(PermissionType.ADMIN)
async def analyze_group_daily(
self, event: AstrMessageEvent, days: int | None = None # 使用基类
):
"""分析群聊日常活动 - 跨平台支持"""
group_id = event.get_group_id()
if not group_id:
yield event.plain_result("❌ 请在群聊中使用此命令")
return
platform_name = event.get_platform_name()
# 获取平台适配器
repository = self._get_repository_for_platform(platform_name, event)
if not repository:
yield event.plain_result(f"❌ 平台 {platform_name} 暂不支持此功能")
return
# 检查平台能力
capabilities = repository.get_capabilities()
if not capabilities.supports_message_history:
yield event.plain_result(
f"❌ 平台 {platform_name} 不支持获取消息历史"
)
return
# 使用统一接口获取消息
messages = await repository.fetch_messages(group_id, days or 1)
if not messages:
yield event.plain_result("❌ 未找到足够的消息记录")
return
# 后续分析逻辑不变...
yield event.plain_result(f"📊 已获取 {len(messages)} 条消息，正在分析...")
# 分析和报告生成使用统一的消息格式
analysis_result = await self.message_analyzer.analyze_unified_messages(
messages, group_id, event.unified_msg_origin
)
# 发送报告
await self._send_report(event, analysis_result)
```
---
## 5. 执行路线图
### Phase 0: 准备工作 (0.5 天)
**目标**: 建立基础设施
| 任务 | 说明 |
|------|------|
| 创建 `src/platform/` 目录 | 平台抽象层 |
| 定义接口文件 | `interfaces.py` |
| 创建工厂类 | `factory.py` |
### Phase 1: OneBot 适配器 (1 天)
**目标**: 将现有 QQ 逻辑封装为适配器
| 任务 | 说明 |
|------|------|
| 实现 `OneBotMessageRepository` | 封装 `get_group_msg_history` |
| 实现 `OneBotMessageSender` | 封装消息发送 |
| 添加消息格式转换 | OneBot → UnifiedMessage |
| 单元测试 | 确保功能不变 |
### Phase 2: 核心逻辑解耦 (1 天)
**目标**: 使核心逻辑平台无关
| 任务 | 说明 |
|------|------|
| 修改 `MessageHandler` | 使用 `UnifiedMessage` |
| 修改 `MessageAnalyzer` | 移除平台假设 |
| 修改 `AutoScheduler` | 使用适配器工厂 |
| 修改 `BotManager` | 重命名 QQ 相关方法 |
### Phase 3: main.py 重构 (0.5 天)
**目标**: 使命令处理器跨平台
| 任务 | 说明 |
|------|------|
| 移除 `AiocqhttpMessageEvent` 导入 | 使用基类 |
| 移除类型检查 | 改用能力检查 |
| 添加平台适配器选择逻辑 | 根据 `platform_name` |
| 更新错误消息 | 更通用的提示 |
### Phase 4: 添加 Telegram 支持 (1 天)
**目标**: 验证跨平台架构
| 任务 | 说明 |
|------|------|
| 实现 `TelegramMessageRepository` | 使用 python-telegram-bot |
| 实现 `TelegramMessageSender` | |
| 测试 Telegram 群分析 | 端到端验证 |
### Phase 5: 添加更多平台 (可选)
| 平台 | 优先级 | 工作量 |
|------|--------|--------|
| Discord | P1 | 1 天 |
| Slack | P2 | 1 天 |
| 飞书 | P2 | 1 天 |
| 钉钉 | P3 | 1 天 |
---
## 6. 风险与缓解
| 风险 | 可能性 | 影响 | 缓解措施 |
|------|--------|------|----------|
| 平台 API 差异大 | 高 | 中 | 统一消息格式 + 能力检查 |
| 消息历史获取受限 | 高 | 高 | 明确标注平台能力，提供降级方案 |
| 表情/特殊消息处理 | 中 | 低 | 只提取文本内容进行分析 |
| 测试覆盖不足 | 中 | 中 | 为每个适配器编写集成测试 |
| 性能差异 | 低 | 低 | 异步处理 + 缓存 |
---
## 7. 总结
### 7.1 关键发现
1. **AstrBot 已有完善的平台抽象** - 不需要自建抽象层
2. **插件硬编码严重但可解耦** - 约 34 处需要修改
3. **核心分析逻辑平台无关** - LLM 分析、报告生成不受影响
4. **渐进式迁移可行** - 可以保持 QQ 功能同时添加新平台
### 7.2 建议优先级
| 优先级 | 任务 | 收益 |
|--------|------|------|
| **P0** | 定义平台抽象接口 | 架构基础 |
| **P0** | 封装 OneBot 适配器 | 保持现有功能 |
| **P1** | 重构 main.py 使用基类 | 解除平台限制 |
| **P1** | 添加 Telegram 支持 | 验证架构 |
| **P2** | 添加 Discord 支持 | 扩大用户群 |
### 7.3 预期成果
- ✅ 插件可在 5+ 主流平台运行
- ✅ 新增平台只需实现适配器接口
- ✅ 核心逻辑无需修改
- ✅ 保持与 AstrBot 框架的对齐
---
## 附录 A: 平台 API 对比
| 功能 | OneBot v11 | Telegram | Discord | Slack |
|------|------------|----------|---------|-------|
| 获取消息历史 | `get_group_msg_history` | `get_chat_history` | `channel.history()` | `conversations.history` |
| 获取群信息 | `get_group_info` | `get_chat` | `get_channel` | `conversations.info` |
| 获取群列表 | `get_group_list` | `get_my_commands` | `guilds` | `conversations.list` |
| 发送文本 | `send_group_msg` | `send_message` | `channel.send` | `chat.postMessage` |
| 发送图片 | `[CQ:image]` | `send_photo` | `channel.send(file=)` | `files.upload` |
| 发送文件 | `[CQ:file]` | `send_document` | `channel.send(file=)` | `files.upload` |
| 转发消息 | `send_group_forward_msg` | N/A | N/A | N/A |
## 附录 B: 参考资料
1. AstrBot 官方文档 - https://astrbot.app/
2. OneBot v11 标准 - https://github.com/botuniverse/onebot-11
3. python-telegram-bot - https://python-telegram-bot.org/
4. Pycord (Discord) - https://pycord.dev/
5. Slack SDK - https://slack.dev/python-slack-sdk/
docs/09_ddd_cross_platform_complete_guide.md
+796
View File
@@ -0,0 +1,796 @@
# 09. DDD + 跨平台重构完整方案 (DDD & Cross-Platform Refactoring Complete Guide)
> **文档日期**: 2026-02-08
> **版本**: v1.0
> **整合文档**: 07_ddd_refactoring_analysis.md + 08_cross_platform_decoupling_analysis.md
> **目的**: 提供完整的 DDD 重构方案，同时解决跨平台解耦问题
---
## 1. 执行摘要
### 1.1 问题分析
本插件存在两个核心问题需要同时解决：
| 问题 | 现状 | 影响 |
|------|------|------|
| **平台耦合** | 34 处 QQ 硬编码 | 完全无法在其他平台使用 |
| **架构分层混乱** | 业务逻辑与基础设施混合 | 测试困难，维护成本高 |
### 1.2 解决方案概述
采用 **DDD 分层架构 + 平台适配器模式**：
```
┌─────────────────────────────────────────────────────────────────┐
│ Interface Layer (main.py) │
│ - 使用 AstrMessageEvent 基类，不再限制 QQ 平台 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Application Layer │
│ - AnalysisOrchestrator: 分析流程编排 │
│ - SchedulingService: 定时任务管理 │
│ - ReportingService: 报告生成与分发 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Domain Layer (平台无关) │
│ - Entities: AnalysisTask, GroupAnalysisResult │
│ - Value Objects: UnifiedMessage, PlatformCapabilities │
│ - Domain Services: TopicAnalyzer, StatisticsCalculator, etc. │
│ - Repository Interfaces: IMessageRepository, IMessageSender │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Infrastructure Layer │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Platform Adapters (实现 Repository Interfaces) │ │
│ │ - OneBotAdapter (QQ) │ │
│ │ - TelegramAdapter │ │
│ │ - DiscordAdapter │ │
│ │ - SlackAdapter │ │
│ │ - LarkAdapter (飞书) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ - LLMClient, ConfigManager, CircuitBreaker, etc. │
└─────────────────────────────────────────────────────────────────┘
```
### 1.3 预期收益
| 收益 | 量化 |
|------|------|
| 跨平台支持 | 从 1 个平台扩展到 5+ 个平台 |
| 可测试性 | 单元测试覆盖率可达 80%+ |
| 可维护性 | 新增平台只需 1 天工作量 |
| 可扩展性 | 新增分析器只需实现接口 |
---
## 2. 限界上下文 (Bounded Contexts)
### 2.1 上下文划分
```
┌─────────────────────────────────────────────────────────────────────────┐
│ 群聊日常分析插件 (Plugin) │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 调度上下文 │ │ 分析上下文 │ │ 报告上下文 │ │ 平台上下文 │ │
│ │ (Scheduling)│ │ (Analysis) │ │ (Reporting) │ │ (Platform) │ │
│ │ │ │ │ │ │ │ [核心] │ │
│ │ - 定时任务 │ │ - 话题分析 │ │ - 报告生成 │ │ - 消息获取 │ │
│ │ - 并发控制 │ │ - 用户画像 │ │ - 格式转换 │ │ - 消息发送 │ │
│ │ - 任务编排 │ │ - 金句提取 │ │ - 重试机制 │ │ - 群组信息 │ │
│ │ │ │ - 统计计算 │ │ │ │ - 平台能力 │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │ │
│ └────────────────┴────────────────┴────────────────┘ │
│ │ │
│ ┌─────────────────────────────────┴─────────────────────────────────┐ │
│ │ 共享内核 (Shared Kernel) │ │
│ │ - ConfigManager: 配置管理 │ │
│ │ - TraceContext: 链路追踪 │ │
│ │ - CircuitBreaker, RateLimiter: 弹性组件 │ │
│ │ - UnifiedMessage: 统一消息模型 [跨平台核心] │ │
│ │ - PlatformCapabilities: 平台能力描述 [跨平台核心] │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
```
### 2.2 上下文职责
| 上下文 | 职责 | 关键组件 |
|--------|------|----------|
| **调度** | 定时任务、并发控制、任务编排 | SchedulingService, AutoScheduler |
| **分析** | 话题分析、用户画像、金句提取、统计 | TopicAnalyzer, UserTitleAnalyzer, StatisticsCalculator |
| **报告** | 报告生成、格式转换、发送、重试 | ReportGenerator, ReportDispatcher, RetryManager |
| **平台** | 消息获取、消息发送、群组信息 | PlatformAdapter, IMessageRepository, IMessageSender |
---
## 3. 领域模型设计
### 3.1 统一消息模型 (UnifiedMessage)
**这是跨平台的核心抽象**，所有平台的消息都会被转换为此格式：
```python
# src/domain/value_objects/unified_message.py
from dataclasses import dataclass
from typing import Optional, Any
from enum import Enum
class MessageContentType(Enum):
TEXT = "text"
IMAGE = "image"
FILE = "file"
EMOJI = "emoji"
REPLY = "reply"
FORWARD = "forward"
UNKNOWN = "unknown"
@dataclass(frozen=True)
class MessageContent:
"""消息内容值对象"""
type: MessageContentType
text: str = ""
url: str = ""
emoji_id: str = ""
raw_data: Any = None # 保留原始数据用于调试
@dataclass(frozen=True)
class UnifiedMessage:
"""
统一消息格式 - 跨平台核心值对象
设计原则：
1. 只保留分析需要的字段
2. 使用平台无关的类型
3. 不可变 (frozen=True)
"""
message_id: str
sender_id: str
sender_name: str
group_id: str
text_content: str # 纯文本，用于 LLM 分析
contents: tuple[MessageContent, ...] # 完整消息链
timestamp: int
platform: str # 来源平台标识
reply_to: Optional[str] = None
def has_text(self) -> bool:
return bool(self.text_content.strip())
def get_emoji_count(self) -> int:
return sum(1 for c in self.contents if c.type == MessageContentType.EMOJI)
```
### 3.2 平台能力描述 (PlatformCapabilities)
```python
# src/domain/value_objects/platform_capabilities.py
from dataclasses import dataclass
@dataclass(frozen=True)
class PlatformCapabilities:
"""
平台能力描述 - 用于运行时决策
每个平台适配器必须声明自己的能力，
应用层根据能力决定是否可以执行某些操作。
"""
platform_name: str
# 消息获取能力
supports_message_history: bool = False
max_message_history_days: int = 0
max_message_count: int = 0
# 群组信息能力
supports_group_list: bool = False
supports_group_info: bool = False
supports_member_list: bool = False
# 消息发送能力
supports_text_message: bool = True
supports_image_message: bool = True
supports_file_message: bool = False
supports_forward_message: bool = False
def can_analyze(self) -> bool:
"""是否支持群聊分析"""
return self.supports_message_history and self.max_message_history_days > 0
# 预定义平台能力
ONEBOT_CAPABILITIES = PlatformCapabilities(
platform_name="onebot",
supports_message_history=True, max_message_history_days=7, max_message_count=10000,
supports_group_list=True, supports_group_info=True, supports_member_list=True,
supports_text_message=True, supports_image_message=True,
supports_file_message=True, supports_forward_message=True,
)
TELEGRAM_CAPABILITIES = PlatformCapabilities(
platform_name="telegram",
supports_message_history=True, max_message_history_days=30, max_message_count=10000,
supports_group_list=True, supports_group_info=True, supports_member_list=True,
supports_text_message=True, supports_image_message=True, supports_file_message=True,
)
DISCORD_CAPABILITIES = PlatformCapabilities(
platform_name="discord",
supports_message_history=True, max_message_history_days=30, max_message_count=10000,
supports_group_list=True, supports_group_info=True, supports_member_list=True,
supports_text_message=True, supports_image_message=True, supports_file_message=True,
)
```
### 3.3 仓储接口
```python
# src/domain/repositories/message_repository.py
from abc import ABC, abstractmethod
from typing import List
from ..value_objects.unified_message import UnifiedMessage
from ..value_objects.platform_capabilities import PlatformCapabilities
class IMessageRepository(ABC):
"""消息仓储接口 - 每个平台适配器必须实现"""
@abstractmethod
async def fetch_messages(
self, group_id: str, days: int, max_count: int = 1000
) -> List[UnifiedMessage]:
"""获取群组历史消息，返回统一格式"""
pass
@abstractmethod
def get_capabilities(self) -> PlatformCapabilities:
"""获取平台能力描述"""
pass
# src/domain/repositories/message_sender.py
class IMessageSender(ABC):
"""消息发送接口"""
@abstractmethod
async def send_text(self, group_id: str, text: str) -> bool:
pass
@abstractmethod
async def send_image(self, group_id: str, image_url: str, caption: str = "") -> bool:
pass
@abstractmethod
async def send_file(self, group_id: str, file_path: str) -> bool:
pass
# src/domain/repositories/group_info_repository.py
@dataclass
class UnifiedGroup:
group_id: str
group_name: str
member_count: int
owner_id: Optional[str] = None
class IGroupInfoRepository(ABC):
"""群组信息仓储接口"""
@abstractmethod
async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]:
pass
@abstractmethod
async def get_group_list(self) -> List[str]:
pass
```
### 3.4 实体设计
```python
# src/domain/entities/analysis_task.py
from dataclasses import dataclass, field
from enum import Enum
import time, uuid
class TaskStatus(Enum):
PENDING = "pending"
CHECKING_PLATFORM = "checking_platform"
FETCHING_MESSAGES = "fetching_messages"
ANALYZING = "analyzing"
GENERATING_REPORT = "generating_report"
SENDING = "sending"
COMPLETED = "completed"
FAILED = "failed"
UNSUPPORTED_PLATFORM = "unsupported_platform"
@dataclass
class AnalysisTask:
"""分析任务实体 - 聚合根"""
id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
group_id: str = ""
platform_name: str = ""
trace_id: str = ""
status: TaskStatus = TaskStatus.PENDING
is_manual: bool = False
created_at: float = field(default_factory=time.time)
started_at: Optional[float] = None
completed_at: Optional[float] = None
result_id: Optional[str] = None
error_message: Optional[str] = None
def start(self, capabilities: PlatformCapabilities) -> bool:
"""开始任务，验证平台能力"""
if not capabilities.can_analyze():
self.status = TaskStatus.UNSUPPORTED_PLATFORM
self.error_message = f"Platform {capabilities.platform_name} does not support analysis"
return False
self.status = TaskStatus.FETCHING_MESSAGES
self.started_at = time.time()
return True
def complete(self, result_id: str):
self.status = TaskStatus.COMPLETED
self.result_id = result_id
self.completed_at = time.time()
def fail(self, error: str):
self.status = TaskStatus.FAILED
self.error_message = error
self.completed_at = time.time()
```
---
## 4. 平台适配器设计
### 4.1 适配器架构
```
┌─────────────────────────────────────────────────────────────────┐
│ PlatformAdapter (Base) │
│ - message_repository: IMessageRepository │
│ - message_sender: IMessageSender │
│ - group_info_repository: IGroupInfoRepository │
│ - capabilities: PlatformCapabilities │
└─────────────────────────────────────────────────────────────────┘
△
┌─────────────────────┼─────────────────────┐
│ │ │
┌───────┴───────┐ ┌─────────┴─────────┐ ┌───────┴───────┐
│ OneBotAdapter │ │ TelegramAdapter │ │ DiscordAdapter │
│ │ │ │ │ │
│ - call_action │ │ - python-telegram │ │ - pycord │
│ - OneBot v11 │ │ -bot API │ │ - Discord API │
└───────────────┘ └───────────────────┘ └───────────────┘
```
### 4.2 OneBot 适配器实现
```python
# src/infrastructure/platform/adapters/onebot_adapter.py
from datetime import datetime, timedelta
from typing import List, Optional, Any
from astrbot.api import logger
from ....domain.repositories.message_repository import IMessageRepository
from ....domain.value_objects.unified_message import UnifiedMessage, MessageContent, MessageContentType
from ....domain.value_objects.platform_capabilities import PlatformCapabilities, ONEBOT_CAPABILITIES
class OneBotMessageRepository(IMessageRepository):
"""OneBot v11 消息仓储实现"""
def __init__(self, bot_instance: Any, bot_self_ids: List[str] = None):
self.bot = bot_instance
self.bot_self_ids = bot_self_ids or []
async def fetch_messages(self, group_id: str, days: int, max_count: int = 1000) -> List[UnifiedMessage]:
if not hasattr(self.bot, "call_action"):
return []
try:
result = await self.bot.call_action(
"get_group_msg_history",
group_id=int(group_id),
count=max_count,
)
if not result or "messages" not in result:
return []
end_time = datetime.now()
start_time = end_time - timedelta(days=days)
messages = []
for raw_msg in result.get("messages", []):
msg_time = datetime.fromtimestamp(raw_msg.get("time", 0))
if not (start_time <= msg_time <= end_time):
continue
sender_id = str(raw_msg.get("sender", {}).get("user_id", ""))
if sender_id in self.bot_self_ids:
continue
unified = self._convert_message(raw_msg, group_id)
if unified:
messages.append(unified)
return messages
except Exception as e:
if "retcode=1200" in str(e):
logger.warning(f"Bot not in group {group_id}")
else:
logger.error(f"OneBot fetch_messages failed: {e}")
return []
def _convert_message(self, raw_msg: dict, group_id: str) -> Optional[UnifiedMessage]:
"""将 OneBot 消息转换为统一格式"""
try:
sender = raw_msg.get("sender", {})
message_chain = raw_msg.get("message", [])
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", "")))
elif seg_type in ("face", "mface", "bface", "sface"):
contents.append(MessageContent(type=MessageContentType.EMOJI, emoji_id=str(seg_data.get("id", ""))))
else:
contents.append(MessageContent(type=MessageContentType.UNKNOWN, raw_data=seg))
return UnifiedMessage(
message_id=str(raw_msg.get("message_id", "")),
sender_id=str(sender.get("user_id", "")),
sender_name=sender.get("nickname", "") or sender.get("card", ""),
group_id=group_id,
text_content="".join(text_parts),
contents=tuple(contents),
timestamp=raw_msg.get("time", 0),
platform="onebot",
)
except Exception as e:
logger.warning(f"Failed to convert message: {e}")
return None
def get_capabilities(self) -> PlatformCapabilities:
return ONEBOT_CAPABILITIES
```
### 4.3 适配器工厂
```python
# src/infrastructure/platform/factory.py
from typing import Optional, Any
class PlatformAdapterFactory:
"""平台适配器工厂"""
_adapters = {
"aiocqhttp": OneBotAdapter,
# "telegram": TelegramAdapter,
# "discord": DiscordAdapter,
}
@classmethod
def create(cls, platform_name: str, bot_instance: Any, config: dict = None) -> Optional[PlatformAdapter]:
adapter_class = cls._adapters.get(platform_name)
if adapter_class is None:
return None
return adapter_class(bot_instance, config)
@classmethod
def get_supported_platforms(cls) -> list[str]:
return list(cls._adapters.keys())
@classmethod
def is_supported(cls, platform_name: str) -> bool:
return platform_name in cls._adapters
```
---
## 5. 应用层设计
### 5.1 分析编排器
```python
# src/application/analysis_orchestrator.py
from typing import Optional
import asyncio
from astrbot.api import logger
from ..domain.entities.analysis_task import AnalysisTask, TaskStatus
from ..domain.entities.analysis_result import GroupAnalysisResult
from ..domain.repositories.message_repository import IMessageRepository
class AnalysisOrchestrator:
"""
分析流程编排器 - 应用层核心
职责：协调消息获取、分析、报告生成
特点：平台无关，通过仓储接口与基础设施层交互
"""
def __init__(self, config_manager, llm_client, history_repository):
self.config_manager = config_manager
self.llm_client = llm_client
self.history_repository = history_repository
# 领域服务
self.statistics_calculator = StatisticsCalculator()
self.topic_analyzer = TopicAnalyzer(llm_client)
self.user_title_analyzer = UserTitleAnalyzer(llm_client)
self.golden_quote_analyzer = GoldenQuoteAnalyzer(llm_client)
async def execute(
self,
task: AnalysisTask,
message_repository: IMessageRepository,
unified_msg_origin: str = "",
) -> Optional[GroupAnalysisResult]:
"""执行分析任务"""
try:
# 1. 验证平台能力
capabilities = message_repository.get_capabilities()
if not task.start(capabilities):
logger.warning(f"Platform {capabilities.platform_name} not supported")
return None
# 2. 获取消息
days = self.config_manager.get_analysis_days()
max_count = self.config_manager.get_max_messages()
messages = await message_repository.fetch_messages(task.group_id, days, max_count)
if not messages:
task.fail("No messages found")
return None
min_threshold = self.config_manager.get_min_messages_threshold()
if len(messages) < min_threshold:
task.fail(f"Not enough messages: {len(messages)} < {min_threshold}")
return None
# 3. 执行分析
task.advance_to(TaskStatus.ANALYZING)
result = GroupAnalysisResult(
group_id=task.group_id,
trace_id=task.trace_id,
message_count=len(messages),
)
# 统计计算 (本地)
result.statistics = self.statistics_calculator.calculate(messages)
# LLM 分析 (并行)
topics, titles, quotes = await asyncio.gather(
self.topic_analyzer.analyze(messages, unified_msg_origin),
self.user_title_analyzer.analyze(messages, unified_msg_origin),
self.golden_quote_analyzer.analyze(messages, unified_msg_origin),
return_exceptions=True
)
# 处理结果
if not isinstance(topics, Exception):
result.topics = topics
if not isinstance(titles, Exception):
result.user_titles = titles
if not isinstance(quotes, Exception):
result.golden_quotes = quotes
# 4. 保存并完成
await self.history_repository.save(task.group_id, result)
task.complete(result.id)
return result
except Exception as e:
task.fail(str(e))
logger.error(f"Analysis failed: {e}", exc_info=True)
return None
```
---
## 6. 目标目录结构
```
astrbot_plugin_group_daily_analysis/ # 去掉 QQ 前缀
├── main.py # Interface Layer
├── src/
│ ├── application/ # Application Layer
│ │ ├── __init__.py
│ │ ├── analysis_orchestrator.py # 分析流程编排
│ │ ├── scheduling_service.py # 定时任务服务
│ │ └── reporting_service.py # 报告服务
│ │
│ ├── domain/ # Domain Layer
│ │ ├── __init__.py
│ │ ├── entities/
│ │ │ ├── analysis_task.py
│ │ │ └── analysis_result.py
│ │ ├── value_objects/
│ │ │ ├── unified_message.py # 统一消息格式
│ │ │ ├── platform_capabilities.py # 平台能力
│ │ │ ├── topic.py, user_title.py, etc.
│ │ ├── services/
│ │ │ ├── topic_analyzer.py
│ │ │ ├── statistics_calculator.py
│ │ │ └── report_generator.py
│ │ └── repositories/
│ │ ├── message_repository.py # IMessageRepository
│ │ ├── message_sender.py # IMessageSender
│ │ └── group_info_repository.py # IGroupInfoRepository
│ │
│ ├── infrastructure/ # Infrastructure Layer
│ │ ├── __init__.py
│ │ ├── platform/
│ │ │ ├── factory.py # 适配器工厂
│ │ │ ├── base.py # 适配器基类
│ │ │ └── adapters/
│ │ │ ├── onebot_adapter.py # OneBot (QQ)
│ │ │ ├── telegram_adapter.py
│ │ │ ├── discord_adapter.py
│ │ │ └── slack_adapter.py
│ │ ├── persistence/
│ │ ├── llm/
│ │ ├── config/
│ │ └── resilience/
│ │
│ └── shared/
│ ├── exceptions.py
│ └── constants.py
│
├── tests/
└── docs/
```
---
## 7. 重构路线图
### 总览
| Phase | 内容 | 工作量 | 优先级 |
|-------|------|--------|--------|
| Phase 0 | 准备工作：目录结构、接口定义 | 1 天 | P0 |
| Phase 1 | 平台适配器：OneBot 实现 | 2-3 天 | P0 |
| Phase 2 | 领域层：值对象、实体、服务 | 2-3 天 | P0 |
| Phase 3 | 应用层：编排器、服务 | 2-3 天 | P0 |
| Phase 4 | 接口层：main.py 重构 | 1-2 天 | P0 |
| Phase 5 | 新平台：Telegram、Discord | 每平台 1 天 | P1 |
| Phase 6 | 测试与文档 | 2 天 | P1 |
| **总计** | | **12-16 天** | |
### Phase 0: 准备工作 (1 天)
```bash
# 创建目录结构
mkdir -p src/{application,domain/{entities,value_objects,services,repositories},infrastructure/{platform/adapters,persistence,llm,config,resilience},shared}
```
**任务清单**:
- [ ] 创建目录结构
- [ ] 定义 UnifiedMessage 值对象
- [ ] 定义 PlatformCapabilities 值对象
- [ ] 定义 IMessageRepository 接口
- [ ] 定义 IMessageSender 接口
- [ ] 定义 IGroupInfoRepository 接口
- [ ] 定义共享异常类
### Phase 1: 平台适配器 (2-3 天)
**任务清单**:
- [ ] 实现 OneBotMessageRepository
- [ ] 实现 OneBotMessageSender
- [ ] 实现 OneBotGroupInfoRepository
- [ ] 实现 OneBotAdapter (组合)
- [ ] 实现 PlatformAdapterFactory
- [ ] 编写单元测试
### Phase 2: 领域层重构 (2-3 天)
**任务清单**:
- [ ] 迁移 data_models.py 到 value_objects/
- [ ] 实现 AnalysisTask 实体
- [ ] 实现 GroupAnalysisResult 实体
- [ ] 重构分析器使用 UnifiedMessage
- [ ] 重构 StatisticsCalculator
### Phase 3: 应用层重构 (2-3 天)
**任务清单**:
- [ ] 实现 AnalysisOrchestrator
- [ ] 实现 SchedulingService
- [ ] 实现 ReportingService
- [ ] 重构 AutoScheduler
### Phase 4: 接口层重构 (1-2 天)
**任务清单**:
- [ ] 移除 AiocqhttpMessageEvent 导入
- [ ] 使用 AstrMessageEvent 基类
- [ ] 添加平台适配器选择逻辑
- [ ] 更新错误消息
### Phase 5: 新平台支持 (每平台 1 天)
**Telegram 适配器**:
- [ ] TelegramMessageRepository
- [ ] TelegramMessageSender
- [ ] TelegramAdapter
**Discord 适配器**:
- [ ] DiscordMessageRepository
- [ ] DiscordMessageSender
- [ ] DiscordAdapter
---
## 8. 风险与缓解
| 风险 | 可能性 | 影响 | 缓解措施 |
|------|--------|------|----------|
| 重构引入回归 | 中 | 高 | 渐进式迁移，保留原代码 |
| 平台 API 差异 | 高 | 中 | 统一消息格式 + 能力检查 |
| 消息历史受限 | 高 | 高 | 明确能力声明，提供降级 |
| 过度设计 | 中 | 中 | YAGNI 原则 |
---
## 9. 总结
### 9.1 核心设计决策
1. **UnifiedMessage** - 跨平台消息抽象，领域层只处理此格式
2. **PlatformCapabilities** - 平台能力声明，运行时决策
3. **IMessageRepository** - 消息获取抽象，每个平台实现
4. **PlatformAdapterFactory** - 适配器工厂，统一创建入口
### 9.2 收益总结
| 维度 | 重构前 | 重构后 |
|------|--------|--------|
| 平台支持 | 仅 QQ | QQ + Telegram + Discord + ... |
| 可测试性 | 困难 | 80%+ 覆盖率 |
| 新增平台 | N/A | 1 天 |
| 新增分析器 | 需要了解 QQ API | 只需实现接口 |
### 9.3 参考文档
- 07_ddd_refactoring_analysis.md - 原 DDD 分析
- 08_cross_platform_decoupling_analysis.md - 跨平台调研
- 06_review.md - 务实重构审查
---
## 附录: 平台能力对比
| 功能 | OneBot | Telegram | Discord | Slack | 飞书 |
|------|--------|----------|---------|-------|------|
| 消息历史 | ✅ 7天 | ✅ 30天 | ✅ 30天 | ✅ 30天 | ✅ 30天 |
| 群列表 | ✅ | ✅ | ✅ | ✅ | ✅ |
| 发送图片 | ✅ | ✅ | ✅ | ✅ | ✅ |
| 发送文件 | ✅ | ✅ | ✅ | ✅ | ✅ |
| 转发消息 | ✅ | ❌ | ❌ | ❌ | ❌ |
| 消息回应 | ❌ | ✅ | ✅ | ✅ | ✅ |
docs/10_platform_abstraction_layer.md
+2044
View File
File diff suppressed because it is too large Load Diff