From a2adfe65617f4bef9d0adda8a5390096fe8bc6d6 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Sun, 8 Feb 2026 16:24:08 +0800 Subject: [PATCH] docs: add platform integration guide and unit tests --- docs/12_platform_integration_guide.md | 495 ++++++++++++++++++++++++++ tests/unit_test_platform.py | 102 ++++++ 2 files changed, 597 insertions(+) create mode 100644 docs/12_platform_integration_guide.md create mode 100644 tests/unit_test_platform.py diff --git a/docs/12_platform_integration_guide.md b/docs/12_platform_integration_guide.md new file mode 100644 index 0000000..bcf3035 --- /dev/null +++ b/docs/12_platform_integration_guide.md @@ -0,0 +1,495 @@ +# 平台接入开发指南 + +本文档说明如何为 QQ 群聊日报分析插件接入新的消息平台。 + +## 目录 + +1. [架构概述](#架构概述) +2. [快速开始](#快速开始) +3. [详细步骤](#详细步骤) +4. [接口说明](#接口说明) +5. [最佳实践](#最佳实践) +6. [示例代码](#示例代码) +7. [测试指南](#测试指南) + +--- + +## 架构概述 + +本插件采用 DDD（领域驱动设计）架构，通过平台适配器模式实现多平台支持： + +``` +┌─────────────────────────────────────────────────────────┐ +│ 应用层 (Application) │ +│ AnalysisOrchestrator │ +└─────────────────────────┬───────────────────────────────┘ + │ 使用 + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 基础设施层 (Infrastructure) │ +│ │ +│ ┌─────────────────────────────────────────────────┐ │ +│ │ PlatformAdapter (抽象基类) │ │ +│ │ - fetch_messages() │ │ +│ │ - send_text/image/file() │ │ +│ │ - get_group_info() │ │ +│ │ - convert_to_raw_format() │ │ +│ └─────────────────────────────────────────────────┘ │ +│ ▲ ▲ ▲ │ +│ │ │ │ │ +│ ┌────────┴───┐ ┌──────┴──────┐ ┌────┴────────┐ │ +│ │OneBotAdapter│ │DiscordAdapter│ │ 新平台Adapter │ │ +│ │ (QQ平台) │ │ (Discord) │ │ (待实现) │ │ +│ └────────────┘ └─────────────┘ └─────────────┘ │ +└─────────────────────────────────────────────────────────┘ +``` + +### 核心组件 + +| 组件 | 路径 | 说明 | +|------|------|------| +| PlatformAdapter | `src/infrastructure/platform/base.py` | 平台适配器抽象基类 | +| PlatformAdapterFactory | `src/infrastructure/platform/factory.py` | 适配器工厂，管理注册和创建 | +| UnifiedMessage | `src/domain/value_objects/unified_message.py` | 统一消息格式 | +| PlatformCapabilities | `src/domain/value_objects/platform_capabilities.py` | 平台能力声明 | + +--- + +## 快速开始 + +接入新平台只需 3 步： + +### 步骤 1：创建适配器文件 + +```bash +# 在 adapters 目录下创建新文件 +touch src/infrastructure/platform/adapters/your_platform_adapter.py +``` + +### 步骤 2：实现适配器类 + +```python +from ..base import PlatformAdapter +from ....domain.value_objects.platform_capabilities import PlatformCapabilities + +class YourPlatformAdapter(PlatformAdapter): + def _init_capabilities(self) -> PlatformCapabilities: + return PlatformCapabilities( + platform_name="your_platform", + supports_history_fetch=True, + # ... 其他能力 + ) + + # 实现所有抽象方法... +``` + +### 步骤 3：注册适配器 + +在 `factory.py` 的 `_register_adapters()` 函数中添加： + +```python +try: + from .adapters.your_platform_adapter import YourPlatformAdapter + PlatformAdapterFactory.register("your_platform", YourPlatformAdapter) +except ImportError: + pass +``` + +--- + +## 详细步骤 + +### 1. 定义平台能力 + +首先，明确你的平台支持哪些功能： + +```python +from ....domain.value_objects.platform_capabilities import PlatformCapabilities + +YOUR_PLATFORM_CAPABILITIES = PlatformCapabilities( + platform_name="your_platform", # 平台标识符 + supports_history_fetch=True, # 是否支持历史消息获取 + supports_image_message=True, # 是否支持图片消息 + supports_file_message=True, # 是否支持文件消息 + supports_text_message=True, # 是否支持文本消息 + max_history_days=7, # 历史消息最大天数 + max_messages_per_fetch=100, # 单次获取最大消息数 + supports_avatar=True, # 是否支持头像获取 + supports_group_info=True, # 是否支持群组信息 + supports_member_list=True, # 是否支持成员列表 +) +``` + +### 2. 实现消息获取 + +```python +async def fetch_messages( + self, + group_id: str, + days: int = 1, + max_count: int = 1000, + before_id: Optional[str] = None, +) -> List[UnifiedMessage]: + """ + 获取群组消息历史 + + 重要： + 1. 过滤掉机器人自己的消息 + 2. 只返回指定时间范围内的消息 + 3. 按时间升序排序 + """ + # 你的实现... +``` + +### 3. 实现消息转换 + +将平台原生消息转换为 `UnifiedMessage`： + +```python +def _convert_message(self, raw_msg: Any, group_id: str) -> Optional[UnifiedMessage]: + """将平台原生消息转换为统一格式""" + contents = [] + + # 处理文本 + if raw_msg.text: + contents.append(MessageContent( + type=MessageContentType.TEXT, + text=raw_msg.text + )) + + # 处理图片 + for img in raw_msg.images: + contents.append(MessageContent( + type=MessageContentType.IMAGE, + url=img.url + )) + + return UnifiedMessage( + message_id=str(raw_msg.id), + sender_id=str(raw_msg.author.id), + sender_name=raw_msg.author.name, + group_id=group_id, + text_content=raw_msg.text or "", + contents=tuple(contents), + timestamp=int(raw_msg.created_at.timestamp()), + platform="your_platform", + ) +``` + +### 4. 实现原生格式转换 + +用于向后兼容现有分析器： + +```python +def convert_to_raw_format(self, messages: List[UnifiedMessage]) -> List[dict]: + """将统一消息转换回平台原生格式""" + return [ + { + "id": msg.message_id, + "author": {"id": msg.sender_id, "name": msg.sender_name}, + "content": msg.text_content, + "timestamp": msg.timestamp, + # ... 平台特定字段 + } + for msg in messages + ] +``` + +### 5. 实现消息发送 + +```python +async def send_text(self, group_id: str, text: str, reply_to: Optional[str] = None) -> bool: + """发送文本消息""" + try: + # 调用平台 API 发送消息 + await self.bot.send_message(group_id, text) + return True + except Exception: + return False + +async def send_image(self, group_id: str, image_path: str, caption: str = "") -> bool: + """发送图片消息""" + # 你的实现... + +async def send_file(self, group_id: str, file_path: str, filename: Optional[str] = None) -> bool: + """发送文件""" + # 你的实现... +``` + +### 6. 实现群组信息获取 + +```python +async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]: + """获取群组信息""" + # 你的实现... + +async def get_member_list(self, group_id: str) -> List[UnifiedMember]: + """获取群组成员列表""" + # 你的实现... +``` + +### 7. 实现头像获取 + +```python +async def get_user_avatar_url(self, user_id: str, size: int = 100) -> Optional[str]: + """获取用户头像 URL""" + # 返回头像 URL 或 None + +async def batch_get_avatar_urls(self, user_ids: List[str], size: int = 100) -> Dict[str, Optional[str]]: + """批量获取用户头像 URL""" + return {uid: await self.get_user_avatar_url(uid, size) for uid in user_ids} +``` + +--- + +## 接口说明 + +### PlatformAdapter 必须实现的方法 + +| 方法 | 说明 | 返回类型 | +|------|------|----------| +| `_init_capabilities()` | 初始化平台能力 | `PlatformCapabilities` | +| `fetch_messages()` | 获取消息历史 | `List[UnifiedMessage]` | +| `convert_to_raw_format()` | 转换为原生格式 | `List[dict]` | +| `send_text()` | 发送文本 | `bool` | +| `send_image()` | 发送图片 | `bool` | +| `send_file()` | 发送文件 | `bool` | +| `get_group_info()` | 获取群组信息 | `Optional[UnifiedGroup]` | +| `get_group_list()` | 获取群组列表 | `List[str]` | +| `get_member_list()` | 获取成员列表 | `List[UnifiedMember]` | +| `get_member_info()` | 获取成员信息 | `Optional[UnifiedMember]` | +| `get_user_avatar_url()` | 获取头像 URL | `Optional[str]` | +| `get_user_avatar_data()` | 获取头像 Base64 | `Optional[str]` | +| `get_group_avatar_url()` | 获取群头像 URL | `Optional[str]` | +| `batch_get_avatar_urls()` | 批量获取头像 | `Dict[str, Optional[str]]` | + +### UnifiedMessage 字段 + +| 字段 | 类型 | 说明 | +|------|------|------| +| `message_id` | `str` | 消息唯一 ID | +| `sender_id` | `str` | 发送者 ID | +| `sender_name` | `str` | 发送者昵称 | +| `sender_card` | `Optional[str]` | 发送者群名片 | +| `group_id` | `str` | 群组 ID | +| `text_content` | `str` | 纯文本内容 | +| `contents` | `Tuple[MessageContent, ...]` | 消息内容列表 | +| `timestamp` | `int` | Unix 时间戳 | +| `platform` | `str` | 平台标识 | +| `reply_to_id` | `Optional[str]` | 回复的消息 ID | + +### MessageContentType 枚举 + +| 类型 | 说明 | +|------|------| +| `TEXT` | 文本 | +| `IMAGE` | 图片 | +| `AT` | @某人 | +| `EMOJI` | 表情 | +| `REPLY` | 回复 | +| `FORWARD` | 转发 | +| `VOICE` | 语音 | +| `VIDEO` | 视频 | +| `FILE` | 文件 | +| `UNKNOWN` | 未知类型 | + +--- + +## 最佳实践 + +### 1. 不要硬编码平台特定逻辑 + +❌ **错误做法**： +```python +# 在应用层硬编码平台判断 +if platform == "qq": + messages = fetch_qq_messages() +elif platform == "discord": + messages = fetch_discord_messages() +``` + +✅ **正确做法**： +```python +# 使用适配器模式 +adapter = PlatformAdapterFactory.create(platform_name, bot_instance, config) +messages = await adapter.fetch_messages(group_id, days, max_count) +``` + +### 2. 使用中文注释 + +所有代码注释必须使用中文： + +```python +def fetch_messages(self, group_id: str, days: int = 1) -> List[UnifiedMessage]: + """ + 获取群组消息历史 + + 参数： + group_id: 群组 ID + days: 获取多少天内的消息 + + 返回： + UnifiedMessage 列表 + """ +``` + +### 3. 优雅处理异常 + +```python +async def fetch_messages(self, ...) -> List[UnifiedMessage]: + try: + # 正常逻辑 + return messages + except SpecificError as e: + logger.warning(f"获取消息失败: {e}") + return [] + except Exception: + # 不要让异常传播到上层 + return [] +``` + +### 4. 过滤机器人自己的消息 + +```python +# 在 __init__ 中保存机器人 ID +self.bot_user_id = config.get("bot_user_id", "") + +# 在 fetch_messages 中过滤 +if str(msg.author.id) == self.bot_user_id: + continue +``` + +### 5. 声明正确的平台能力 + +如果平台不支持某功能，在 `PlatformCapabilities` 中正确声明： + +```python +PlatformCapabilities( + supports_history_fetch=False, # 不支持历史消息获取 + max_history_days=0, # 无法获取历史消息 +) +``` + +--- + +## 示例代码 + +完整的适配器示例请参考： + +- **OneBot 适配器**（QQ）：`src/infrastructure/platform/adapters/onebot_adapter.py` +- **Discord 适配器**（骨架）：`src/infrastructure/platform/adapters/discord_adapter.py` + +--- + +## 测试指南 + +### 1. 单元测试 + +为适配器编写单元测试： + +```python +# tests/unit/infrastructure/platform/test_your_adapter.py + +import pytest +from src.infrastructure.platform.adapters.your_platform_adapter import YourPlatformAdapter + +class TestYourPlatformAdapter: + def test_init_capabilities(self): + adapter = YourPlatformAdapter(mock_bot, {}) + caps = adapter.get_capabilities() + assert caps.platform_name == "your_platform" + assert caps.supports_history_fetch == True + + @pytest.mark.asyncio + async def test_fetch_messages(self): + adapter = YourPlatformAdapter(mock_bot, {}) + messages = await adapter.fetch_messages("group_123", days=1) + assert isinstance(messages, list) +``` + +### 2. Docker 容器内验证 + +在 Docker 容器内验证适配器注册： + +```bash +docker exec astrbot python -c " +from data.plugins.astrbot_plugin_qq_group_daily_analysis.src.infrastructure.platform import PlatformAdapterFactory +print('支持的平台:', PlatformAdapterFactory.get_supported_platforms()) +print('Discord 支持:', PlatformAdapterFactory.is_supported('discord')) +" +``` + +### 3. 集成测试 + +确保适配器与 `AnalysisOrchestrator` 正确集成： + +```python +from src.application.analysis_orchestrator import AnalysisOrchestrator + +orchestrator = AnalysisOrchestrator.create_for_platform( + platform_name="your_platform", + bot_instance=bot, + config={}, +) +assert orchestrator is not None +assert orchestrator.can_analyze() == True +``` + +--- + +## 常见问题 + +### Q: 如何处理平台特定的消息类型？ + +使用 `MessageContentType.UNKNOWN` 并在 `raw_data` 中保存原始数据： + +```python +contents.append(MessageContent( + type=MessageContentType.UNKNOWN, + raw_data={"platform_specific_type": "sticker", "data": sticker_data} +)) +``` + +### Q: 如何支持分页获取消息？ + +使用 `before_id` 参数： + +```python +async def fetch_messages(self, ..., before_id: Optional[str] = None): + if before_id: + # 从此消息 ID 之前开始获取 + messages = await api.get_history(before=before_id, limit=max_count) + else: + messages = await api.get_history(limit=max_count) +``` + +### Q: 如何处理不支持的功能？ + +在能力声明中标记为不支持，并在方法中返回空/默认值： + +```python +# 能力声明 +PlatformCapabilities(supports_member_list=False) + +# 方法实现 +async def get_member_list(self, group_id: str) -> List[UnifiedMember]: + return [] # 平台不支持，返回空列表 +``` + +--- + +## 贡献检查清单 + +在提交 PR 之前，请确保： + +- [ ] 适配器继承自 `PlatformAdapter` +- [ ] 实现了所有抽象方法 +- [ ] 在工厂中注册了适配器 +- [ ] 所有注释使用中文 +- [ ] 编写了单元测试 +- [ ] 在 Docker 容器内验证通过 +- [ ] 更新了相关文档 + +--- + +*最后更新：2026-02-08* diff --git a/tests/unit_test_platform.py b/tests/unit_test_platform.py new file mode 100644 index 0000000..61f1e79 --- /dev/null +++ b/tests/unit_test_platform.py @@ -0,0 +1,102 @@ +import unittest +import sys +import os +from unittest.mock import MagicMock, AsyncMock + +# Add plugin root to path so we can import src +# Assuming this file is in tests/ +plugin_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if plugin_root not in sys.path: + sys.path.insert(0, plugin_root) + +# Add AstrBot root to path so we can import astrbot.api +astrbot_root = os.path.abspath(os.path.join(plugin_root, "../../../")) +if astrbot_root not in sys.path: + sys.path.insert(0, astrbot_root) + +print(f"Added to sys.path: {plugin_root}, {astrbot_root}") + +try: + from src.infrastructure.platform.factory import PlatformAdapterFactory + from src.infrastructure.platform.base import PlatformAdapter + from src.infrastructure.platform.adapters.discord_adapter import DiscordAdapter, DISCORD_CAPABILITIES + from src.application.analysis_orchestrator import AnalysisOrchestrator + from src.domain.value_objects.unified_message import UnifiedMessage +except ImportError as e: + print(f"Import Error: {e}") + print(f"sys.path: {sys.path}") + # Try importing as package + try: + from astrbot_plugin_qq_group_daily_analysis.src.infrastructure.platform.factory import PlatformAdapterFactory + # ... and others + except ImportError: + pass + raise e + +class TestPlatformArchitecture(unittest.TestCase): + + def setUp(self): + # Reset factory for isolation if needed, though hard with class methods + pass + + def test_discord_adapter_capabilities(self): + """测试 Discord 适配器能力配置""" + adapter = DiscordAdapter(bot_instance=MagicMock(), config={"bot_user_id": "123"}) + caps = adapter.get_capabilities() + + self.assertEqual(caps.platform_name, "discord") + self.assertTrue(caps.supports_message_history) + self.assertTrue(caps.supports_image_message) + # Check correct attribute name and value (30 from predefined caps) + self.assertEqual(caps.max_message_history_days, 30) + + def test_factory_registration(self): + """测试适配器工厂注册机制""" + # Discord should be registered by import + self.assertTrue(PlatformAdapterFactory.is_supported("discord")) + self.assertTrue(PlatformAdapterFactory.is_supported("aiocqhttp")) + + # Test creation + adapter = PlatformAdapterFactory.create("discord", MagicMock(), {}) + self.assertIsInstance(adapter, DiscordAdapter) + + def test_orchestrator_raw_conversion(self): + """测试编排器的原始格式转换 (验证硬编码移除)""" + # Mock adapter + mock_adapter = MagicMock(spec=PlatformAdapter) + mock_adapter.get_capabilities.return_value = DISCORD_CAPABILITIES + + # Mock fetch_messages return + mock_msg = UnifiedMessage( + message_id="1", sender_id="u1", sender_name="User", + group_id="g1", text_content="test", contents=[], + timestamp=1234567890, platform="discord" + ) + mock_adapter.fetch_messages = AsyncMock(return_value=[mock_msg]) + + # Mock convert_to_raw_format + mock_adapter.convert_to_raw_format.return_value = [{"id": "1", "content": "raw"}] + + # Create orchestrator + orchestrator = AnalysisOrchestrator(adapter=mock_adapter) + + # Run sync wrapper for async method (simplified for unit test structure) + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + # Call fetch_messages_as_raw + raw_msgs = loop.run_until_complete( + orchestrator.fetch_messages_as_raw("g1") + ) + + # Verify result + self.assertEqual(len(raw_msgs), 1) + self.assertEqual(raw_msgs[0]["content"], "raw") + + # Verify adapter method was called (PROVING adapter pattern is used) + mock_adapter.convert_to_raw_format.assert_called_once() + loop.close() + +if __name__ == "__main__": + unittest.main()