一、过程

  1. 我是先看视频教程。因为官方文档太长了,太乱了,而且我也没耐心。
  2. 代码部分其实很好理解,难度不大。 
  3. 比较麻烦的是,需要与tg客户端进行交互, 操作比较多,看视频教程其实是个好选择。

二、代码 

# -*- coding: UTF-8 -*-
# 2024年7月9日, 0009 14:11

from telegram import Update # pip install python-telegram-bot
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

 

"""
 

1. Telegram Desktop --> 搜索 botfather --> start -- > 输入 newbot
2. @dragon_dance_bot  : t.me/dragon_dance_bot 
3. token: xxx
4. more doc: https://core.telegram.org/bots/api

start - start the bot
help - show some help message
custom - some other command. 

"""

token = "xxx"
bot_username = "@dragon_dance_bot"


# commands   一些命令!  目前就3个命令
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("这是一条欢迎消息!")


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("帮助信息: 请输入 xxx")


async def custom_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("这是一条私人定制消息!")


# 处理相应的工具函数!
def handle_response(text: str) -> str:
    clean_text = text.lower().strip()

    if "hello" in clean_text:
        return "Hi!"
    elif "nice" in clean_text:
        return "Real Nice!"
    else:
        return "Ooooo, What?"


# 这里的这个变量名, update,  我觉得很奇怪。 它其实是一个消息传递的中间件。
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message_type = update.message.chat.type  # 是群聊,还是私聊
    text =  update.message.text

    # 打印一些内容
    print(f"User id: {update.message.chat.id} in {message_type}, {text}")

    if message_type == "group":
        if bot_username in text:
            new_text = text.replace(bot_username, "").strip()
            response = handle_response(new_text)
        else:
            return
    else:
        response = handle_response(text)

    print("Bot 回复的消息是: ", response)
    await update.message.reply_text(response) # 这里才是真正的回复消息。


# log  日志记录一些消息
async def error(update: Update, context: ContextTypes.DEFAULT_TYPE):
    print(f"Update {update}  caused by {context.error}")


if __name__ == '__main__':
    print("Start bot server......")
    app = Application.builder().token(token).build()

    # commands 一些命令
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("help", help_command))
    app.add_handler(CommandHandler("custom", custom_command))


    # Message 消息处理相关的命令!
    app.add_handler(MessageHandler(filters.TEXT, handle_message))

    # Errors
    app.add_error_handler(error)

    # other stuff
    print("Read user message........")
    app.run_polling(poll_interval=1)   # 每1秒读取一次用户的消息!


3. 这里就是做一个简短的记录。下次用起来比较方便一些。

4. 部署问题

heroku.com  注册登录不方便,需要各种验证。

replit.com  环境配置太恶心了,只能使用 poetry. 
poetry 这个名字是挺好听,用起来也是很垃圾,太恶心。

所以,部署的问题,目前搁置。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