| 1 | """Base channel interface for chat platforms.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from abc import ABC, abstractmethod |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | from loguru import logger |
| 10 | |
| 11 | from nanobot.bus.events import InboundMessage, OutboundMessage |
| 12 | from nanobot.bus.queue import MessageBus |
| 13 | |
| 14 | |
| 15 | class BaseChannel(ABC): |
| 16 | """ |
| 17 | Abstract base class for chat channel implementations. |
| 18 | |
| 19 | Each channel (Telegram, Discord, etc.) should implement this interface |
| 20 | to integrate with the nanobot message bus. |
| 21 | """ |
| 22 | |
| 23 | name: str = "base" |
| 24 | display_name: str = "Base" |
| 25 | transcription_provider: str = "groq" |
| 26 | transcription_api_key: str = "" |
| 27 | transcription_api_base: str = "" |
| 28 | transcription_language: str | None = None |
| 29 | |
| 30 | def __init__(self, config: Any, bus: MessageBus): |
| 31 | """ |
| 32 | Initialize the channel. |
| 33 | |
| 34 | Args: |
| 35 | config: Channel-specific configuration. |
| 36 | bus: The message bus for communication. |
| 37 | """ |
| 38 | self.config = config |
| 39 | self.bus = bus |
| 40 | self._running = False |
| 41 | |
| 42 | async def transcribe_audio(self, file_path: str | Path) -> str: |
| 43 | """Transcribe an audio file via Whisper (OpenAI or Groq). Returns empty string on failure.""" |
| 44 | if not self.transcription_api_key: |
| 45 | return "" |
| 46 | try: |
| 47 | if self.transcription_provider == "openai": |
| 48 | from nanobot.providers.transcription import OpenAITranscriptionProvider |
| 49 | provider = OpenAITranscriptionProvider( |
| 50 | api_key=self.transcription_api_key, |
| 51 | api_base=self.transcription_api_base or None, |
| 52 | language=self.transcription_language or None, |
| 53 | ) |
| 54 | else: |
| 55 | from nanobot.providers.transcription import GroqTranscriptionProvider |
| 56 | provider = GroqTranscriptionProvider( |
| 57 | api_key=self.transcription_api_key, |
| 58 | api_base=self.transcription_api_base or None, |
| 59 | language=self.transcription_language or None, |
| 60 | ) |
| 61 | return await provider.transcribe(file_path) |
| 62 | except Exception as e: |
| 63 | logger.warning("{}: audio transcription failed: {}", self.name, e) |
| 64 | return "" |
| 65 | |
| 66 | async def login(self, force: bool = False) -> bool: |
| 67 | """ |
| 68 | Perform channel-specific interactive login (e.g. QR code scan). |
| 69 | |
| 70 | Args: |
| 71 | force: If True, ignore existing credentials and force re-authentication. |
| 72 | |
| 73 | Returns True if already authenticated or login succeeds. |
| 74 | Override in subclasses that support interactive login. |
| 75 | """ |
| 76 | return True |
| 77 | |
| 78 | @abstractmethod |
| 79 | async def start(self) -> None: |
| 80 | """ |
| 81 | Start the channel and begin listening for messages. |
| 82 | |
| 83 | This should be a long-running async task that: |
| 84 | 1. Connects to the chat platform |
| 85 | 2. Listens for incoming messages |
| 86 | 3. Forwards messages to the bus via _handle_message() |
| 87 | """ |
| 88 | pass |
| 89 | |
| 90 | @abstractmethod |
| 91 | async def stop(self) -> None: |
| 92 | """Stop the channel and clean up resources.""" |
| 93 | pass |
| 94 | |
| 95 | @abstractmethod |
| 96 | async def send(self, msg: OutboundMessage) -> None: |
| 97 | """ |
| 98 | Send a message through this channel. |
| 99 | |
| 100 | Args: |
| 101 | msg: The message to send. |
| 102 | |
| 103 | Implementations should raise on delivery failure so the channel manager |
| 104 | can apply any retry policy in one place. |
| 105 | """ |
| 106 | pass |
| 107 | |
| 108 | async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: |
| 109 | """Deliver a streaming text chunk. |
| 110 | |
| 111 | Override in subclasses to enable streaming. Implementations should |
| 112 | raise on delivery failure so the channel manager can retry. |
| 113 | |
| 114 | Streaming contract: ``_stream_delta`` is a chunk, ``_stream_end`` ends |
| 115 | the current segment, and stateful implementations must key buffers by |
| 116 | ``_stream_id`` rather than only by ``chat_id``. |
| 117 | """ |
| 118 | pass |
| 119 | |
| 120 | @property |
| 121 | def supports_streaming(self) -> bool: |
| 122 | """True when config enables streaming AND this subclass implements send_delta.""" |
| 123 | cfg = self.config |
| 124 | streaming = cfg.get("streaming", False) if isinstance(cfg, dict) else getattr(cfg, "streaming", False) |
| 125 | return bool(streaming) and type(self).send_delta is not BaseChannel.send_delta |
| 126 | |
| 127 | def is_allowed(self, sender_id: str) -> bool: |
| 128 | """Check if *sender_id* is permitted. Empty list → deny all; ``"*"`` → allow all.""" |
| 129 | if isinstance(self.config, dict): |
| 130 | if "allow_from" in self.config: |
| 131 | allow_list = self.config.get("allow_from") |
| 132 | else: |
| 133 | allow_list = self.config.get("allowFrom", []) |
| 134 | else: |
| 135 | allow_list = getattr(self.config, "allow_from", []) |
| 136 | if not allow_list: |
| 137 | logger.warning("{}: allow_from is empty — all access denied", self.name) |
| 138 | return False |
| 139 | if "*" in allow_list: |
| 140 | return True |
| 141 | return str(sender_id) in allow_list |
| 142 | |
| 143 | async def _handle_message( |
| 144 | self, |
| 145 | sender_id: str, |
| 146 | chat_id: str, |
| 147 | content: str, |
| 148 | media: list[str] | None = None, |
| 149 | metadata: dict[str, Any] | None = None, |
| 150 | session_key: str | None = None, |
| 151 | ) -> None: |
| 152 | """ |
| 153 | Handle an incoming message from the chat platform. |
| 154 | |
| 155 | This method checks permissions and forwards to the bus. |
| 156 | |
| 157 | Args: |
| 158 | sender_id: The sender's identifier. |
| 159 | chat_id: The chat/channel identifier. |
| 160 | content: Message text content. |
| 161 | media: Optional list of media URLs. |
| 162 | metadata: Optional channel-specific metadata. |
| 163 | session_key: Optional session key override (e.g. thread-scoped sessions). |
| 164 | """ |
| 165 | if not self.is_allowed(sender_id): |
| 166 | logger.warning( |
| 167 | "Access denied for sender {} on channel {}. " |
| 168 | "Add them to allowFrom list in config to grant access.", |
| 169 | sender_id, self.name, |
| 170 | ) |
| 171 | return |
| 172 | |
| 173 | meta = metadata or {} |
| 174 | if self.supports_streaming: |
| 175 | meta = {**meta, "_wants_stream": True} |
| 176 | |
| 177 | msg = InboundMessage( |
| 178 | channel=self.name, |
| 179 | sender_id=str(sender_id), |
| 180 | chat_id=str(chat_id), |
| 181 | content=content, |
| 182 | media=media or [], |
| 183 | metadata=meta, |
| 184 | session_key_override=session_key, |
| 185 | ) |
| 186 | |
| 187 | await self.bus.publish_inbound(msg) |
| 188 | |
| 189 | @classmethod |
| 190 | def default_config(cls) -> dict[str, Any]: |
| 191 | """Return default config for onboard. Override in plugins to auto-populate config.json.""" |
| 192 | return {"enabled": False} |
| 193 | |
| 194 | @property |
| 195 | def is_running(self) -> bool: |
| 196 | """Check if the channel is running.""" |
| 197 | return self._running |
| 198 |