我开始学习 Python,并想知道是否有可能以某种方式在 python-telegram-bot 上制作“FSM”。我认为这是一个愚蠢的问题,但是......
我想将电报机器人放在谷歌云功能第一代上,所以我使用此代码使机器人工作
main.py:
import asyncio
import os
import re
from functions_framework import http
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
filters,
CallbackQueryHandler,
ConversationHandler,
)
from telegram import Update
from handlers.handlers import start
from text_const import CANCEL
from handlers.fsm import *
@http
def telegram_bot(request):
return asyncio.run(main(request))
async def main(request):
token = '...'
app = Application.builder().token(token).build()
bot = app.bot
app.add_handler(CommandHandler("start", start))
app.add_handler(conv_handler)
app.add_handler(MessageHandler(filters.Regex(f"^{CANCEL}$"),
cancel_deal))
if request.method == 'GET':
await
bot.set_webhook(f'https://{request.host}/telegram_bot')
return "set"
async with app:
update = Update.de_json(request.json, bot)
await app.process_update(update)
return "ok"
正如我之前所说,我刚刚开始学习Python。在创建对话脚本“aiogram 中的 FSM”时,我惊呆了。我将所有脚本逻辑移至单独的 fsm.py 文件中
fsm.py:
from telegram import Update, InlineKeyboardMarkup
from telegram.ext import (
Application,
MessageHandler,
filters,
CallbackQueryHandler,
ConversationHandler,
ContextTypes,
)
from text_const import *
async def request(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text('Hi, U good?')
return 1
async def first_response(update: Update, context: ContextTypes.DEFAULT_TYPE):
answer = update.message.text
if answer.lower() in ['yeap', 'yes']:
await update.message.reply_text('U male?')
return 2
else:
await update.message.reply_text('Bye...')
return ConversationHandler.END
async def second_response(update: Update, context: ContextTypes.DEFAULT_TYPE):
sex = {'Male': 'U male', 'Female': 'U female'}
user_answer = update.message.text
answer_sex = sex.get(user_answer, 'not found')
await update.message.reply_text('TY:')
await update.message.reply_text(answer_sex)
return ConversationHandler.END
async def cancel_deal(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("Bye! Good day!")
return ConversationHandler.END
conv_handler = ConversationHandler(
entry_points= [MessageHandler(filters.Regex(f"^{REQUEST}$"
), request)],
states={
1: [MessageHandler(filters.Text, first_response)],
2: [MessageHandler(filters.Text, second_response)]
},
fallbacks=[MessageHandler(filters.Regex(f"^{CANCEL}$"),
cancel_deal)]
)
之后,我们向机器人发送了一个负责触发对话脚本的短语,但是当我尝试输入任何消息时,我在日志中收到错误:
Traceback (most recent call last): File "/layers/google.python.pip/pip/lib/python3.12/site-packages/telegram/ext/_application.py", line 1254, in process_update check = handler.check_update(update) # Should the handler handle this update?
File "/layers/google.python.pip/pip/lib/python3.12/site-packages/telegram/ext/_handlers/conversationhandler.py", line 795, in check_update
check = candidate.check_update(update)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/layers/google.python.pip/pip/lib/python3.12/site-packages/telegram/ext/_handlers/messagehandler.py", line 99, in check_update
return self.filters.check_update(update) or False
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Function execution took 969 ms, finished with status code: 200
TypeError: MessageFilter.check_update() missing 1 required positional argument: 'update'
告诉我为什么会发生这种情况,也许你可以指出代码中的缺陷或弱点,我将不胜感激!
困倦时学习代码不是一个好主意。错误是我错误地写了“过滤器模块”。 曾是: 过滤器.文本 并要求: 过滤器.TEXT
conv_handler = ConversationHandler(
entry_points= [MessageHandler(filters.Regex(f"^{REQUEST}$"), request)],
states={
1: [MessageHandler(filters.TEXT, first_response)],
2: [MessageHandler(filters.TEXT, second_response)]
},
fallbacks=[MessageHandler(filters.Regex(f"^{CANCEL}$"), cancel_deal)]
)