TL;DR： GitHub仓库在这里

自从学会使用python-telegram-bot包之后，我一直觉得实际编写程序的时候这个包有一些地方不太方便。比如：

• 每次给bot添加一个新的command，就需要

  updater.dispatcher.add_handler(CommandHandler("command", command))
  

  而实际编写的时候常常会忘记要添加这个CommandHandler，等到开始测试的时候才发现，于是又要改代码。而且，一般而言我会把callback函数名和指令名字写成完全一样的，添加handler的时候重复打一次指令名还要加个双引号是一件很头疼的事情。之前写的dicebot里，我是这样处理的：将所有handlers写在同一个文件里，之后import这个只定义了handlers的文件，并遍历里面全部的函数来添加handler。这样做有两个坏处，其中一个就是handlers的文件import别的文件时，如果那个文件dum有定义函数fun，就不能from dum import fun或者from dum import *，只能import dum，然后要调用fun时就需要dum.fun()，非常麻烦。第二个坏处是，每个command都是由几个功能更少的函数组合起来的，这些函数必须要写在别的文件里，破坏了代码的可读性。

• 每个commandHandler接收到消息时，获取消息来源的聊天、用户等信息是非常基本的事情，基本上每个命令都需要做这件事。是否可以在每个command前固定执行一些语句来获取这些基础信息呢？

• 该包提供的功能里，一条消息仅仅能被一个handler处理，不能像酷Q机器人一样，按照顺序先传给第一个handler，如果这个handler不处理就传给下一个，被其中一个handler处理了的话就停止。如果在一个bot上想要实现多个功能就会比较困难。

• 如果bot架设在大陆并使用代理的话，经常会出现消息发送失败的情况，以及消息过长时，会出现超过字数限制，无法发送的情况。这时如果正在执行某个handler，并且这个handler的实现没有用try语句将发送消息的函数包裹住的话，这个command就会被迫停止。如果能够多尝试发送几次，说不定代理不稳定的问题就能得以解决。当超出字数限制时，就需要将消息分成几段发送。

思考&尝试

为了尝试解决第一个问题，首先需要使用类，这个类表示一个tgbot，并将handler以及实际实现它的子函数都写为类的方法。可以给方法前加上一个类装饰器，在初始化需要用add_handler时，遍历所有成员方法，判断它的类型是否是该装饰器的类型，如果是则添加到handlers即可。这样的话，在调用类装饰器的__call__之前，可以先调用一个预执行的方法，就顺带解决了第二个问题。

至于第三个问题，一般而言需要多个handler的情况是：有多个功能需要文本、按钮交互。但按钮和文本只能被一个handler来执行。可以直接参考酷Q机器人的做法，让这些handlers只返回passed或者blocked这样的信号，最后，用一个方法作为直接被add_handler接收的方法，这个方法的作用只有按顺序传给这些handlers，如果其中一个返回了blocked状态，就结束运行，否则继续往下传递给其他handlers。这样的话，又要考虑遍历全部handlers的问题了。于是我想到，这里可以用菱形继承来解决这个问题。

第四个问题则只需要重新定义一个消息发送函数即可，如果在第二个问题提到的预执行函数中获取到了聊天信息，消息发送函数甚至只需要一个字符串作为参数就可以了，因为聊天信息可以存储在对象中。这样不会产生消息发错人的问题，因为python-telegram-bot的消息处理是阻塞的而非并发的。

基本结构

首先定义一个基本的类

class baseBot(object):
    def __init__(self):
        if proxy:
            self.updater = Updater(token=token, use_context=True, request_kwargs={
                                   "proxy_url": proxy_url})
        else:
            self.updater = Updater(token=token, use_context=True)
        ...

基本类baseBot只实现最基础的功能，然后实际的功能类全部继承baseBot，每个类实现一个具体的功能：

class exampleBot1(baseBot):
    def __init(self):
        if not hasattr(self, "updater"):
            baseBot.__init__(self)
        ...

class exampleBot2(baseBot):
    def __init(self):
        if not hasattr(self, "updater"):
            baseBot.__init__(self)
        ...

最后，用一个最终类继承全部功能类：

class finalBot(exampleBot1, exampleBot2):
    def __init__(self):
        for cls in self.__class__.__bases__:
            cls.__init__(self)    

