近期贵树要开发一个机器人,要借助telegrambot,但又要同时运行web服务器,借助web服务器控制telegrambot

最早采用python-telegram-bot,这个带有异步,可是好像于sqlacademy有冲突,只要使用MySQL,就要报错,可是不用又不行。

后来发现了aiogram,比这个简单,而且也是异步,试验后没bug

安装方法

pip install aiogram

先建立一个bot

bot = Bot(TOKEN, parse_mode=ParseMode.HTML)
````

### 简单使用

import asyncio
import logging
import sys
from os import getenv

from aiogram import Bot, Dispatcher, Router, types
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import Message
from aiogram.utils.markdown import hbold

Bot token can be obtained via https://t.me/BotFather

TOKEN = getenv("BOT_TOKEN")

All handlers should be attached to the Router (or Dispatcher)

dp = Dispatcher()

@dp.message(CommandStart())
async def command_start_handler(message: Message) -> None:

"""
This handler receives messages with `/start` command
"""
# Most event objects have aliases for API methods that can be called in events' context
# For example if you want to answer to incoming message you can use `message.answer(...)` alias
# and the target chat will be passed to :ref:`aiogram.methods.send_message.SendMessage`
# method automatically or call API method directly via
# Bot instance: `bot.send_message(chat_id=message.chat.id, ...)`
await message.answer(f"Hello, {hbold(message.from_user.full_name)}!")

@dp.message()
async def echo_handler(message: types.Message) -> None:

"""
Handler will forward receive a message back to the sender

By default, message handler will handle all message types (like a text, photo, sticker etc.)
"""
try:
    # Send a copy of the received message
    await message.send_copy(chat_id=message.chat.id)
except TypeError:
    # But not all the types is supported to be copied so need to handle it
    await message.answer("Nice try!")

async def main() -> None:

# Initialize Bot instance with a default parse mode which will be passed to all API calls
bot = Bot(TOKEN, parse_mode=ParseMode.HTML)
# And the run events dispatching
await dp.start_polling(bot)

if name == "__main__":

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
asyncio.run(main())

*代码显示不正常,日后修复*

### 异步同时使用

async def main() -> None:

await dp.start_polling(bot)

添加这个代码

async def run_aiotelebots():

await aiotelebots.main()

loop = asyncio.get_event_loop()
task = loop.create_task(run_aiotelebots())


这个在fastapi main文件中添加就行

然后就可以同时使用了
最后修改:2024 年 01 月 06 日
如果觉得我的文章对你有用,请随意赞赏