feat: add /newchat command and topic precheck for groups

BukeLy · BukeLy · commit 444efc7ae266 · 2026-01-12T00:01:24.000+08:00
- Add /newchat local command to create new Topic and send first message
- Add topic precheck when bot joins group (is_forum + can_manage_topics)
- Send detailed setup instructions if precheck fails
- Refactor consumer to handle newchat SQS message format
diff --git a/agent-sdk-client/config.toml b/agent-sdk-client/config.toml
@@ -8,6 +8,7 @@ commands = [
 [local_commands]
 # Local-only commands handled by the client
 help = "Hello World"
+newchat = "创建新对话"
 
 [security]
 # User IDs allowed to add bot to groups and send private messages.
diff --git a/agent-sdk-client/consumer.py b/agent-sdk-client/consumer.py
@@ -47,57 +47,68 @@ async def process_message(message_data: dict) -> None:
     config = Config.from_env()
     bot = Bot(config.telegram_token)
 
-    # Reconstruct Update object from stored data
-    update = Update.de_json(message_data['telegram_update'], bot)
-    message = update.message or update.edited_message
-
-    if not message:
-        logger.warning("Received update with no message or edited_message")
-        return
-
-    cmd = config.get_command(message.text)
-    if cmd:
-        if config.is_local_command(cmd):
-            logger.info(
-                "Handling local command in consumer (fallback path)",
-                extra={'chat_id': message.chat_id, 'message_id': message.message_id},
-            )
-            try:
-                await bot.send_message(
-                    chat_id=message.chat_id,
-                    text=config.local_response(cmd),
-                    message_thread_id=message.message_thread_id,
-                    reply_to_message_id=message.message_id,
-                )
-            except Exception:
-                logger.warning("Failed to send local command response", exc_info=True)
+    is_newchat = message_data.get('is_newchat', False)
+
+    if is_newchat:
+        # newchat 消息直接使用 message_data
+        chat_id = message_data['chat_id']
+        thread_id = message_data['thread_id']
+        user_message = message_data['text']
+        message_id = message_data.get('message_id')
+    else:
+        # 正常消息解析 Update
+        update = Update.de_json(message_data['telegram_update'], bot)
+        message = update.message or update.edited_message
+
+        if not message:
+            logger.warning("Received update with no message or edited_message")
             return
 
-        if not config.is_agent_command(cmd):
-            # Defensive guard: producer should already block non-agent commands.
-            logger.info(
-                "Skipping non-agent command (consumer fallback)",
-                extra={
-                    'chat_id': message.chat_id,
-                    'message_id': message.message_id,
-                },
-            )
-            try:
-                await bot.send_message(
-                    chat_id=message.chat_id,
-                    text=config.unknown_command_message(),
-                    message_thread_id=message.message_thread_id,
-                    reply_to_message_id=message.message_id,
+        chat_id = message.chat_id
+        thread_id = message.message_thread_id
+        user_message = message.text
+        message_id = message.message_id
+
+        cmd = config.get_command(user_message)
+        if cmd:
+            if config.is_local_command(cmd):
+                logger.info(
+                    "Handling local command in consumer (fallback path)",
+                    extra={'chat_id': chat_id, 'message_id': message_id},
                 )
-            except Exception:
-                logger.warning("Failed to send local command response", exc_info=True)
-            return
+                try:
+                    await bot.send_message(
+                        chat_id=chat_id,
+                        text=config.local_response(cmd),
+                        message_thread_id=thread_id,
+                        reply_to_message_id=message_id,
+                    )
+                except Exception:
+                    logger.warning("Failed to send local command response", exc_info=True)
+                return
+
+            if not config.is_agent_command(cmd):
+                # Defensive guard: producer should already block non-agent commands.
+                logger.info(
+                    "Skipping non-agent command (consumer fallback)",
+                    extra={'chat_id': chat_id, 'message_id': message_id},
+                )
+                try:
+                    await bot.send_message(
+                        chat_id=chat_id,
+                        text=config.unknown_command_message(),
+                        message_thread_id=thread_id,
+                        reply_to_message_id=message_id,
+                    )
+                except Exception:
+                    logger.warning("Failed to send local command response", exc_info=True)
+                return
 
     # Send typing indicator
     await bot.send_chat_action(
-        chat_id=message.chat_id,
+        chat_id=chat_id,
         action=ChatAction.TYPING,
-        message_thread_id=message.message_thread_id,
+        message_thread_id=thread_id,
     )
 
     # Initialize result with default error response
@@ -118,31 +129,31 @@ async def process_message(message_data: dict) -> None:
                     'Content-Type': 'application/json',
                 },
                 json={
-                    'user_message': message.text,
-                    'chat_id': str(message.chat_id),
-                    'thread_id': str(message.message_thread_id) if message.message_thread_id else None,
+                    'user_message': user_message,
+                    'chat_id': str(chat_id),
+                    'thread_id': str(thread_id) if thread_id else None,
                 },
             )
             response.raise_for_status()
             result = response.json()
 
     except httpx.TimeoutException:
