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

@@ -0,0 +1,5 @@
+# Application Layer - Orchestration and Use Cases
+from .analysis_orchestrator import AnalysisOrchestrator
+from .message_converter import MessageConverter
+
+__all__ = ["AnalysisOrchestrator", "MessageConverter"]

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)

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)

src/domain/__init__.py

@@ -0,0 +1 @@
+# Domain Layer - Platform-agnostic business logic

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"]

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)

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

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",
+]

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=="

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

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",
+]

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())

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"

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]

src/infrastructure/__init__.py

@@ -0,0 +1 @@
+# Infrastructure Layer

src/infrastructure/platform/__init__.py

@@ -0,0 +1,5 @@
+# Platform Adapters
+from .factory import PlatformAdapterFactory
+from .base import PlatformAdapter
+
+__all__ = ["PlatformAdapterFactory", "PlatformAdapter"]

src/infrastructure/platform/adapters/__init__.py

@@ -0,0 +1,4 @@
+# Platform Adapters
+from .onebot_adapter import OneBotAdapter
+
+__all__ = ["OneBotAdapter"]

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
+        }

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

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()