nanobot 消息总线与渠道架构深度分析

class MessageBus:
    def __init__(self):
        self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()
        self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()
        self._outbound_subscribers: dict[str, list[Callable[[OutboundMessage], Awaitable[None]]]] = {}
        self._running = False

async def publish_inbound(self, msg: InboundMessage) -> None:
    """Publish a message from a channel to the agent."""
    await self.inbound.put(msg)

async def consume_inbound(self) -> InboundMessage:
    """Consume the next inbound message (blocks until available)."""
    return await self.inbound.get()

def subscribe_outbound(
    self, 
    channel: str, 
    callback: Callable[[OutboundMessage], Awaitable[None]]
) -> None:
    """Subscribe to outbound messages for a specific channel."""
    if channel not in self._outbound_subscribers:
        self._outbound_subscribers[channel] = []
    self._outbound_subscribers[channel].append(callback)

async def dispatch_outbound(self) -> None:
    """
    Dispatch outbound messages to subscribed channels.
    Run this as a background task.
    """
    self._running = True
    while self._running:
        try:
            msg = await asyncio.wait_for(self.outbound.get(), timeout=1.0)
            subscribers = self._outbound_subscribers.get(msg.channel, [])
            for callback in subscribers:
                try:
                    await callback(msg)
                except Exception as e:
                    logger.error(f"Error dispatching to {msg.channel}: {e}")
        except asyncio.TimeoutError:
            continue

@dataclass
class InboundMessage:
    """Message received from a chat channel."""
    
    channel: str  # telegram, discord, slack, whatsapp
    sender_id: str  # User identifier
    chat_id: str  # Chat/channel identifier
    content: str  # Message text
    timestamp: datetime = field(default_factory=datetime.now)
    media: list[str] = field(default_factory=list)  # Media URLs
    metadata: dict[str, Any] = field(default_factory=dict)  # Channel-specific data
    
    @property
    def session_key(self) -> str:
        """Unique key for session identification."""
        return f"{self.channel}:{self.chat_id}"

@dataclass
class OutboundMessage:
    """Message to send to a chat channel."""
    
    channel: str
    chat_id: str
    content: str
    reply_to: str | None = None
    media: list[str] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

@property
def session_key(self) -> str:
    """Unique key for session identification."""
    return f"{self.channel}:{self.chat_id}"

metadata={
    "message_id": message.message_id,
    "user_id": user.id,
    "username": user.username,
    "first_name": user.first_name,
    "is_group": message.chat.type != "private"
}

metadata={
    "message_id": data.get("id"),
    "timestamp": data.get("timestamp"),
    "is_group": data.get("isGroup", False)
}

# Download media if present
if media_file and self._app:
    try:
        file = await self._app.bot.get_file(media_file.file_id)
        ext = self._get_extension(media_type, getattr(media_file, 'mime_type', None))
        
        # Save to workspace/media/
        from pathlib import Path
        media_dir = Path.home() / ".nanobot" / "media"
        media_dir.mkdir(parents=True, exist_ok=True)
        
        file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
        await file.download_to_drive(str(file_path))
        
        media_paths.append(str(file_path))

def __init__(self, config: Config, bus: MessageBus):
    self.config = config
    self.bus = bus
    self.channels: dict[str, BaseChannel] = {}
    self._dispatch_task: asyncio.Task | None = None
    
    self._init_channels()

def _init_channels(self) -> None:
    """Initialize channels based on config."""
    
    # Telegram channel
    if self.config.channels.telegram.enabled:
        try:
            from nanobot.channels.telegram import TelegramChannel
            self.channels["telegram"] = TelegramChannel(
                self.config.channels.telegram,
                self.bus,
                groq_api_key=self.config.providers.groq.api_key,
            )
            logger.info("Telegram channel enabled")
        except ImportError as e:
            logger.warning(f"Telegram channel not available: {e}")
    
    # WhatsApp channel
    if self.config.channels.whatsapp.enabled:
        try:
            from nanobot.channels.whatsapp import WhatsAppChannel
            self.channels["whatsapp"] = WhatsAppChannel(
                self.config.channels.whatsapp, self.bus
            )
            logger.info("WhatsApp channel enabled")
        except ImportError as e:
            logger.warning(f"WhatsApp channel not available: {e}")

