「TelBot」打造一个基于PTB的Telegram Bot

本文是基于 python-telegram-bot 的 document 整理记录的一些学习笔记。

0x00 Telegram Bot

Telegram bot 的功能：Telegram bot机器人主要服务于电报（telegram）群组。

1、定时推送消息

机器人入群会按指定的时间和频率进行消息播报，可以是文本、图片、视频，更多的与客户进行交流和互动，在活跃群气氛的同时增加客户的粘性。

2、自动致欢迎词

新人入群自动致欢迎词，给新人融入感以及被关注感。

3、删除广告

可以删除图片、链接、视频、转发消息、隐藏链接等广告内容，群管理员所发内容不受机器人限制。

4、剔除用户

设置指定的非法关键词，一旦有发送含有关键词的消息则协助管理员直接将其踢出群组，让您的群组不再受不和谐信息的困扰！

5、活跃群组

机器人可以根据剧本在电报群里推送消息，模拟用户在群里交流，活跃群氛围，带动客户的聊天兴趣。

6、空投活动

机器人会搜集用户信息，可以通过是否关注了群组、频道、输入了钱包地址、以及是否关注twitter等内容来确定用户获得的糖果的数量，在获得用户信息的同时也更多的避免了“羊毛党”。

7、关键词回复

用户发送含有指定关键词信息，机器人会自动根据该关键词回复对应的内容。

8、爬虫服务

机器人可以根据数据源，爬取合适的新闻或资讯以及产品报价等信息发送到群组。

9、机器人定制

除了以上的功能之外，Telegrambot也可以根据用户的需求定制机器人。

0x01 introduction

Installing

1	$ pip install python-telegram-bot --upgrade

1
2
3

$ git clone https://github.com/python-telegram-bot/python-telegram-bot --recursive
$ cd python-telegram-bot
$ python setup.py install

Check

1
2
3

import telegram
bot = telegram.Bot(token='TOKEN')
print(bot.get_me())

如果正常运行，应当有以下结果

1	{"first_name": "Toledo's Palace Bot", "username": "ToledosPalaceBot"}

Your First Bot

「配置Updater 与dispatcher」

telegram.ext.Updater 和 telegram.ext.Dispatcher 是最重要的两个类。

Updater类不断从telegram获取新的更新消息并传递给dispatcher；
Dispatcher类与一个任务队列（ Job Queue）绑定在一起，调度执行从bot发来的请求。此外，你可以在 Dispatcher中注册不同类型的处理程序（Handler），以执行不同的回调函数。

from telegram.ext import Updater

TOKEN = '...'
REQUEST_KWARGS={
    'proxy_url': 'http://url:port/',
}

updater = Updater(token='TOKEN', use_context=True, request_kwargs=REQUEST_KWARGS)
dispatcher = updater.dispatcher

TOKEN ：操作bot权限的唯一凭证
REQUEST_KWARGS ：配置代理
use_context ：True 就vans了

「开启log」

1
2
3

import logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                     level=logging.INFO)

「定义与绑定handler」

定义回调函数 callback_xxx
实例化一个handler，并绑定对应的回调函数
- 常用的handler有 CommandHandler 与 MessageHandler等，详见 tutorial。
绑定handler与dispatcher

from telegram.ext import CommandHandler, MessageHandler, Filter

def callback_start(update: Update, context: CallbackContext) -> None:
    """Send a message when the command /start is issued."""
    update.message.reply_text('Hi!')
    # or
    # context.bot.send_message(chat_id=update.effective_chat.id, text="I'm a bot, please talk to me!")

def callback_echo(update, context):
    """Echo the user message."""
    update.message.reply_text(update.message.text)
    update.message.forward_date()

    
start_handler = CommandHandler('start', callback_start)
echo_handler = MessageHandler(Filters.text & ~Filters.command, callback_echo)

dispatcher.add_handler(start_handler)
dispatcher.add_handler(echo_handler)

