resolve merge conflicts: keep both dingtalk and email channels

nanobot/channels/dingtalk.py

@@ -0,0 +1,238 @@
+"""DingTalk/DingDing channel implementation using Stream Mode."""
+
+import asyncio
+import json
+import time
+from typing import Any
+
+from loguru import logger
+import httpx
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.base import BaseChannel
+from nanobot.config.schema import DingTalkConfig
+
+try:
+    from dingtalk_stream import (
+        DingTalkStreamClient,
+        Credential,
+        CallbackHandler,
+        CallbackMessage,
+        AckMessage,
+    )
+    from dingtalk_stream.chatbot import ChatbotMessage
+
+    DINGTALK_AVAILABLE = True
+except ImportError:
+    DINGTALK_AVAILABLE = False
+    # Fallback so class definitions don't crash at module level
+    CallbackHandler = object  # type: ignore[assignment,misc]
+    CallbackMessage = None  # type: ignore[assignment,misc]
+    AckMessage = None  # type: ignore[assignment,misc]
+    ChatbotMessage = None  # type: ignore[assignment,misc]
+
+
+class NanobotDingTalkHandler(CallbackHandler):
+    """
+    Standard DingTalk Stream SDK Callback Handler.
+    Parses incoming messages and forwards them to the Nanobot channel.
+    """
+
+    def __init__(self, channel: "DingTalkChannel"):
+        super().__init__()
+        self.channel = channel
+
+    async def process(self, message: CallbackMessage):
+        """Process incoming stream message."""
+        try:
+            # Parse using SDK's ChatbotMessage for robust handling
+            chatbot_msg = ChatbotMessage.from_dict(message.data)
+
+            # Extract text content; fall back to raw dict if SDK object is empty
+            content = ""
+            if chatbot_msg.text:
+                content = chatbot_msg.text.content.strip()
+            if not content:
+                content = message.data.get("text", {}).get("content", "").strip()
+
+            if not content:
+                logger.warning(
+                    f"Received empty or unsupported message type: {chatbot_msg.message_type}"
+                )
+                return AckMessage.STATUS_OK, "OK"
+
+            sender_id = chatbot_msg.sender_staff_id or chatbot_msg.sender_id
+            sender_name = chatbot_msg.sender_nick or "Unknown"
+
+            logger.info(f"Received DingTalk message from {sender_name} ({sender_id}): {content}")
+
+            # Forward to Nanobot via _on_message (non-blocking).
+            # Store reference to prevent GC before task completes.
+            task = asyncio.create_task(
+                self.channel._on_message(content, sender_id, sender_name)
+            )
+            self.channel._background_tasks.add(task)
+            task.add_done_callback(self.channel._background_tasks.discard)
+
+            return AckMessage.STATUS_OK, "OK"
+
+        except Exception as e:
+            logger.error(f"Error processing DingTalk message: {e}")
+            # Return OK to avoid retry loop from DingTalk server
+            return AckMessage.STATUS_OK, "Error"
+
+
+class DingTalkChannel(BaseChannel):
+    """
+    DingTalk channel using Stream Mode.
+
+    Uses WebSocket to receive events via `dingtalk-stream` SDK.
+    Uses direct HTTP API to send messages (SDK is mainly for receiving).
+
+    Note: Currently only supports private (1:1) chat. Group messages are
+    received but replies are sent back as private messages to the sender.
+    """
+
+    name = "dingtalk"
+
+    def __init__(self, config: DingTalkConfig, bus: MessageBus):
+        super().__init__(config, bus)
+        self.config: DingTalkConfig = config
+        self._client: Any = None
+        self._http: httpx.AsyncClient | None = None
+
+        # Access Token management for sending messages
+        self._access_token: str | None = None
+        self._token_expiry: float = 0
+
+        # Hold references to background tasks to prevent GC
+        self._background_tasks: set[asyncio.Task] = set()
+
+    async def start(self) -> None:
+        """Start the DingTalk bot with Stream Mode."""
+        try:
+            if not DINGTALK_AVAILABLE:
+                logger.error(
+                    "DingTalk Stream SDK not installed. Run: pip install dingtalk-stream"
+                )
+                return
+
+            if not self.config.client_id or not self.config.client_secret:
+                logger.error("DingTalk client_id and client_secret not configured")
+                return
+
+            self._running = True
+            self._http = httpx.AsyncClient()
+
+            logger.info(
+                f"Initializing DingTalk Stream Client with Client ID: {self.config.client_id}..."
+            )
+            credential = Credential(self.config.client_id, self.config.client_secret)
+            self._client = DingTalkStreamClient(credential)
+
+            # Register standard handler
+            handler = NanobotDingTalkHandler(self)
+            self._client.register_callback_handler(ChatbotMessage.TOPIC, handler)
+
+            logger.info("DingTalk bot started with Stream Mode")
+
+            # client.start() is an async infinite loop handling the websocket connection
+            await self._client.start()
+
+        except Exception as e:
+            logger.exception(f"Failed to start DingTalk channel: {e}")
+
+    async def stop(self) -> None:
+        """Stop the DingTalk bot."""
+        self._running = False
+        # Close the shared HTTP client
+        if self._http:
+            await self._http.aclose()
+            self._http = None
+        # Cancel outstanding background tasks
+        for task in self._background_tasks:
+            task.cancel()
+        self._background_tasks.clear()
+
+    async def _get_access_token(self) -> str | None:
+        """Get or refresh Access Token."""
+        if self._access_token and time.time() < self._token_expiry:
+            return self._access_token
+
+        url = "https://api.dingtalk.com/v1.0/oauth2/accessToken"
+        data = {
+            "appKey": self.config.client_id,
+            "appSecret": self.config.client_secret,
+        }
+
+        if not self._http:
+            logger.warning("DingTalk HTTP client not initialized, cannot refresh token")
+            return None
+
+        try:
+            resp = await self._http.post(url, json=data)
+            resp.raise_for_status()
+            res_data = resp.json()
+            self._access_token = res_data.get("accessToken")
+            # Expire 60s early to be safe
+            self._token_expiry = time.time() + int(res_data.get("expireIn", 7200)) - 60
+            return self._access_token
+        except Exception as e:
+            logger.error(f"Failed to get DingTalk access token: {e}")
+            return None
+
+    async def send(self, msg: OutboundMessage) -> None:
+        """Send a message through DingTalk."""
+        token = await self._get_access_token()
+        if not token:
+            return
+
+        # oToMessages/batchSend: sends to individual users (private chat)
+        # https://open.dingtalk.com/document/orgapp/robot-batch-send-messages
+        url = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend"
+
+        headers = {"x-acs-dingtalk-access-token": token}
+
+        data = {
+            "robotCode": self.config.client_id,
+            "userIds": [msg.chat_id],  # chat_id is the user's staffId
+            "msgKey": "sampleMarkdown",
+            "msgParam": json.dumps({
+                "text": msg.content,
+                "title": "Nanobot Reply",
+            }),
+        }
+
+        if not self._http:
+            logger.warning("DingTalk HTTP client not initialized, cannot send")
+            return
+
+        try:
+            resp = await self._http.post(url, json=data, headers=headers)
+            if resp.status_code != 200:
+                logger.error(f"DingTalk send failed: {resp.text}")
+            else:
+                logger.debug(f"DingTalk message sent to {msg.chat_id}")
+        except Exception as e:
+            logger.error(f"Error sending DingTalk message: {e}")
+
+    async def _on_message(self, content: str, sender_id: str, sender_name: str) -> None:
+        """Handle incoming message (called by NanobotDingTalkHandler).
+
+        Delegates to BaseChannel._handle_message() which enforces allow_from
+        permission checks before publishing to the bus.
+        """
+        try:
+            logger.info(f"DingTalk inbound: {content} from {sender_name}")
+            await self._handle_message(
+                sender_id=sender_id,
+                chat_id=sender_id,  # For private chat, chat_id == sender_id
+                content=str(content),
+                metadata={
+                    "sender_name": sender_name,
+                    "platform": "dingtalk",
+                },
+            )
+        except Exception as e:
+            logger.error(f"Error publishing DingTalk message: {e}")