commandHandler专用的装饰器

CommandHandler的callback必须能接收两个参数，一个是Update类型，一个是CallbackContext类型。可以直接用比较朴素的方法实现下面的commandCallbackMethod，但是问题在于，最终实际调用时要调用的__wrapped__需要baseBot对象，以及需要在预执行函数里将聊天信息存入baseBot对象，就必须获取到这个对象的引用。好在每次调用时，__get__函数会被执行一次，此时会传入instance，这样就能方便地获取到方法从属的对象的一个引用了。

这里使用TYPE_CHECKING来提供类型检查功能，因为这个文件utils是要被baseBot文件import的，为了防止循环import，在本文件中对baseBot的import要在if TYPE_CHECKING:语句下面，并且用引号框起来它的类型。这样在实际执行的时候，因为TYPE_CHECKING常数为False，就不会造成循环import。当然，只是为了类型检查而import的CallbackContext也可以放在这里减少实际执行时的import。

这里预执行函数是baseBot的一个方法renewStatus(update)，可以获取到update里的chat，from_user等信息。

# utils
from typing import TYPE_CHECKING, Callable, TypeVar
from telegram import Update
from typing_extensions import final
if TYPE_CHECKING:
    from telegram.ext import CallbackContext
    from basebot import baseBot
RT = TypeVar('RT')

...

@final
class commandCallbackMethod(object):
    """表示一个指令的callback函数，仅限于类的成员方法。
    调用时，会执行一次指令的前置函数。"""

    def __init__(self, func: Callable[[Update, 'CallbackContext'], RT]) -> None:
        wraps(func)(self)
        self.instance: 'baseBot' = None

    def __call__(self, *args, **kwargs):
        numOfArgs = len(args)+len(kwargs.keys())
        if numOfArgs != 2:
            raise RuntimeError(f"指令的callback function参数个数应为2，但接受到{numOfArgs}个")
        if len(args) > 0:
            self.preExcute(args[0])
        else:
            if all(type(kwargs[x]) is not Update for x in kwargs):
                raise RuntimeError("指令接收到的参数类型有误，第一个参数应为Update")
            for key in kwargs:
                if type(kwargs[key]) is Update:
                    self.preExcute(kwargs[key])
                    break
        return self.__wrapped__(self.instance, *args, **kwargs)

    def preExcute(self, update: Update) -> None:
        """在每个command Handler前调用，是指令的前置函数"""
        if self.instance is None:
            raise RuntimeError("command callback method还未获取实例")
        self.instance.renewStatus(update)

    def __get__(self, instance, cls):
        if instance is None:
            raise TypeError("该装饰器仅适用于方法")
        if self.instance is None:
            self.instance = instance
        return self

renewStatus()实现如下，其中的几个没具体写的函数都只返回一些基本信息，看名字就知道是干啥的。

# baseBot

class baseBot(object):
    ...

    def renewStatus(self, update: Update) -> None:
        """在每个command Handler前调用，是指令的前置函数"""
        self.lastchat = getchatid(update)
        if update.callback_query is None:
            self.lastuser = getfromid(update)
            self.lastmsgid = getmsgid(update)
        else:
            self.lastuser = update.callback_query.from_user.id
            self.lastmsgid = -1

定义handleStatus

handleStatus类只需要两个bool类型就完全足够了：一个代表是否blocked，一个代表执行是否正常进行。实际上后者是完全不必要的，我有时为了方便才会需要这样返回值。所以bool完全可以替代handleStatus类。

# utils

class handleStatus(object):
    def __init__(self, normal: bool, block: bool) -> None:
        self.block: bool = block
        self.normal: bool = normal

    def __bool__(self):
        ...

    def blocked(self):
        return self.block


@final
class handleBlocked(handleStatus):
    def __init__(self, normal: bool = True) -> None:
        super().__init__(normal=normal, block=True)

    def __bool__(self):
        return self.normal


@final
class handlePassed(handleStatus):
    def __init__(self) -> None:
        super().__init__(True, False)

    def __bool__(self):
        return True

接下来才是实际问题。finalBot继承所有功能类之后，可以用self.__class__.__bases__遍历所有父类的某个固定方法，这时候对返回的handleStatus对象进行block与否的判断，即可正常执行。遍历时，是按照继承顺序从左到右遍历，所以要将高优先级的功能类写在左边。