args[0]：Filters
- 对于CommandHandler来说，’start’ 表示只响应 /start 的文本。
- 对于MessageHandler来说，Filters.text & ~Filters.command 表示响应所有文本及未定义命令。
args[1]：callback_func
- 自定义的回调函数

「启动updater」

1	updater.start_polling()

这样，一个简单地 echobot 已经完成啦，give it a try！

0x02 tutorials

Types of Handlers

为了处理不同的 update 消息，telegram.ext 提供了

telegram.ext.MessageHandler for all updates for all message updates
multiple handlers for all the other different types of updates, e.g.
- telegram.ext.CallbackQueryhandler for update.callback_query
- telegram.ext.InlineQueryHandler for update.inline_query

「CommandHandlers with arguments」

def start_callback(update, context):
    user_says = " ".join(context.args)
    update.message.reply_text("You said: " + user_says)


dispatcher.add_handler(CommandHandler("start", start_callback))

对于 /start Hello World! ，context.args 即为 ‘Hello Wolrd!’ ，可以据此处理用户的不同请求。

「MessageHandlers with Pattern matching: Filters.regex」

Message is either video, photo, or document (generic file) 消息是视频、照片或文档(通用文件)

from telegram.ext import MessageHandler, Filters

handler = MessageHandler(Filters.video | Filters.photo | Filters.document, 
                         callback)

Message is a forwarded photo 信息是转发的照片

1	handler = MessageHandler(Filters.forwarded & Filters.photo, callback)

Message is text and contains a link 邮件是文本，包含一个链接

from telegram import MessageEntity

handler = MessageHandler(
    Filters.text & (Filters.entity(MessageEntity.URL) |
                    Filters.entity(MessageEntity.TEXT_LINK)),
    callback)

Message is a photo and it’s not forwarded 信息是一张照片，不会被转发

1	handler = MessageHandler(Filters.photo & (~ Filters.forwarded), callback)

For more complex inputs you can employ the telegram.ext.MessageHandler with telegram.ext.Filters.regex, which internally uses the re-module to match textual user input with a supplied pattern.

Advanced Filters

Custom filters 自定义过滤器

Storing data persistent

用户数据会以字典形式保存在 context.user_data 中。

from uuid import uuid4
from telegram.ext import Updater, CommandHandler

def put(update, context):
    """Usage: /put value"""
    # Generate ID and seperate value from command
    key = str(uuid4())
    # We don't use context.args here, because the value may contain whitespaces
    value = update.message.text.partition(' ')[2]

    # Store value
    context.user_data[key] = value
    # Send the key to the user
    update.message.reply_text(key)

def get(update, context):
    """Usage: /get uuid"""
    # Seperate ID from command
    key = context.args[0]

    # Load value and send it to the user
    value = context.user_data.get(key, 'Not found')
    update.message.reply_text(value)

if __name__ == '__main__':
    updater = Updater('TOKEN', use_context=True)
    dp = updater.dispatcher

    dp.add_handler(CommandHandler('put', put))
    dp.add_handler(CommandHandler('get', get))

    updater.start_polling()
    updater.idle()

为了实现永久存储，需要 persistence classes 支持。

Create a persistence object (e.g. my_persistence = PicklePersistence(filename='my_file'))
Construct Updater with the persistence (Updater('TOKEN', persistence=my_persistence, use_context=True))

更多细节参考 Making your bot persistent。

Exception Handing

当你的bot发生异常时，自动给开发者发送log。

「查询开发者ID」

方法有很多，其中之一是向你的bot发送消息，然后使用以下命令检索更新：

1	https://api.telegram.org/bot<BOTID>/getUpdates

这里要注意，保证你的bot脚本并没有运行，否则getUpdates的信息就直接被bot那边拿走了。

「定义error handler」

DEVELOPER_CHAT_ID = 123456789