-        logger.warning(f"Agent Server timeout for chat_id={message.chat_id}")
+        logger.warning(f"Agent Server timeout for chat_id={chat_id}")
         await bot.send_message(
-            chat_id=message.chat_id,
+            chat_id=chat_id,
             text="Request timed out.",
-            message_thread_id=message.message_thread_id,
+            message_thread_id=thread_id,
         )
         raise  # Re-raise to trigger SQS retry for transient errors
 
     except Exception as e:
-        logger.exception(f"Agent Server error for chat_id={message.chat_id}")
+        logger.exception(f"Agent Server error for chat_id={chat_id}")
         error_text = f"Error: {str(e)[:200]}"
         try:
             await bot.send_message(
-                chat_id=message.chat_id,
+                chat_id=chat_id,
                 text=error_text,
-                message_thread_id=message.message_thread_id,
+                message_thread_id=thread_id,
             )
         except Exception as send_error:
             logger.error(f"Failed to send error message to Telegram: {send_error}")
@@ -160,21 +171,21 @@ async def process_message(message_data: dict) -> None:
     # Send response to Telegram
     try:
         await bot.send_message(
-            chat_id=message.chat_id,
+            chat_id=chat_id,
             text=text,
             parse_mode=ParseMode.MARKDOWN_V2,
-            message_thread_id=message.message_thread_id,
-            reply_to_message_id=message.message_id,
+            message_thread_id=thread_id,
+            reply_to_message_id=message_id,
         )
     except BadRequest as e:
         if "parse entities" in str(e).lower():
             safe_text = escape_markdown(text, version=2)
             await bot.send_message(
-                chat_id=message.chat_id,
+                chat_id=chat_id,
                 text=safe_text,
                 parse_mode=ParseMode.MARKDOWN_V2,
-                message_thread_id=message.message_thread_id,
-                reply_to_message_id=message.message_id,
+                message_thread_id=thread_id,
+                reply_to_message_id=message_id,
             )
         else:
             raise
diff --git a/agent-sdk-client/handler.py b/agent-sdk-client/handler.py
@@ -149,6 +149,91 @@ def _handle_local_command(bot: Bot, message, config: Config, cmd: str) -> bool:
     return True
 
 
