我开发了一个 Telegram 机器人“迷你应用程序”,允许用户访问网站并通过输入文本来解释他们的问题来打开“票证”。接下来,通过机器人界面向用户发送一条消息“管理员即将到达”,并通过 Webhook 发送一条 Discord 消息,通知版主已创建新票证,包括格式为 / 的命令管理_开始
在名为“TelegramChat.py”的第二个文件中,如下所述,代码使用两个单独的 Telegram 机器人在用户和管理员之间实现消息传递系统。该代码应该支持三个主要功能:
#!/usr/bin/env python
# pylint: disable=unused-argument
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# Configuration of bot tokens
USER_BOT_TOKEN = ""
ADMIN_BOT_TOKEN = ""
# Dictionary to track conversations: key = user ID, value = admin ID
conversations = {}
# Initialize applications for user and admin bots
user_bot_app = Application.builder().token(USER_BOT_TOKEN).build()
admin_bot_app = Application.builder().token(ADMIN_BOT_TOKEN).build()
async def admin_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Admin initiates a conversation with a user."""
if context.args:
user_id = context.args[0] # Assuming the command is "/admin_start <user_id>"
admin_id = update.effective_user.id
conversations[user_id] = str(admin_id)
await context.bot.send_message(chat_id=admin_id, text=f"Connected with user {user_id}.")
await context.bot.send_message(chat_id=user_id, text="An admin has joined the chat.")
else:
await update.message.reply_text("Please provide the user ID. Format: /admin_start <user_id>.")
async def handle_user_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Forward messages from users to their linked admin."""
user_id = str(update.effective_user.id)
if user_id in conversations:
admin_id = conversations[user_id]
await context.bot.send_message(chat_id=admin_id, text=f"Message from user {user_id}: {update.message.text}")
else:
await update.message.reply_text("Please wait, connecting you to an admin...")
async def handle_admin_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Forward messages from the admin to the linked user."""
admin_id = str(update.effective_user.id)
user_ids = [user_id for user_id, admin in conversations.items() if admin == admin_id]
if user_ids:
for user_id in user_ids:
await context.bot.send_message(chat_id=user_id, text=f"Message from admin: {update.message.text}")
# Add handlers to the user bot
user_bot_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_user_message))
# Add handlers to the admin bot
admin_bot_app.add_handler(CommandHandler("start", admin_start))
admin_bot_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_admin_message))
if __name__ == "__main__":
# Run the bots in polling mode
user_bot_app.run_polling()
admin_bot_app.run_polling()
对于管理员:命令
/admin_start <user_id>
允许管理员发起与用户的对话。然后,管理员的 ID 会以用户的 ID 为键记录在 conversations
字典中。
来自用户的消息:当用户发送消息时,如果已与管理员建立对话(在
conversations
字典中验证),则消息将转发给该管理员。
来自管理员的消息:当管理员发送消息时,该消息将转发给
conversations
字典中链接到该管理员的所有用户。
但是,命令 /admin_start
任何人都可以帮助诊断为什么
/admin_start <user_id>
命令没有按预期启动对话吗?
此外,为了测试代码,我创建了一个名为“CombinedBot.py”的文件,其中组合了 Ticket.py 和 TelegramChat.py。我已经遇到过“telegram.error.Conflict”类型的问题,但现在我不再遇到它们了。 “Ticket.py”中的部分可以正常工作,但聊天部分却不能。
机器人无法发起与用户的第一次交互,用户必须先找到机器人名称并单击开始。顺便说一句,您只需要 1 个机器人即可实现所有这些功能。