From c1d3bf5ecec849e5d3518435393483ae9f6bb9b6 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Sun, 8 Feb 2026 14:13:09 +0800 Subject: [PATCH] feat: add DDD architecture layers (domain, infrastructure, application) - Domain layer: UnifiedMessage, PlatformCapabilities, repository interfaces - Infrastructure layer: PlatformAdapter base, OneBotAdapter, factory - Application layer: AnalysisOrchestrator, MessageConverter This provides cross-platform abstraction for group analysis plugin. --- src/application/__init__.py | 5 + src/application/analysis_orchestrator.py | 227 ++++++++++ src/application/message_converter.py | 195 +++++++++ src/domain/__init__.py | 1 + src/domain/entities/__init__.py | 5 + src/domain/entities/analysis_result.py | 115 ++++++ src/domain/entities/analysis_task.py | 70 ++++ src/domain/repositories/__init__.py | 10 + src/domain/repositories/avatar_repository.py | 79 ++++ src/domain/repositories/message_repository.py | 113 +++++ src/domain/value_objects/__init__.py | 14 + .../value_objects/platform_capabilities.py | 188 +++++++++ src/domain/value_objects/unified_group.py | 33 ++ src/domain/value_objects/unified_message.py | 109 +++++ src/infrastructure/__init__.py | 1 + src/infrastructure/platform/__init__.py | 5 + .../platform/adapters/__init__.py | 4 + .../platform/adapters/onebot_adapter.py | 388 ++++++++++++++++++ src/infrastructure/platform/base.py | 54 +++ src/infrastructure/platform/factory.py | 74 ++++ 20 files changed, 1690 insertions(+) create mode 100644 src/application/__init__.py create mode 100644 src/application/analysis_orchestrator.py create mode 100644 src/application/message_converter.py create mode 100644 src/domain/__init__.py create mode 100644 src/domain/entities/__init__.py create mode 100644 src/domain/entities/analysis_result.py create mode 100644 src/domain/entities/analysis_task.py create mode 100644 src/domain/repositories/__init__.py create mode 100644 src/domain/repositories/avatar_repository.py create mode 100644 src/domain/repositories/message_repository.py create mode 100644 src/domain/value_objects/__init__.py create mode 100644 src/domain/value_objects/platform_capabilities.py create mode 100644 src/domain/value_objects/unified_group.py create mode 100644 src/domain/value_objects/unified_message.py create mode 100644 src/infrastructure/__init__.py create mode 100644 src/infrastructure/platform/__init__.py create mode 100644 src/infrastructure/platform/adapters/__init__.py create mode 100644 src/infrastructure/platform/adapters/onebot_adapter.py create mode 100644 src/infrastructure/platform/base.py create mode 100644 src/infrastructure/platform/factory.py diff --git a/src/application/__init__.py b/src/application/__init__.py new file mode 100644 index 0000000..6426f74 --- /dev/null +++ b/src/application/__init__.py @@ -0,0 +1,5 @@ +# Application Layer - Orchestration and Use Cases +from .analysis_orchestrator import AnalysisOrchestrator +from .message_converter import MessageConverter + +__all__ = ["AnalysisOrchestrator", "MessageConverter"] diff --git a/src/application/analysis_orchestrator.py b/src/application/analysis_orchestrator.py new file mode 100644 index 0000000..ad5fa2d --- /dev/null +++ b/src/application/analysis_orchestrator.py @@ -0,0 +1,227 @@ +""" +Analysis Orchestrator - Application layer coordinator + +This orchestrator bridges the new DDD architecture with the existing +analysis logic, providing a gradual migration path. + +Architecture Decision: +- The orchestrator uses PlatformAdapter for message fetching (new DDD way) +- But delegates to existing analyzers for LLM analysis (preserving working code) +- MessageConverter provides bidirectional conversion for compatibility +""" + +from typing import Optional, List, Dict, Any +from dataclasses import dataclass + +from astrbot.api import logger + +from ..domain.value_objects.unified_message import UnifiedMessage +from ..domain.value_objects.platform_capabilities import PlatformCapabilities +from ..infrastructure.platform import PlatformAdapter, PlatformAdapterFactory +from .message_converter import MessageConverter + + +@dataclass +class AnalysisConfig: + """Configuration for analysis operation""" + days: int = 1 + max_messages: int = 1000 + min_messages_threshold: int = 10 + output_format: str = "image" + + +class AnalysisOrchestrator: + """ + Analysis orchestrator - coordinates the analysis workflow. + + Responsibilities: + 1. Use PlatformAdapter to fetch messages (DDD approach) + 2. Convert messages for compatibility with existing analyzers + 3. Coordinate analysis flow + 4. Provide platform capability checks + + This class serves as the bridge between: + - New DDD infrastructure (PlatformAdapter, UnifiedMessage) + - Existing analysis logic (MessageHandler, LLMAnalyzer, etc.) + """ + + def __init__( + self, + adapter: PlatformAdapter, + config: AnalysisConfig = None, + ): + """ + Initialize the orchestrator. + + Args: + adapter: Platform adapter for message operations + config: Analysis configuration + """ + self.adapter = adapter + self.config = config or AnalysisConfig() + + @classmethod + def create_for_platform( + cls, + platform_name: str, + bot_instance: Any, + config: dict = None, + analysis_config: AnalysisConfig = None, + ) -> Optional["AnalysisOrchestrator"]: + """ + Factory method to create orchestrator for a specific platform. + + Args: + platform_name: Platform name (e.g., "aiocqhttp", "telegram") + bot_instance: Bot instance from AstrBot + config: Platform-specific config + analysis_config: Analysis configuration + + Returns: + AnalysisOrchestrator or None if platform not supported + """ + adapter = PlatformAdapterFactory.create(platform_name, bot_instance, config) + if adapter is None: + logger.warning(f"Platform '{platform_name}' not supported for analysis") + return None + + return cls(adapter, analysis_config) + + def get_capabilities(self) -> PlatformCapabilities: + """Get platform capabilities.""" + return self.adapter.get_capabilities() + + def can_analyze(self) -> bool: + """Check if the platform supports analysis.""" + return self.adapter.get_capabilities().can_analyze() + + def can_send_report(self, format: str = "image") -> bool: + """Check if the platform can send reports in the specified format.""" + return self.adapter.get_capabilities().can_send_report(format) + + async def fetch_messages( + self, + group_id: str, + days: int = None, + max_count: int = None, + ) -> List[UnifiedMessage]: + """ + Fetch messages using the platform adapter. + + Args: + group_id: Group ID to fetch messages from + days: Number of days (defaults to config) + max_count: Maximum message count (defaults to config) + + Returns: + List of UnifiedMessage + """ + days = days or self.config.days + max_count = max_count or self.config.max_messages + + # Apply platform capability limits + caps = self.adapter.get_capabilities() + effective_days = caps.get_effective_days(days) + effective_count = caps.get_effective_count(max_count) + + if effective_days < days: + logger.info( + f"Platform limits: requested {days} days, " + f"using {effective_days} days" + ) + + return await self.adapter.fetch_messages( + group_id=group_id, + days=effective_days, + max_count=effective_count, + ) + + async def fetch_messages_as_raw( + self, + group_id: str, + days: int = None, + max_count: int = None, + ) -> List[dict]: + """ + Fetch messages and convert to raw dict format. + + This provides backward compatibility with existing analyzers + that expect raw dict messages. + + Args: + group_id: Group ID to fetch messages from + days: Number of days + max_count: Maximum message count + + Returns: + List of raw message dicts (OneBot format) + """ + unified_messages = await self.fetch_messages(group_id, days, max_count) + return MessageConverter.batch_to_onebot(unified_messages) + + async def get_group_info(self, group_id: str): + """Get group information.""" + return await self.adapter.get_group_info(group_id) + + async def get_member_avatars( + self, + user_ids: List[str], + size: int = 100, + ) -> Dict[str, Optional[str]]: + """ + Batch get user avatar URLs. + + Args: + user_ids: List of user IDs + size: Avatar size + + Returns: + Dict mapping user_id to avatar URL (or None) + """ + return await self.adapter.batch_get_avatar_urls(user_ids, size) + + async def send_text(self, group_id: str, text: str) -> bool: + """Send text message to group.""" + return await self.adapter.send_text(group_id, text) + + async def send_image( + self, + group_id: str, + image_path: str, + caption: str = "", + ) -> bool: + """Send image to group.""" + return await self.adapter.send_image(group_id, image_path, caption) + + async def send_file( + self, + group_id: str, + file_path: str, + filename: str = None, + ) -> bool: + """Send file to group.""" + return await self.adapter.send_file(group_id, file_path, filename) + + def validate_message_count(self, messages: List[UnifiedMessage]) -> bool: + """ + Check if message count meets minimum threshold. + + Args: + messages: List of messages + + Returns: + True if count is sufficient + """ + return len(messages) >= self.config.min_messages_threshold + + def get_analysis_text(self, messages: List[UnifiedMessage]) -> str: + """ + Convert messages to analysis text format for LLM. + + Args: + messages: List of UnifiedMessage + + Returns: + Formatted text for LLM analysis + """ + return MessageConverter.unified_to_analysis_text(messages) diff --git a/src/application/message_converter.py b/src/application/message_converter.py new file mode 100644 index 0000000..ba80ca7 --- /dev/null +++ b/src/application/message_converter.py @@ -0,0 +1,195 @@ +""" +Message Converter - Bridges raw platform messages to UnifiedMessage + +This module provides backward compatibility by converting between +raw platform message formats and the new UnifiedMessage format. +""" + +from typing import List, Dict, Any, Optional +from datetime import datetime + +from ..domain.value_objects.unified_message import ( + UnifiedMessage, + MessageContent, + MessageContentType, +) + + +class MessageConverter: + """ + Converts between raw platform messages and UnifiedMessage format. + + This provides a migration path: existing code can continue using + raw dicts while new code uses UnifiedMessage. + """ + + @staticmethod + def from_onebot_message(raw_msg: dict, group_id: str) -> Optional[UnifiedMessage]: + """ + Convert OneBot v11 raw message to UnifiedMessage. + + Args: + raw_msg: Raw message dict from OneBot API + group_id: Group ID + + Returns: + UnifiedMessage or None if conversion fails + """ + try: + sender = raw_msg.get("sender", {}) + message_chain = raw_msg.get("message", []) + + # Handle string message format + 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 + )) + + # Extract reply_to from contents + 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: + """ + Convert UnifiedMessage back to OneBot v11 raw format. + + For backward compatibility with existing code that expects raw dicts. + """ + 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}}) + + return { + "message_id": unified.message_id, + "sender": { + "user_id": unified.sender_id, + "nickname": unified.sender_name, + "card": unified.sender_card or "", + }, + "group_id": unified.group_id, + "message": message_chain, + "time": unified.timestamp, + } + + @staticmethod + def batch_from_onebot(raw_messages: List[dict], group_id: str) -> List[UnifiedMessage]: + """Convert a batch of OneBot messages to UnifiedMessage list.""" + 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]: + """Convert a batch of UnifiedMessage to OneBot raw format.""" + return [MessageConverter.to_onebot_message(msg) for msg in unified_messages] + + @staticmethod + def unified_to_analysis_text(messages: List[UnifiedMessage]) -> str: + """ + Convert UnifiedMessage list to analysis text format for LLM. + + This is the format expected by the existing LLM analyzers. + """ + lines = [] + for msg in messages: + if msg.has_text(): + lines.append(msg.to_analysis_format()) + return "\n".join(lines) diff --git a/src/domain/__init__.py b/src/domain/__init__.py new file mode 100644 index 0000000..5a7872d --- /dev/null +++ b/src/domain/__init__.py @@ -0,0 +1 @@ +# Domain Layer - Platform-agnostic business logic diff --git a/src/domain/entities/__init__.py b/src/domain/entities/__init__.py new file mode 100644 index 0000000..8e49949 --- /dev/null +++ b/src/domain/entities/__init__.py @@ -0,0 +1,5 @@ +# Domain Entities +from .analysis_task import AnalysisTask, TaskStatus +from .analysis_result import GroupAnalysisResult + +__all__ = ["AnalysisTask", "TaskStatus", "GroupAnalysisResult"] diff --git a/src/domain/entities/analysis_result.py b/src/domain/entities/analysis_result.py new file mode 100644 index 0000000..6904ae2 --- /dev/null +++ b/src/domain/entities/analysis_result.py @@ -0,0 +1,115 @@ +""" +Group Analysis Result Entity +""" + +from dataclasses import dataclass, field +from typing import List, Optional +import uuid +import time + + +@dataclass +class SummaryTopic: + """Topic summary""" + topic: str + contributors: List[str] + detail: str + + +@dataclass +class UserTitle: + """User title/portrait""" + name: str + user_id: str + title: str + mbti: str + reason: str + avatar_url: Optional[str] = None + avatar_data: Optional[str] = None + + +@dataclass +class GoldenQuote: + """Golden quote""" + content: str + sender: str + reason: str + user_id: str = "" + avatar_url: Optional[str] = None + + +@dataclass +class TokenUsage: + """Token usage statistics""" + prompt_tokens: int = 0 + completion_tokens: int = 0 + total_tokens: int = 0 + + +@dataclass +class EmojiStatistics: + """Emoji statistics""" + face_count: int = 0 + mface_count: int = 0 + bface_count: int = 0 + sface_count: int = 0 + other_emoji_count: int = 0 + face_details: dict = field(default_factory=dict) + + @property + def total_emoji_count(self) -> int: + return ( + self.face_count + + self.mface_count + + self.bface_count + + self.sface_count + + self.other_emoji_count + ) + + +@dataclass +class ActivityVisualization: + """Activity visualization data""" + hourly_activity: dict = field(default_factory=dict) + daily_activity: dict = field(default_factory=dict) + user_activity_ranking: list = field(default_factory=list) + peak_hours: list = field(default_factory=list) + activity_heatmap_data: dict = field(default_factory=dict) + + +@dataclass +class GroupStatistics: + """Group statistics""" + message_count: int = 0 + total_characters: int = 0 + participant_count: int = 0 + most_active_period: str = "" + emoji_count: int = 0 + emoji_statistics: EmojiStatistics = field(default_factory=EmojiStatistics) + activity_visualization: ActivityVisualization = field(default_factory=ActivityVisualization) + + +@dataclass +class GroupAnalysisResult: + """Group analysis result entity""" + id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) + group_id: str = "" + group_name: str = "" + trace_id: str = "" + platform: str = "" + + # Analysis results + message_count: int = 0 + statistics: GroupStatistics = field(default_factory=GroupStatistics) + topics: List[SummaryTopic] = field(default_factory=list) + user_titles: List[UserTitle] = field(default_factory=list) + golden_quotes: List[GoldenQuote] = field(default_factory=list) + + # Metadata + token_usage: TokenUsage = field(default_factory=TokenUsage) + analysis_date: str = "" + created_at: float = field(default_factory=time.time) + + def has_content(self) -> bool: + """Check if result has any analysis content""" + return bool(self.topics or self.user_titles or self.golden_quotes) diff --git a/src/domain/entities/analysis_task.py b/src/domain/entities/analysis_task.py new file mode 100644 index 0000000..818a797 --- /dev/null +++ b/src/domain/entities/analysis_task.py @@ -0,0 +1,70 @@ +""" +Analysis Task Entity - Aggregate Root +""" + +from dataclasses import dataclass, field +from enum import Enum +from typing import Optional +import time +import 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: + """Analysis task entity - Aggregate root""" + 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, can_analyze: bool) -> bool: + """Start task, validate platform capability""" + if not can_analyze: + self.status = TaskStatus.UNSUPPORTED_PLATFORM + self.error_message = f"Platform {self.platform_name} does not support analysis" + return False + self.status = TaskStatus.FETCHING_MESSAGES + self.started_at = time.time() + return True + + def advance_to(self, status: TaskStatus): + """Advance to next status""" + self.status = status + + def complete(self, result_id: str): + """Mark task as completed""" + self.status = TaskStatus.COMPLETED + self.result_id = result_id + self.completed_at = time.time() + + def fail(self, error: str): + """Mark task as failed""" + self.status = TaskStatus.FAILED + self.error_message = error + self.completed_at = time.time() + + @property + def duration(self) -> Optional[float]: + """Get task duration in seconds""" + if self.started_at and self.completed_at: + return self.completed_at - self.started_at + return None diff --git a/src/domain/repositories/__init__.py b/src/domain/repositories/__init__.py new file mode 100644 index 0000000..89854a7 --- /dev/null +++ b/src/domain/repositories/__init__.py @@ -0,0 +1,10 @@ +# Repository Interfaces +from .message_repository import IMessageRepository, IMessageSender, IGroupInfoRepository +from .avatar_repository import IAvatarRepository + +__all__ = [ + "IMessageRepository", + "IMessageSender", + "IGroupInfoRepository", + "IAvatarRepository", +] diff --git a/src/domain/repositories/avatar_repository.py b/src/domain/repositories/avatar_repository.py new file mode 100644 index 0000000..3ade4a9 --- /dev/null +++ b/src/domain/repositories/avatar_repository.py @@ -0,0 +1,79 @@ +""" +Avatar Repository Interface - Cross-platform avatar abstraction +""" + +from abc import ABC, abstractmethod +from typing import List, Optional, Dict + + +class IAvatarRepository(ABC): + """ + Avatar repository interface + + Different platforms have different ways to get avatars: + - QQ/OneBot: URL template (q1.qlogo.cn) + - Telegram: API call (getUserProfilePhotos + getFile) + - Discord: CDN URL template (cdn.discordapp.com) + - Slack: users.info API profile.image_* fields + """ + + @abstractmethod + async def get_user_avatar_url( + self, + user_id: str, + size: int = 100, + ) -> Optional[str]: + """ + Get user avatar URL + + Args: + user_id: User ID + size: Desired avatar size (will pick closest available) + + Returns: + Avatar URL, or None if unavailable + """ + pass + + @abstractmethod + async def get_user_avatar_data( + self, + user_id: str, + size: int = 100, + ) -> Optional[str]: + """ + Get user avatar as Base64 data + + For scenarios needing embedded images (e.g., HTML template rendering) + + Returns: + Base64 encoded image data (data:image/png;base64,...), + or None if unavailable + """ + pass + + @abstractmethod + async def get_group_avatar_url( + self, + group_id: str, + size: int = 100, + ) -> Optional[str]: + """Get group avatar URL""" + pass + + @abstractmethod + async def batch_get_avatar_urls( + self, + user_ids: List[str], + size: int = 100, + ) -> Dict[str, Optional[str]]: + """ + Batch get user avatar URLs + + For report generation needing multiple avatars at once + """ + pass + + def get_default_avatar_url(self) -> str: + """Get default avatar URL (when user avatar unavailable)""" + return "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCI+PHBhdGggZD0iTTEyIDEyYzIuMjEgMCA0LTEuNzkgNC00cy0xLjc5LTQtNC00LTQgMS43OS00IDQgMS43OSA0IDQgNHptMCAyYy0yLjY3IDAtOCAxLjM0LTggNHYyaDE2di0yYzAtMi42Ni01LjMzLTQtOC00eiIvPjwvc3ZnPg==" diff --git a/src/domain/repositories/message_repository.py b/src/domain/repositories/message_repository.py new file mode 100644 index 0000000..a58343f --- /dev/null +++ b/src/domain/repositories/message_repository.py @@ -0,0 +1,113 @@ +""" +Message Repository Interfaces - Platform-agnostic abstractions +""" + +from abc import ABC, abstractmethod +from typing import List, Optional, Dict + +from ..value_objects.unified_message import UnifiedMessage +from ..value_objects.platform_capabilities import PlatformCapabilities +from ..value_objects.unified_group import UnifiedGroup, UnifiedMember + + +class IMessageRepository(ABC): + """ + Message repository interface + + Each platform adapter must implement this interface. + All methods return unified format, hiding platform differences. + """ + + @abstractmethod + async def fetch_messages( + self, + group_id: str, + days: int = 1, + max_count: int = 1000, + before_id: Optional[str] = None, + ) -> List[UnifiedMessage]: + """ + Fetch group message history + + Args: + group_id: Group ID + days: Fetch messages from last N days + max_count: Maximum message count + before_id: Fetch messages before this ID (for pagination) + + Returns: + List of unified messages, sorted by time ascending + """ + pass + + @abstractmethod + def get_capabilities(self) -> PlatformCapabilities: + """Get platform capabilities""" + pass + + @abstractmethod + def get_platform_name(self) -> str: + """Get platform name""" + pass + + +class IMessageSender(ABC): + """Message sender interface""" + + @abstractmethod + async def send_text( + self, + group_id: str, + text: str, + reply_to: Optional[str] = None, + ) -> bool: + """Send text message""" + pass + + @abstractmethod + async def send_image( + self, + group_id: str, + image_path: str, + caption: str = "", + ) -> bool: + """Send image message""" + pass + + @abstractmethod + async def send_file( + self, + group_id: str, + file_path: str, + filename: Optional[str] = None, + ) -> bool: + """Send file""" + pass + + +class IGroupInfoRepository(ABC): + """Group info repository interface""" + + @abstractmethod + async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]: + """Get group information""" + pass + + @abstractmethod + async def get_group_list(self) -> List[str]: + """Get all group IDs the bot is in""" + pass + + @abstractmethod + async def get_member_list(self, group_id: str) -> List[UnifiedMember]: + """Get group member list""" + pass + + @abstractmethod + async def get_member_info( + self, + group_id: str, + user_id: str, + ) -> Optional[UnifiedMember]: + """Get specific member info""" + pass diff --git a/src/domain/value_objects/__init__.py b/src/domain/value_objects/__init__.py new file mode 100644 index 0000000..b7c0182 --- /dev/null +++ b/src/domain/value_objects/__init__.py @@ -0,0 +1,14 @@ +# Value Objects +from .unified_message import UnifiedMessage, MessageContent, MessageContentType +from .platform_capabilities import PlatformCapabilities, PLATFORM_CAPABILITIES +from .unified_group import UnifiedGroup, UnifiedMember + +__all__ = [ + "UnifiedMessage", + "MessageContent", + "MessageContentType", + "PlatformCapabilities", + "PLATFORM_CAPABILITIES", + "UnifiedGroup", + "UnifiedMember", +] diff --git a/src/domain/value_objects/platform_capabilities.py b/src/domain/value_objects/platform_capabilities.py new file mode 100644 index 0000000..5fed2df --- /dev/null +++ b/src/domain/value_objects/platform_capabilities.py @@ -0,0 +1,188 @@ +""" +Platform Capabilities Value Object - Runtime decision support + +Each platform adapter declares its capabilities, +application layer decides operations based on capabilities. +""" + +from dataclasses import dataclass +from typing import Optional + + +@dataclass(frozen=True) +class PlatformCapabilities: + """ + Platform capability description + + Design principles: + 1. All fields have default values (most conservative assumption) + 2. Immutable + 3. Provide convenient check methods + """ + # Platform identification + platform_name: str + platform_version: str = "unknown" + + # Message retrieval capabilities + supports_message_history: bool = False + max_message_history_days: int = 0 + max_message_count: int = 0 + supports_message_search: bool = False + + # Group info capabilities + supports_group_list: bool = False + supports_group_info: bool = False + supports_member_list: bool = False + supports_member_info: bool = False + + # Message sending capabilities + supports_text_message: bool = True + supports_image_message: bool = False + supports_file_message: bool = False + supports_forward_message: bool = False + supports_reply_message: bool = False + max_text_length: int = 4096 + max_image_size_mb: float = 10.0 + + # Special capabilities + supports_at_all: bool = False + supports_recall: bool = False + supports_edit: bool = False + + # Avatar capabilities + supports_user_avatar: bool = True + supports_group_avatar: bool = False + avatar_needs_api_call: bool = False + avatar_sizes: tuple = (100,) + + # Check methods + def can_analyze(self) -> bool: + """Whether supports group chat analysis (core capability)""" + return ( + self.supports_message_history + and self.max_message_history_days > 0 + and self.max_message_count > 0 + ) + + def can_send_report(self, format: str = "image") -> bool: + """Whether can send report""" + if format == "text": + return self.supports_text_message + elif format == "image": + return self.supports_image_message + elif format == "pdf": + return self.supports_file_message + return False + + def get_effective_days(self, requested_days: int) -> int: + """Get actual available days""" + return min(requested_days, self.max_message_history_days) + + def get_effective_count(self, requested_count: int) -> int: + """Get actual available message count""" + return min(requested_count, self.max_message_count) + + +# Predefined platform capabilities +ONEBOT_V11_CAPABILITIES = PlatformCapabilities( + platform_name="onebot", + platform_version="v11", + 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_member_info=True, + supports_text_message=True, + supports_image_message=True, + supports_file_message=True, + supports_forward_message=True, + supports_reply_message=True, + max_text_length=4500, + supports_at_all=True, + supports_recall=True, + supports_user_avatar=True, + supports_group_avatar=True, + avatar_needs_api_call=False, + avatar_sizes=(40, 100, 140, 160, 640), +) + +TELEGRAM_CAPABILITIES = PlatformCapabilities( + platform_name="telegram", + platform_version="bot_api_7.x", + supports_message_history=False, + max_message_history_days=0, + max_message_count=0, + supports_group_list=False, + supports_group_info=True, + supports_member_list=True, + supports_text_message=True, + supports_image_message=True, + supports_file_message=True, + supports_reply_message=True, + max_text_length=4096, + max_image_size_mb=50.0, + supports_edit=True, + supports_user_avatar=True, + supports_group_avatar=True, + avatar_needs_api_call=True, + avatar_sizes=(160, 320, 640), +) + +DISCORD_CAPABILITIES = PlatformCapabilities( + platform_name="discord", + platform_version="api_v10", + 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, + supports_reply_message=True, + max_text_length=2000, + max_image_size_mb=8.0, + supports_edit=True, + supports_user_avatar=True, + supports_group_avatar=True, + avatar_needs_api_call=False, + avatar_sizes=(16, 32, 64, 128, 256, 512, 1024, 2048, 4096), +) + +SLACK_CAPABILITIES = PlatformCapabilities( + platform_name="slack", + platform_version="web_api", + supports_message_history=True, + max_message_history_days=90, + max_message_count=1000, + 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_reply_message=True, + max_text_length=40000, + supports_edit=True, + supports_user_avatar=True, + supports_group_avatar=False, + avatar_needs_api_call=True, + avatar_sizes=(24, 32, 48, 72, 192, 512, 1024), +) + +# Capability lookup table +PLATFORM_CAPABILITIES = { + "aiocqhttp": ONEBOT_V11_CAPABILITIES, + "onebot": ONEBOT_V11_CAPABILITIES, + "telegram": TELEGRAM_CAPABILITIES, + "discord": DISCORD_CAPABILITIES, + "slack": SLACK_CAPABILITIES, +} + + +def get_capabilities(platform_name: str) -> Optional[PlatformCapabilities]: + """Get capabilities by platform name""" + return PLATFORM_CAPABILITIES.get(platform_name.lower()) diff --git a/src/domain/value_objects/unified_group.py b/src/domain/value_objects/unified_group.py new file mode 100644 index 0000000..f826332 --- /dev/null +++ b/src/domain/value_objects/unified_group.py @@ -0,0 +1,33 @@ +""" +Unified Group Value Objects - Cross-platform group abstraction +""" + +from dataclasses import dataclass +from typing import Optional + + +@dataclass(frozen=True) +class UnifiedMember: + """Unified member information""" + user_id: str + nickname: str + card: Optional[str] = None # Group card + role: str = "member" # owner, admin, member + join_time: Optional[int] = None + avatar_url: Optional[str] = None + avatar_data: Optional[str] = None # Base64 for template rendering + + def get_display_name(self) -> str: + return self.card or self.nickname or self.user_id + + +@dataclass(frozen=True) +class UnifiedGroup: + """Unified group information""" + group_id: str + group_name: str + member_count: int = 0 + owner_id: Optional[str] = None + create_time: Optional[int] = None + description: Optional[str] = None + platform: str = "unknown" diff --git a/src/domain/value_objects/unified_message.py b/src/domain/value_objects/unified_message.py new file mode 100644 index 0000000..25e1e0b --- /dev/null +++ b/src/domain/value_objects/unified_message.py @@ -0,0 +1,109 @@ +""" +Unified Message Value Object - Cross-platform core abstraction + +All platform messages are converted to this format for analysis. +""" + +from dataclasses import dataclass, field +from typing import Optional, Any, Tuple +from enum import Enum +from datetime import datetime + + +class MessageContentType(Enum): + """Message content type enumeration""" + TEXT = "text" + IMAGE = "image" + FILE = "file" + EMOJI = "emoji" + REPLY = "reply" + FORWARD = "forward" + AT = "at" + VOICE = "voice" + VIDEO = "video" + LOCATION = "location" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class MessageContent: + """ + Message content segment value object + + Immutable, used to compose message chains + """ + type: MessageContentType + text: str = "" + url: str = "" + emoji_id: str = "" + emoji_name: str = "" + at_user_id: str = "" + raw_data: Any = None + + def is_text(self) -> bool: + return self.type == MessageContentType.TEXT + + def is_emoji(self) -> bool: + return self.type == MessageContentType.EMOJI + + +@dataclass(frozen=True) +class UnifiedMessage: + """ + Unified message format - Cross-platform core value object + + Design principles: + 1. Only keep fields needed for analysis + 2. Use platform-agnostic types + 3. Immutable (frozen=True) - thread-safe + 4. All IDs are strings - avoid platform differences + """ + # Basic identification + message_id: str + sender_id: str + sender_name: str + group_id: str + + # Message content + text_content: str # Extracted plain text for LLM analysis + contents: Tuple[MessageContent, ...] = field(default_factory=tuple) + + # Time information + timestamp: int = 0 # Unix timestamp + + # Platform information + platform: str = "unknown" + + # Optional information + reply_to_id: Optional[str] = None + sender_card: Optional[str] = None # Group card/nickname + + # Analysis helper methods + def has_text(self) -> bool: + """Whether has text content""" + return bool(self.text_content.strip()) + + def get_display_name(self) -> str: + """Get display name, prefer group card""" + return self.sender_card or self.sender_name or self.sender_id + + def get_emoji_count(self) -> int: + """Get emoji count""" + return sum(1 for c in self.contents if c.is_emoji()) + + def get_text_length(self) -> int: + """Get text length""" + return len(self.text_content) + + def get_datetime(self) -> datetime: + """Get message datetime""" + return datetime.fromtimestamp(self.timestamp) + + def to_analysis_format(self) -> str: + """Convert to analysis format (for LLM)""" + name = self.get_display_name() + return f"[{name}]: {self.text_content}" + + +# Type alias +MessageList = list[UnifiedMessage] diff --git a/src/infrastructure/__init__.py b/src/infrastructure/__init__.py new file mode 100644 index 0000000..cd05840 --- /dev/null +++ b/src/infrastructure/__init__.py @@ -0,0 +1 @@ +# Infrastructure Layer diff --git a/src/infrastructure/platform/__init__.py b/src/infrastructure/platform/__init__.py new file mode 100644 index 0000000..67ac024 --- /dev/null +++ b/src/infrastructure/platform/__init__.py @@ -0,0 +1,5 @@ +# Platform Adapters +from .factory import PlatformAdapterFactory +from .base import PlatformAdapter + +__all__ = ["PlatformAdapterFactory", "PlatformAdapter"] diff --git a/src/infrastructure/platform/adapters/__init__.py b/src/infrastructure/platform/adapters/__init__.py new file mode 100644 index 0000000..e95a596 --- /dev/null +++ b/src/infrastructure/platform/adapters/__init__.py @@ -0,0 +1,4 @@ +# Platform Adapters +from .onebot_adapter import OneBotAdapter + +__all__ = ["OneBotAdapter"] diff --git a/src/infrastructure/platform/adapters/onebot_adapter.py b/src/infrastructure/platform/adapters/onebot_adapter.py new file mode 100644 index 0000000..68da401 --- /dev/null +++ b/src/infrastructure/platform/adapters/onebot_adapter.py @@ -0,0 +1,388 @@ +""" +OneBot v11 Platform Adapter + +Supports NapCat, go-cqhttp, Lagrange, and other OneBot implementations. +""" + +from datetime import datetime, timedelta +from typing import List, Optional, Any, Dict +import aiohttp +import base64 + +from ....domain.value_objects.unified_message import ( + UnifiedMessage, + MessageContent, + MessageContentType, +) +from ....domain.value_objects.platform_capabilities import ( + PlatformCapabilities, + ONEBOT_V11_CAPABILITIES, +) +from ....domain.value_objects.unified_group import UnifiedGroup, UnifiedMember +from ..base import PlatformAdapter + + +class OneBotAdapter(PlatformAdapter): + """OneBot v11 protocol adapter""" + + # QQ Avatar URL templates + USER_AVATAR_TEMPLATE = "https://q1.qlogo.cn/g?b=qq&nk={user_id}&s={size}" + USER_AVATAR_HD_TEMPLATE = "https://q.qlogo.cn/headimg_dl?dst_uin={user_id}&spec={size}&img_type=jpg" + GROUP_AVATAR_TEMPLATE = "https://p.qlogo.cn/gh/{group_id}/{group_id}/{size}/" + + AVAILABLE_SIZES = [40, 100, 140, 160, 640] + + def __init__(self, bot_instance: Any, config: dict = None): + super().__init__(bot_instance, config) + self.bot_self_ids = [str(id) for id in config.get("bot_qq_ids", [])] if config else [] + + def _init_capabilities(self) -> PlatformCapabilities: + return ONEBOT_V11_CAPABILITIES + + def _get_nearest_size(self, requested_size: int) -> int: + """Get nearest available size""" + return min(self.AVAILABLE_SIZES, key=lambda x: abs(x - requested_size)) + + # ==================== IMessageRepository ==================== + + async def fetch_messages( + self, + group_id: str, + days: int = 1, + max_count: int = 1000, + before_id: Optional[str] = None, + ) -> List[UnifiedMessage]: + """Fetch group message 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, + ) + + 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) + + messages.sort(key=lambda m: m.timestamp) + return messages + + except Exception: + return [] + + def _convert_message(self, raw_msg: dict, group_id: str) -> Optional[UnifiedMessage]: + """Convert OneBot message to unified format""" + 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 = 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 + + # ==================== IMessageSender ==================== + + async def send_text( + self, + group_id: str, + text: str, + reply_to: Optional[str] = None, + ) -> bool: + """Send text message""" + try: + message = [{"type": "text", "data": {"text": text}}] + + if reply_to: + message.insert(0, {"type": "reply", "data": {"id": reply_to}}) + + await self.bot.call_action( + "send_group_msg", + group_id=int(group_id), + message=message, + ) + return True + except Exception: + return False + + async def send_image( + self, + group_id: str, + image_path: str, + caption: str = "", + ) -> bool: + """Send image message""" + try: + message = [] + + if caption: + message.append({"type": "text", "data": {"text": caption}}) + + if image_path.startswith(("http://", "https://")): + file_str = image_path + else: + file_str = f"file:///{image_path}" + + message.append({"type": "image", "data": {"file": file_str}}) + + await self.bot.call_action( + "send_group_msg", + group_id=int(group_id), + message=message, + ) + return True + except Exception: + return False + + async def send_file( + self, + group_id: str, + file_path: str, + filename: Optional[str] = None, + ) -> bool: + """Send file""" + try: + await self.bot.call_action( + "upload_group_file", + group_id=int(group_id), + file=file_path, + name=filename or file_path.split("/")[-1], + ) + return True + except Exception: + return False + + # ==================== IGroupInfoRepository ==================== + + async def get_group_info(self, group_id: str) -> Optional[UnifiedGroup]: + """Get group information""" + try: + result = await self.bot.call_action( + "get_group_info", + group_id=int(group_id), + ) + + if not result: + return None + + return UnifiedGroup( + group_id=str(result.get("group_id", group_id)), + group_name=result.get("group_name", ""), + member_count=result.get("member_count", 0), + owner_id=str(result.get("owner_id", "")) or None, + create_time=result.get("group_create_time"), + platform="onebot", + ) + except Exception: + return None + + async def get_group_list(self) -> List[str]: + """Get all group IDs the bot is in""" + try: + result = await self.bot.call_action("get_group_list") + return [str(g.get("group_id", "")) for g in result or []] + except Exception: + return [] + + async def get_member_list(self, group_id: str) -> List[UnifiedMember]: + """Get group member list""" + try: + result = await self.bot.call_action( + "get_group_member_list", + group_id=int(group_id), + ) + + members = [] + for m in result or []: + members.append(UnifiedMember( + user_id=str(m.get("user_id", "")), + nickname=m.get("nickname", ""), + card=m.get("card", "") or None, + role=m.get("role", "member"), + join_time=m.get("join_time"), + )) + return members + except Exception: + return [] + + async def get_member_info( + self, + group_id: str, + user_id: str, + ) -> Optional[UnifiedMember]: + """Get specific member info""" + try: + result = await self.bot.call_action( + "get_group_member_info", + group_id=int(group_id), + user_id=int(user_id), + ) + + if not result: + return None + + return UnifiedMember( + user_id=str(result.get("user_id", user_id)), + nickname=result.get("nickname", ""), + card=result.get("card", "") or None, + role=result.get("role", "member"), + join_time=result.get("join_time"), + ) + except Exception: + return None + + # ==================== IAvatarRepository ==================== + + async def get_user_avatar_url( + self, + user_id: str, + size: int = 100, + ) -> Optional[str]: + """Get QQ user avatar URL""" + actual_size = self._get_nearest_size(size) + if actual_size >= 640: + return self.USER_AVATAR_HD_TEMPLATE.format(user_id=user_id, size=640) + return self.USER_AVATAR_TEMPLATE.format(user_id=user_id, size=actual_size) + + async def get_user_avatar_data( + self, + user_id: str, + size: int = 100, + ) -> Optional[str]: + """Get QQ user avatar as Base64 data""" + url = await self.get_user_avatar_url(user_id, size) + if not url: + return None + + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp: + if resp.status == 200: + data = await resp.read() + b64 = base64.b64encode(data).decode('utf-8') + content_type = resp.headers.get('Content-Type', 'image/png') + return f"data:{content_type};base64,{b64}" + except Exception: + pass + return None + + async def get_group_avatar_url( + self, + group_id: str, + size: int = 100, + ) -> Optional[str]: + """Get QQ group avatar URL""" + actual_size = self._get_nearest_size(size) + return self.GROUP_AVATAR_TEMPLATE.format(group_id=group_id, size=actual_size) + + async def batch_get_avatar_urls( + self, + user_ids: List[str], + size: int = 100, + ) -> Dict[str, Optional[str]]: + """Batch get QQ user avatar URLs (no API call needed)""" + return { + user_id: await self.get_user_avatar_url(user_id, size) + for user_id in user_ids + } diff --git a/src/infrastructure/platform/base.py b/src/infrastructure/platform/base.py new file mode 100644 index 0000000..26d5a21 --- /dev/null +++ b/src/infrastructure/platform/base.py @@ -0,0 +1,54 @@ +""" +Platform Adapter Base Class +""" + +from abc import ABC, abstractmethod +from typing import Any, Optional, List, Dict + +from ...domain.repositories.message_repository import ( + IMessageRepository, + IMessageSender, + IGroupInfoRepository, +) +from ...domain.repositories.avatar_repository import IAvatarRepository +from ...domain.value_objects.platform_capabilities import PlatformCapabilities +from ...domain.value_objects.unified_message import UnifiedMessage +from ...domain.value_objects.unified_group import UnifiedGroup, UnifiedMember + + +class PlatformAdapter( + IMessageRepository, + IMessageSender, + IGroupInfoRepository, + IAvatarRepository, + ABC +): + """ + Platform adapter base class + + Combines message repository, message sender, group info, and avatar interfaces. + Each platform adapter inherits this class and implements all methods. + """ + + def __init__(self, bot_instance: Any, config: dict = None): + self.bot = bot_instance + self.config = config or {} + self._capabilities: Optional[PlatformCapabilities] = None + + @property + def capabilities(self) -> PlatformCapabilities: + """Platform capabilities (lazy initialization)""" + if self._capabilities is None: + self._capabilities = self._init_capabilities() + return self._capabilities + + @abstractmethod + def _init_capabilities(self) -> PlatformCapabilities: + """Initialize platform capabilities, subclass must implement""" + raise NotImplementedError + + def get_capabilities(self) -> PlatformCapabilities: + return self.capabilities + + def get_platform_name(self) -> str: + return self.capabilities.platform_name diff --git a/src/infrastructure/platform/factory.py b/src/infrastructure/platform/factory.py new file mode 100644 index 0000000..d553615 --- /dev/null +++ b/src/infrastructure/platform/factory.py @@ -0,0 +1,74 @@ +""" +Platform Adapter Factory +""" + +from typing import Optional, Any, Dict, Type + +from .base import PlatformAdapter + + +class PlatformAdapterFactory: + """ + Platform adapter factory + + Creates adapter instances based on platform name. + Uses registry pattern for easy extension. + """ + + _adapters: Dict[str, Type[PlatformAdapter]] = {} + + @classmethod + def register(cls, platform_name: str, adapter_class: Type[PlatformAdapter]): + """Register a new adapter""" + cls._adapters[platform_name.lower()] = adapter_class + + @classmethod + def create( + cls, + platform_name: str, + bot_instance: Any, + config: dict = None, + ) -> Optional[PlatformAdapter]: + """ + Create platform adapter + + Args: + platform_name: Platform name (e.g., "aiocqhttp", "telegram") + bot_instance: AstrBot bot instance + config: Configuration dict + + Returns: + Platform adapter instance, or None if unsupported + """ + adapter_class = cls._adapters.get(platform_name.lower()) + + if adapter_class is None: + return None + + try: + return adapter_class(bot_instance, config) + except Exception: + return None + + @classmethod + def get_supported_platforms(cls) -> list[str]: + """Get all supported platform names""" + return list(cls._adapters.keys()) + + @classmethod + def is_supported(cls, platform_name: str) -> bool: + """Check if platform is supported""" + return platform_name.lower() in cls._adapters + + +# Import adapters to register them +def _register_adapters(): + try: + from .adapters.onebot_adapter import OneBotAdapter + PlatformAdapterFactory.register("aiocqhttp", OneBotAdapter) + PlatformAdapterFactory.register("onebot", OneBotAdapter) + except ImportError: + pass + + +_register_adapters()