nanobot/channels/feishu.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import re
 import threading
 from collections import OrderedDict
 from typing import Any
@@ -156,6 +157,44 @@ class FeishuChannel(BaseChannel):
         loop = asyncio.get_running_loop()
         await loop.run_in_executor(None, self._add_reaction_sync, message_id, emoji_type)
     
+    # Regex to match markdown tables (header + separator + data rows)
+    _TABLE_RE = re.compile(
+        r"((?:^[ \t]*\|.+\|[ \t]*\n)(?:^[ \t]*\|[-:\s|]+\|[ \t]*\n)(?:^[ \t]*\|.+\|[ \t]*\n?)+)",
+        re.MULTILINE,
+    )
+
+    @staticmethod
+    def _parse_md_table(table_text: str) -> dict | None:
+        """Parse a markdown table into a Feishu table element."""
+        lines = [l.strip() for l in table_text.strip().split("\n") if l.strip()]
+        if len(lines) < 3:
+            return None
+        split = lambda l: [c.strip() for c in l.strip("|").split("|")]
+        headers = split(lines[0])
+        rows = [split(l) for l in lines[2:]]
+        columns = [{"tag": "column", "name": f"c{i}", "display_name": h, "width": "auto"}
+                   for i, h in enumerate(headers)]
+        return {
+            "tag": "table",
+            "page_size": len(rows) + 1,
+            "columns": columns,
+            "rows": [{f"c{i}": r[i] if i < len(r) else "" for i in range(len(headers))} for r in rows],
+        }
+
+    def _build_card_elements(self, content: str) -> list[dict]:
+        """Split content into markdown + table elements for Feishu card."""
+        elements, last_end = [], 0
+        for m in self._TABLE_RE.finditer(content):
+            before = content[last_end:m.start()].strip()
+            if before:
+                elements.append({"tag": "markdown", "content": before})
+            elements.append(self._parse_md_table(m.group(1)) or {"tag": "markdown", "content": m.group(1)})
+            last_end = m.end()
+        remaining = content[last_end:].strip()
+        if remaining:
+            elements.append({"tag": "markdown", "content": remaining})
+        return elements or [{"tag": "markdown", "content": content}]
+
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Feishu."""
         if not self._client:
@@ -170,15 +209,20 @@ class FeishuChannel(BaseChannel):
             else:
                 receive_id_type = "open_id"
             
-            # Build text message content
-            content = json.dumps({"text": msg.content})
+            # Build card with markdown + table support
+            elements = self._build_card_elements(msg.content)
+            card = {
+                "config": {"wide_screen_mode": True},
+                "elements": elements,
+            }
+            content = json.dumps(card, ensure_ascii=False)
             
             request = CreateMessageRequest.builder() \
                 .receive_id_type(receive_id_type) \
                 .request_body(
                     CreateMessageRequestBody.builder()
                     .receive_id(msg.chat_id)
-                    .msg_type("text")
+                    .msg_type("interactive")
                     .content(content)
                     .build()
                 ).build()

nanobot/channels/manager.py

@@ -1,7 +1,9 @@
 """Channel manager for coordinating chat channels."""
 
+from __future__ import annotations
+
 import asyncio
-from typing import Any
+from typing import Any, TYPE_CHECKING
 
 from loguru import logger
 
@@ -10,6 +12,9 @@ from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.schema import Config
 
+if TYPE_CHECKING:
+    from nanobot.session.manager import SessionManager
+
 
 class ChannelManager:
     """