async def start_all(self) -> None:
    """Start WhatsApp channel and the outbound dispatcher."""
    if not self.channels:
        logger.warning("No channels enabled")
        return
    
    # Start outbound dispatcher
    self._dispatch_task = asyncio.create_task(self._dispatch_outbound())
    
    # Start WhatsApp channel
    tasks = []
    for name, channel in self.channels.items():
        logger.info(f"Starting {name} channel...")
        tasks.append(asyncio.create_task(channel.start()))
    
    # Wait for all to complete (they should run forever)
    await asyncio.gather(*tasks, return_exceptions=True)

async def _dispatch_outbound(self) -> None:
    """Dispatch outbound messages to the appropriate channel."""
    logger.info("Outbound dispatcher started")
    
    while True:
        try:
            msg = await asyncio.wait_for(
                self.bus.consume_outbound(),
                timeout=1.0
            )
            
            channel = self.channels.get(msg.channel)
            if channel:
                try:
                    await channel.send(msg)
                except Exception as e:
                    logger.error(f"Error sending to {msg.channel}: {e}")
            else:
                logger.warning(f"Unknown channel: {msg.channel}")
                
        except asyncio.TimeoutError:
            continue
        except asyncio.CancelledError:
            break

async def stop_all(self) -> None:
    """Stop all channels and the dispatcher."""
    logger.info("Stopping all channels...")
    
    # Stop dispatcher
    if self._dispatch_task:
        self._dispatch_task.cancel()
        try:
            await self._dispatch_task
        except asyncio.CancelledError:
            pass
    
    # Stop all channels
    for name, channel in self.channels.items():
        try:
            await channel.stop()
            logger.info(f"Stopped {name} channel")
        except Exception as e:
            logger.error(f"Error stopping {name}: {e}")

@abstractmethod
async def start(self) -> None:
    """
    Start the channel and begin listening for messages.
    
    This should be a long-running async task that:
    1. Connects to the chat platform
    2. Listens for incoming messages
    3. Forwards messages to the bus via _handle_message()
    """
    pass

@abstractmethod
async def stop(self) -> None:
    """Stop the channel and clean up resources."""
    pass

@abstractmethod
async def send(self, msg: OutboundMessage) -> None:
    """
    Send a message through this channel.
    
    Args:
        msg: The message to send.
    """
    pass

def is_allowed(self, sender_id: str) -> bool:
    """
    Check if a sender is allowed to use this bot.
    
    Args:
        sender_id: The sender's identifier.
    
    Returns:
        True if allowed, False otherwise.
    """
    allow_list = getattr(self.config, "allow_from", [])
    
    # If no allow list, allow everyone
    if not allow_list:
        return True
    
    sender_str = str(sender_id)
    if sender_str in allow_list:
        return True
    if "|" in sender_str:
        for part in sender_str.split("|"):
            if part and part in allow_list:
                return True
    return False

async def _handle_message(
    self,
    sender_id: str,
    chat_id: str,
    content: str,
    media: list[str] | None = None,
    metadata: dict[str, Any] | None = None
) -> None:
    """
    Handle an incoming message from the chat platform.
    
    This method checks permissions and forwards to the bus.
    """
    if not self.is_allowed(sender_id):
        return
    
    msg = InboundMessage(
        channel=self.name,
        sender_id=str(sender_id),
        chat_id=str(chat_id),
        content=content,
        media=media or [],
        metadata=metadata or {}
    )
    
    await self.bus.publish_inbound(msg)