+async def _check_forum_requirements(bot: Bot, chat_id: int) -> tuple[bool, str]:
+    """检查群组 Topic 功能要求。
+
+    Returns:
+        (is_ok, error_message) - 如果满足要求返回 (True, "")，否则返回 (False, 错误提示)
+    """
+    try:
+        chat = await bot.get_chat(chat_id)
+        if not chat.is_forum:
+            return False, (
+                "⚠️ 群组未开启 Topics 功能\n\n"
+                "请按以下步骤开启:\n"
+                "1. 打开群组设置\n"
+                "2. 点击「Topics」\n"
+                "3. 开启 Topics 功能\n"
+                "4. 重新添加 Bot"
+            )
+
+        me = await bot.get_me()
+        member = await bot.get_chat_member(chat_id, me.id)
+        if not getattr(member, 'can_manage_topics', False):
+            return False, (
+                "⚠️ Bot 缺少「管理 Topics」权限\n\n"
+                "请按以下步骤授权:\n"
+                "1. 打开群组设置 > 管理员\n"
+                "2. 选择此 Bot\n"
+                "3. 开启「Manage Topics」权限"
+            )
+        return True, ""
+    except Exception as e:
+        logger.warning(f"Failed to check forum requirements: {e}")
+        return False, f"检查权限失败: {str(e)[:100]}"
+
+
+async def _handle_newchat_command(
+    bot: Bot, message, config: Config, sqs, prompts: str
+) -> bool:
+    """处理 /newchat - 创建 Topic 并发送消息到 SQS。
+
+    Args:
+        bot: Telegram Bot 实例
+        message: Telegram Message 对象
+        config: 配置对象
+        sqs: SQS 客户端
+        prompts: 用户输入的消息内容
+
+    Returns:
+        True 如果成功，False 如果失败
+    """
+    from datetime import datetime
+
+    chat_id = message.chat_id
+    topic_name = f"Chat {datetime.now().strftime('%m/%d %H:%M')}"
+
+    try:
+        forum_topic = await bot.create_forum_topic(chat_id=chat_id, name=topic_name)
+        new_thread_id = forum_topic.message_thread_id
+
+        message_body = {
+            'chat_id': chat_id,
+            'message_id': message.message_id,
+            'text': prompts,
+            'thread_id': new_thread_id,
+            'is_newchat': True,
+        }
+
+        success = _send_to_sqs_safe(sqs, config.queue_url, message_body)
+        if not success:
+            await bot.send_message(
+                chat_id=chat_id,
+                text="发送消息失败，请重试",
+                message_thread_id=new_thread_id,
+            )
+        return success
+
+    except Exception as e:
+        logger.warning(f"Failed to create forum topic: {e}")
+        await bot.send_message(
+            chat_id=chat_id,
+            text=f"创建 Topic 失败: {str(e)[:100]}",
+            message_thread_id=message.message_thread_id,
+        )
+        return False
+
+
 def lambda_handler(event: dict, context: Any) -> dict:
     """Lambda entry point - Producer.
 
@@ -182,6 +267,23 @@ def lambda_handler(event: dict, context: Any) -> dict:
                 extra={'chat_id': chat_id, 'inviter_id': inviter_id},
             )
             _send_metric('SecurityBlock.UnauthorizedGroup')
+        else:
+            # 授权群组的 Topic 预检
+            member_update = update.my_chat_member
+            old_status = member_update.old_chat_member.status
+            new_status = member_update.new_chat_member.status
+
+            if old_status in ('left', 'kicked') and new_status in (
+                'member',
+                'administrator',
+            ):
+                chat_id = member_update.chat.id
+                is_ok, error_msg = asyncio.run(
+                    _check_forum_requirements(bot, chat_id)
+                )
+                if not is_ok:
+                    asyncio.run(bot.send_message(chat_id=chat_id, text=error_msg))
+                    _send_metric('TopicPrecheck.Failed')
         return {'statusCode': 200}
 
     message = update.message or update.edited_message
@@ -201,6 +303,23 @@ def lambda_handler(event: dict, context: Any) -> dict:
             return {'statusCode': 200}
 
     cmd = config.get_command(message.text)
+
+    # /newchat 特殊处理 - 创建 Topic 后发 SQS
+    if cmd == '/newchat':
+        prompts = message.text[len('/newchat') :].strip()
+        if not prompts:
+            bot.send_message(
+                chat_id=message.chat_id,
+                text="用法: /newchat <消息内容>",
+                message_thread_id=message.message_thread_id,
+            )
+            return {'statusCode': 200}
+
+        sqs = _get_sqs_client()
+        asyncio.run(_handle_newchat_command(bot, message, config, sqs, prompts))
+        return {'statusCode': 200}
+
+    # 其他 local command 正常处理
     if cmd and config.is_local_command(cmd):
         _handle_local_command(bot, message, config, cmd)
         return {'statusCode': 200}