@@ -21,9 +26,10 @@ class ChannelManager:
     - Route outbound messages
     """
     
-    def __init__(self, config: Config, bus: MessageBus):
+    def __init__(self, config: Config, bus: MessageBus, session_manager: "SessionManager | None" = None):
         self.config = config
         self.bus = bus
+        self.session_manager = session_manager
         self.channels: dict[str, BaseChannel] = {}
         self._dispatch_task: asyncio.Task | None = None
         
@@ -40,6 +46,7 @@ class ChannelManager:
                     self.config.channels.telegram,
                     self.bus,
                     groq_api_key=self.config.providers.groq.api_key,
+                    session_manager=self.session_manager,
                 )
                 logger.info("Telegram channel enabled")
             except ImportError as e:
@@ -78,6 +85,17 @@ class ChannelManager:
             except ImportError as e:
                 logger.warning(f"Feishu channel not available: {e}")
 
+        # DingTalk channel
+        if self.config.channels.dingtalk.enabled:
+            try:
+                from nanobot.channels.dingtalk import DingTalkChannel
+                self.channels["dingtalk"] = DingTalkChannel(
+                    self.config.channels.dingtalk, self.bus
+                )
+                logger.info("DingTalk channel enabled")
+            except ImportError as e:
+                logger.warning(f"DingTalk channel not available: {e}")
+
         # Email channel
         if self.config.channels.email.enabled:
             try:
@@ -89,8 +107,15 @@ class ChannelManager:
             except ImportError as e:
                 logger.warning(f"Email channel not available: {e}")
     
+    async def _start_channel(self, name: str, channel: BaseChannel) -> None:
+        """Start a channel and log any exceptions."""
+        try:
+            await channel.start()
+        except Exception as e:
+            logger.error(f"Failed to start channel {name}: {e}")
+
     async def start_all(self) -> None:
-        """Start WhatsApp channel and the outbound dispatcher."""
+        """Start all channels and the outbound dispatcher."""
         if not self.channels:
             logger.warning("No channels enabled")
             return
@@ -98,11 +123,11 @@ class ChannelManager:
         # Start outbound dispatcher
         self._dispatch_task = asyncio.create_task(self._dispatch_outbound())
         
-        # Start WhatsApp channel
+        # Start channels
         tasks = []
         for name, channel in self.channels.items():
             logger.info(f"Starting {name} channel...")
-            tasks.append(asyncio.create_task(channel.start()))
+            tasks.append(asyncio.create_task(self._start_channel(name, channel)))
         
         # Wait for all to complete (they should run forever)
         await asyncio.gather(*tasks, return_exceptions=True)

nanobot/channels/telegram.py

@@ -1,17 +1,23 @@
 """Telegram channel implementation using python-telegram-bot."""
 
+from __future__ import annotations
+
 import asyncio
 import re
+from typing import TYPE_CHECKING
 
 from loguru import logger
-from telegram import Update
-from telegram.ext import Application, MessageHandler, filters, ContextTypes
+from telegram import BotCommand, Update
+from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
 
 from nanobot.bus.events import OutboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.schema import TelegramConfig
 
+if TYPE_CHECKING:
+    from nanobot.session.manager import SessionManager
+
 
 def _markdown_to_telegram_html(text: str) -> str:
     """