给baseBot定义方法，但不进行任何实现：

class baseBot(object):
    ...

    def textHandler(self, update: Update, context: CallbackContext) -> handleStatus:
        """Override"""
        return handlePassed()

功能类实现方法：

# file1

class exampleBot1(baseBot):
    ...

    def textHandler(self, update: Update, context: CallbackContext) -> handleStatus:
        if not cond1:
            return handlePassed()
        dosth1()
        return handleBlocked()

# file2

class exampleBot2(baseBot):
    ...

    def textHandler(self, update: Update, context: CallbackContext) -> handleStatus:
        if not cond2:
            return handlePassed()
        dosth2()
        return handleBlocked()

最终的类：

class finalBot(exampleBot1, exampleBot2):
    ...

    def textHandler(self, update: Update, context: CallbackContext) -> bool:
        self.renewStatus(update) # 预执行函数
        for cls in self.__class__.__bases__:
            status: handleStatus = cls.textHandler(self, update, context)
            if status.blocked():
                return status.normal
        return False

实际实现这些方法的时候，判断是否满足要在该方法中处理的条件，如果不在该方法中处理就返回handlePassed()，否则，正常处理之后返回handleBlocked()，处理出现异常则返回handleBlocked(False)。

额外的消息发送函数

如果每个handler执行前都获取到了lastchat，lastuser，lastmsgid等信息，那么如下的实现可以让我们把回复消息的语句直接写成self.reply(msg)，方便不少。

class baseBot(object):
    ...

    def reply(self, *args, **kwargs) -> int:
        """调用send_message方法，回复或发送消息。
        支持telegram bot中`send_message`方法的keyword argument，
        如`reply_markup`，`reply_to_message_id`，`parse_mode`，`timeout`。
        返回值是message id"""
        ans = None
        chat_id: Optional[int] = None
        text: str = None
        if len(args) > 0:
            if type(args[0]) is int:
                chat_id = args[0]
                kwargs["chat_id"] = chat_id
            elif type(args[0]) is str:
                text = args[0]
                kwargs["text"] = text

            if len(args) > 1:
                assert(type(args[0]) is int)
                assert(type(args[1]) is str)
                text = args[1]
                kwargs["text"] = text

        if chat_id is None and "chat_id" in kwargs:
            chat_id = kwargs["chat_id"]

        if text is None and "text" in kwargs:
            text = kwargs["text"]

        if not text:
            raise ValueError("发生错误：发送消息时没有文本")

        if chat_id is None:
            kwargs["chat_id"] = self.lastchat
            if self.lastmsgid >= 0 and "reply_to_message_id" not in kwargs:
                kwargs["reply_to_message_id"] = self.lastmsgid

        txts = text.split("\n")

        if len(txts) > 10 and len(text) >= 1024:
            while len(txts) > 10:
                kwargs["text"] = "\n".join(txts[:10])
                txts = txts[10:]
                for i in range(5):
                    try:
                        ans = self.bot.send_message(**kwargs).message_id
                        if "reply_markup" in kwargs:
                            kwargs.pop("reply_markup")
                    except Exception as e:
                        if i == 4:
                            raise e
                        time.sleep(5)
                    else:
                        break

        if len(txts) > 0:
            kwargs["text"] = "\n".join(txts)
            for i in range(5):
                try:
                    ans = self.bot.send_message(**kwargs).message_id
                except Exception as e:
                    if i == 4:
                        raise e
                    time.sleep(5)
                else:
                    break

        if ans is None:
            raise ValueError("没有成功发送消息")
        return ans

实际上，这个方法支持的用法很多，比如：

self.reply(tgid, msg) # 给某人发送消息
self.reply(msg, reply_markup=rp_markup) # 发送带按钮的文本
self.reply(msg, reply_to_message_id=None) # 发送消息时不回复消息
self.reply(msg, parse_mode="MarkdownV2")

用同样的方法也可以重新实现delete_message，edit_message_text等函数。

小结

实际上我已经用这个模板写了一个bot，有在Telegram使里用ssh连VPS、搜图以及闹钟三个功能，用这个模板的话，之后能非常轻松地进一步扩展bot的功能。这个模板已经放在GitHub上，如果你觉得有用的话，不妨点个Star。