def _markdown_to_telegram_html(text: str) -> str:
    """
    Convert markdown to Telegram-safe HTML.
    """
    if not text:
        return ""
    
    # 1. Extract and protect code blocks (preserve content from other processing)
    code_blocks: list[str] = []
    def save_code_block(m: re.Match) -> str:
        code_blocks.append(m.group(1))
        return f"\x00CB{len(code_blocks) - 1}\x00"
    
    text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text)
    
    # 2. Extract and protect inline code
    inline_codes: list[str] = []
    def save_inline_code(m: re.Match) -> str:
        inline_codes.append(m.group(1))
        return f"\x00IC{len(inline_codes) - 1}\x00"
    
    text = re.sub(r'`([^`]+)`', save_inline_code, text)
    
    # 3. Headers # Title -> just the title text
    text = re.sub(r'^#{1,6}\s+(.+)$', r'\1', text, flags=re.MULTILINE)
    
    # 4. Blockquotes > text -> just the text (before HTML escaping)
    text = re.sub(r'^>\s*(.*)$', r'\1', text, flags=re.MULTILINE)
    
    # 5. Escape HTML special characters
    text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
    
    # 6. Links [text](url) - must be before bold/italic to handle nested cases
    text = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', text)
    
    # 7. Bold **text** or __text__
    text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
    text = re.sub(r'__(.+?)__', r'<b>\1</b>', text)
    
    # 8. Italic _text_ (avoid matching inside words like some_var_name)
    text = re.sub(r'(?<![a-zA-Z0-9])_([^_]+)_(?![a-zA-Z0-9])', r'<i>\1</i>', text)
    
    # 9. Strikethrough ~~text~~
    text = re.sub(r'~~(.+?)~~', r'<s>\1</s>', text)
    
    # 10. Bullet lists - item -> • item
    text = re.sub(r'^[-*]\s+', '• ', text, flags=re.MULTILINE)
    
    # 11. Restore inline code with HTML tags
    for i, code in enumerate(inline_codes):
        # Escape HTML in code content
        escaped = code.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        text = text.replace(f"\x00IC{i}\x00", f"<code>{escaped}</code>")
    
    # 12. Restore code blocks with HTML tags
    for i, code in enumerate(code_blocks):
        # Escape HTML in code content
        escaped = code.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        text = text.replace(f"\x00CB{i}\x00", f"<pre><code>{escaped}</code></pre>")
    
    return text

# Handle voice transcription
if media_type == "voice" or media_type == "audio":
    from nanobot.providers.transcription import GroqTranscriptionProvider
    transcriber = GroqTranscriptionProvider(api_key=self.groq_api_key)
    transcription = await transcriber.transcribe(file_path)
    if transcription:
        logger.info(f"Transcribed {media_type}: {transcription[:50]}...")
        content_parts.append(f"[transcription: {transcription}]")
    else:
        content_parts.append(f"[{media_type}: {file_path}]")

class GroqTranscriptionProvider:
    """
    Voice transcription provider using Groq's Whisper API.
    
    Groq offers extremely fast transcription with a generous free tier.
    """
    
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or os.environ.get("GROQ_API_KEY")
        self.api_url = "https://api.groq.com/openai/v1/audio/transcriptions"
    
    async def transcribe(self, file_path: str | Path) -> str:
        """
        Transcribe an audio file using Groq.
        """
        if not self.api_key:
            logger.warning("Groq API key not configured for transcription")
            return ""
        
        path = Path(file_path)
        if not path.exists():
            logger.error(f"Audio file not found: {file_path}")
            return ""
        
        try:
            async with httpx.AsyncClient() as client:
                with open(path, "rb") as f:
                    files = {
                        "file": (path.name, f),
                        "model": (None, "whisper-large-v3"),
                    }
                    headers = {
                        "Authorization": f"Bearer {self.api_key}",
                    }
                    
                    response = await client.post(
                        self.api_url,
                        headers=headers,
                        files=files,
                        timeout=60.0
                    )
                    
                    response.raise_for_status()
                    data = response.json()
                    return data.get("text", "")
                    
        except Exception as e:
            logger.error(f"Groq transcription error: {e}")
            return ""

async def start(self) -> None:
    """Start the WhatsApp channel by connecting to the bridge."""
    import websockets
    
    bridge_url = self.config.bridge_url
    
    logger.info(f"Connecting to WhatsApp bridge at {bridge_url}...")
    
    self._running = True
    
    while self._running:
        try:
            async with websockets.connect(bridge_url) as ws:
                self._ws = ws
                self._connected = True
                logger.info("Connected to WhatsApp bridge")
                
                # Listen for messages
                async for message in ws:
                    try:
                        await self._handle_bridge_message(message)
                    except Exception as e:
                        logger.error(f"Error handling bridge message: {e}")
                
        except asyncio.CancelledError:
            break
        except Exception as e:
            self._connected = False
            self._ws = None
            logger.warning(f"WhatsApp bridge connection error: {e}")
            
            if self._running:
                logger.info("Reconnecting in 5 seconds...")
                await asyncio.sleep(5)