@@ -85,12 +91,27 @@ class TelegramChannel(BaseChannel):
     
     name = "telegram"
     
-    def __init__(self, config: TelegramConfig, bus: MessageBus, groq_api_key: str = ""):
+    # Commands registered with Telegram's command menu
+    BOT_COMMANDS = [
+        BotCommand("start", "Start the bot"),
+        BotCommand("reset", "Reset conversation history"),
+        BotCommand("help", "Show available commands"),
+    ]
+    
+    def __init__(
+        self,
+        config: TelegramConfig,
+        bus: MessageBus,
+        groq_api_key: str = "",
+        session_manager: SessionManager | None = None,
+    ):
         super().__init__(config, bus)
         self.config: TelegramConfig = config
         self.groq_api_key = groq_api_key
+        self.session_manager = session_manager
         self._app: Application | None = None
         self._chat_ids: dict[str, int] = {}  # Map sender_id to chat_id for replies
+        self._typing_tasks: dict[str, asyncio.Task] = {}  # chat_id -> typing loop task
     
     async def start(self) -> None:
         """Start the Telegram bot with long polling."""
@@ -101,11 +122,15 @@ class TelegramChannel(BaseChannel):
         self._running = True
         
         # Build the application
-        self._app = (
-            Application.builder()
-            .token(self.config.token)
-            .build()
-        )
+        builder = Application.builder().token(self.config.token)
+        if self.config.proxy:
+            builder = builder.proxy(self.config.proxy).get_updates_proxy(self.config.proxy)
+        self._app = builder.build()
+        
+        # Add command handlers
+        self._app.add_handler(CommandHandler("start", self._on_start))
+        self._app.add_handler(CommandHandler("reset", self._on_reset))
+        self._app.add_handler(CommandHandler("help", self._on_help))
         
         # Add message handler for text, photos, voice, documents
         self._app.add_handler(
@@ -116,20 +141,22 @@ class TelegramChannel(BaseChannel):
             )
         )
         
-        # Add /start command handler
-        from telegram.ext import CommandHandler
-        self._app.add_handler(CommandHandler("start", self._on_start))
-        
         logger.info("Starting Telegram bot (polling mode)...")
         
         # Initialize and start polling
         await self._app.initialize()
         await self._app.start()
         
-        # Get bot info
+        # Get bot info and register command menu
         bot_info = await self._app.bot.get_me()
         logger.info(f"Telegram bot @{bot_info.username} connected")
         