def error_handler(update: Update, context: CallbackContext) -> None:
    """Log the error and send a telegram message to notify the developer."""
    # Log the error before we do anything else, so we can see it even if something breaks.
    logger.error(msg="Exception while handling an update:", exc_info=context.error)

    # traceback.format_exception returns the usual python message about an exception, but as a
    # list of strings rather than a single string, so we have to join them together.
    tb_list = traceback.format_exception(None, context.error, context.error.__traceback__)
    tb_string = ''.join(tb_list)

    # Build the message with some markup and additional information about what happened.
    # You might need to add some logic to deal with messages longer than the 4096 character limit.
    message = (
        f'An exception was raised while handling an update\n'
        f'<pre>update = {html.escape(json.dumps(update.to_dict(), indent=2, ensure_ascii=False))}'
        '</pre>\n\n'
        f'<pre>context.chat_data = {html.escape(str(context.chat_data))}</pre>\n\n'
        f'<pre>context.user_data = {html.escape(str(context.user_data))}</pre>\n\n'
        f'<pre>{html.escape(tb_string)}</pre>'
    )

    # Finally, send the message
    context.bot.send_message(chat_id=DEVELOPER_CHAT_ID, text=message, parse_mode=ParseMode.HTML)

「绑定dispatcher」

1	dispatcher.add_error_handler(error_handler)

「测试」

def bad_command(update: Update, context: CallbackContext) -> None:
    """Raise an error to trigger the error handler."""
    context.bot.wrong_method_name()
 
dispatcher.add_handler(CommandHandler('bad_command', bad_command))

Set timer

「实现定时发送消息」

你可以通过下面的方法来创建不同频率和时间的工作任务：

job_queue.run_once, job_queue.run_repeating, job_queue.run_daily and job_queue.run_monthly

1	context.job_queue.run_repeating(callback, due_time, context=chat_id, name=str(chat_id))

「注意时区问题」

参考 python时区设置

1	context.job_queue.run_daily(alarm, time=datetime.time(22, 0, 0, 0, tzinfo= pytz.timezone('Asia/Shanghai')), context=chat_id, name=str(chat_id))

「example」

# Define a few command handlers. These usually take the two arguments update and
# context. Error handlers also receive the raised TelegramError object in error.
def callback_start(update: Update, context: CallbackContext) -> None:
    update.message.reply_text('Hi! Use /set <seconds> to set a timer')


def alarm(context):
    """Send the alarm message."""
    job = context.job
    context.bot.send_message(job.context, text='Beep!')
    # 这里的context.job.context似乎等价于update.effective_chat.id


def remove_job_if_exists(name, context):
    """Remove job with given name. Returns whether job was removed."""
    current_jobs = context.job_queue.get_jobs_by_name(name)
    if not current_jobs:
        return False
    for job in current_jobs:
        job.schedule_removal()
    return True


def callback_set_timer(update: Update, context: CallbackContext) -> None:
    """Add a job to the queue."""
    chat_id = update.message.chat_id
    try:
        # args[0] should contain the time for the timer in seconds
        due = int(context.args[0])
        if due < 0:
            update.message.reply_text('Sorry we can not go back to future!')
            return

        job_removed = remove_job_if_exists(str(chat_id), context)
        context.job_queue.run_once(alarm, due, context=chat_id, name=str(chat_id))
        

        text = 'Timer successfully set!'
        if job_removed:
            text += ' Old one was removed.'
        update.message.reply_text(text)

    except (IndexError, ValueError):
        update.message.reply_text('Usage: /set <seconds>')


def callback_unset(update: Update, context: CallbackContext) -> None:
    """Remove the job if the user changed their mind."""
    chat_id = update.message.chat_id
    job_removed = remove_job_if_exists(str(chat_id), context)
    text = 'Timer successfully cancelled!' if job_removed else 'You have no active timer.'
    update.message.reply_text(text)

Avoiding flood limits

参考这里

Deploy

需要海外VPS站点，loading…

0x03 examples

入门操作

python-telegram-bot/examples/

高阶玩法

conversationbot2
inlineboard2
deeplinking

0x04 practices

loading…