async def _handle_bridge_message(self, raw: str) -> None:
    """Handle a message from the bridge."""
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        logger.warning(f"Invalid JSON from bridge: {raw[:100]}")
        return
    
    msg_type = data.get("type")
    
    if msg_type == "message":
        # Incoming message from WhatsApp
        sender = data.get("sender", "")
        content = data.get("content", "")
        
        # sender is typically: <phone>@s.whatsapp.net
        # Extract just the phone number as chat_id
        chat_id = sender.split("@")[0] if "@" in sender else sender
        
        # Handle voice transcription if it's a voice message
        if content == "[Voice Message]":
            logger.info(f"Voice message received from {chat_id}, but direct download from bridge is not yet supported.")
            content = "[Voice Message: Transcription not available for WhatsApp yet]"
        
        await self._handle_message(
            sender_id=chat_id,
            chat_id=sender,  # Use full JID for replies
            content=content,
            metadata={
                "message_id": data.get("id"),
                "timestamp": data.get("timestamp"),
                "is_group": data.get("isGroup", False)
            }
        )
    
    elif msg_type == "status":
        # Connection status update
        status = data.get("status")
        logger.info(f"WhatsApp status: {status}")
        
        if status == "connected":
            self._connected = True
        elif status == "disconnected":
            self._connected = False
    
    elif msg_type == "qr":
        # QR code for authentication
        logger.info("Scan QR code in the bridge terminal to connect WhatsApp")
    
    elif msg_type == "error":
        logger.error(f"WhatsApp bridge error: {data.get('error')}")

async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
    """
    Process a system message (e.g., subagent announce).
    
    The chat_id field contains "original_channel:original_chat_id" to route
    the response back to the correct destination.
    """
    logger.info(f"Processing system message from {msg.sender_id}")
    
    # Parse origin from chat_id (format: "channel:chat_id")
    if ":" in msg.chat_id:
        parts = msg.chat_id.split(":", 1)
        origin_channel = parts[0]
        origin_chat_id = parts[1]
    else:
        # Fallback
        origin_channel = "cli"
        origin_chat_id = msg.chat_id
    
    # Use the origin session for context
    session_key = f"{origin_channel}:{origin_chat_id}"
    session = self.sessions.get_or_create(session_key)
    
    # ... 处理逻辑 ...
    
    return OutboundMessage(
        channel=origin_channel,
        chat_id=origin_chat_id,
        content=final_content
    )

@app.command()
def gateway(
    port: int = typer.Option(18790, "--port", "-p", help="Gateway port"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
):
    """Start the nanobot gateway."""
    # ... 配置加载 ...
    
    # Create components
    bus = MessageBus()
    
    # Create provider
    provider = LiteLLMProvider(
        api_key=api_key,
        api_base=api_base,
        default_model=config.agents.defaults.model
    )
    
    # Create agent
    agent = AgentLoop(
        bus=bus,
        provider=provider,
        workspace=config.workspace_path,
        model=config.agents.defaults.model,
        max_iterations=config.agents.defaults.max_tool_iterations,
        brave_api_key=config.tools.web.search.api_key or None
    )
    
    # Create cron service
    # Create heartbeat service
    
    # Create channel manager
    channels = ChannelManager(config, bus)
    
    async def run():
        try:
            await cron.start()
            await heartbeat.start()
            await asyncio.gather(
                agent.run(),
                channels.start_all(),
            )
        except KeyboardInterrupt:
            console.print("\nShutting down...")
            heartbeat.stop()
            cron.stop()
            agent.stop()
            await channels.stop_all()
    
    asyncio.run(run())

msg = await asyncio.wait_for(self.outbound.get(), timeout=1.0)

# Extract and protect code blocks
code_blocks: list[str] = []
def save_code_block(m: re.Match) -> str:
    code_blocks.append(m.group(1))
    return f"\x00CB{len(code_blocks) - 1}\x00"

text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text)

if ":" in msg.chat_id:
    parts = msg.chat_id.split(":", 1)
    origin_channel = parts[0]
    origin_chat_id = parts[1]

if "|" in sender_str:
    for part in sender_str.split("|"):
        if part and part in allow_list:
            return True