+        try:
+            await self._app.bot.set_my_commands(self.BOT_COMMANDS)
+            logger.debug("Telegram bot commands registered")
+        except Exception as e:
+            logger.warning(f"Failed to register bot commands: {e}")
+        
         # Start polling (this runs until stopped)
         await self._app.updater.start_polling(
             allowed_updates=["message"],
@@ -144,6 +171,10 @@ class TelegramChannel(BaseChannel):
         """Stop the Telegram bot."""
         self._running = False
         
+        # Cancel all typing indicators
+        for chat_id in list(self._typing_tasks):
+            self._stop_typing(chat_id)
+        
         if self._app:
             logger.info("Stopping Telegram bot...")
             await self._app.updater.stop()
@@ -157,6 +188,9 @@ class TelegramChannel(BaseChannel):
             logger.warning("Telegram bot not running")
             return
         
+        # Stop typing indicator for this chat
+        self._stop_typing(msg.chat_id)
+        
         try:
             # chat_id should be the Telegram chat ID (integer)
             chat_id = int(msg.chat_id)
@@ -188,9 +222,45 @@ class TelegramChannel(BaseChannel):
         user = update.effective_user
         await update.message.reply_text(
             f"👋 Hi {user.first_name}! I'm nanobot.\n\n"
-            "Send me a message and I'll respond!"
+            "Send me a message and I'll respond!\n"
+            "Type /help to see available commands."
         )
     
+    async def _on_reset(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle /reset command — clear conversation history."""
+        if not update.message or not update.effective_user:
+            return
+        
+        chat_id = str(update.message.chat_id)
+        session_key = f"{self.name}:{chat_id}"
+        
+        if self.session_manager is None:
+            logger.warning("/reset called but session_manager is not available")
+            await update.message.reply_text("⚠️ Session management is not available.")
+            return
+        
+        session = self.session_manager.get_or_create(session_key)
+        msg_count = len(session.messages)
+        session.clear()
+        self.session_manager.save(session)
+        
+        logger.info(f"Session reset for {session_key} (cleared {msg_count} messages)")
+        await update.message.reply_text("🔄 Conversation history cleared. Let's start fresh!")
+    
+    async def _on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle /help command — show available commands."""
+        if not update.message:
+            return
+        
+        help_text = (
+            "🐈 <b>nanobot commands</b>\n\n"
+            "/start — Start the bot\n"
+            "/reset — Reset conversation history\n"
+            "/help — Show this help message\n\n"
+            "Just send me a text message to chat!"
+        )
+        await update.message.reply_text(help_text, parse_mode="HTML")
+    
     async def _on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle incoming messages (text, photos, voice, documents)."""
         if not update.message or not update.effective_user:
@@ -273,10 +343,15 @@ class TelegramChannel(BaseChannel):
         
         logger.debug(f"Telegram message from {sender_id}: {content[:50]}...")
         
+        str_chat_id = str(chat_id)
+        
+        # Start typing indicator before processing
+        self._start_typing(str_chat_id)
+        
         # Forward to the message bus
         await self._handle_message(
             sender_id=sender_id,
-            chat_id=str(chat_id),
+            chat_id=str_chat_id,
             content=content,
             media=media_paths,
             metadata={
@@ -288,6 +363,29 @@ class TelegramChannel(BaseChannel):
             }
         )
     
+    def _start_typing(self, chat_id: str) -> None:
+        """Start sending 'typing...' indicator for a chat."""
+        # Cancel any existing typing task for this chat
+        self._stop_typing(chat_id)
+        self._typing_tasks[chat_id] = asyncio.create_task(self._typing_loop(chat_id))
+    
+    def _stop_typing(self, chat_id: str) -> None:
+        """Stop the typing indicator for a chat."""
+        task = self._typing_tasks.pop(chat_id, None)
+        if task and not task.done():
+            task.cancel()
+    
+    async def _typing_loop(self, chat_id: str) -> None:
+        """Repeatedly send 'typing' action until cancelled."""
+        try:
+            while self._app:
+                await self._app.bot.send_chat_action(chat_id=int(chat_id), action="typing")
+                await asyncio.sleep(4)
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logger.debug(f"Typing indicator stopped for {chat_id}: {e}")
+    
     def _get_extension(self, media_type: str, mime_type: str | None) -> str:
         """Get file extension based on media type."""
         if mime_type:

nanobot/channels/whatsapp.py

@@ -100,21 +100,25 @@ class WhatsAppChannel(BaseChannel):
         
         if msg_type == "message":
             # Incoming message from WhatsApp
+            # Deprecated by whatsapp: old phone number style typically: <phone>@s.whatspp.net
+            pn = data.get("pn", "")
+            # New LID sytle typically: 
             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
+            # Extract just the phone number or lid as chat_id
+            user_id = pn if pn else sender
+            sender_id = user_id.split("@")[0] if "@" in user_id else user_id
+            logger.info(f"Sender {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.")
+                logger.info(f"Voice message received from {sender_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
+                sender_id=sender_id,
+                chat_id=sender,  # Use full LID for replies
                 content=content,
                 metadata={
                     "message_id": data.get("id"),