Python-Telegram-bot跨文件传参及分步获取用户输入问题

python-telegram-bot是事件驱动框架，单条消息只会触发一次对应handler，不会在同一个函数内阻塞等待用户下一条输入，必须用ConversationHandler做状态化的多步对话流转，具体实现分两步：

第一步：重构外部call.py文件 #

不要直接在call.py里写死执行逻辑、用exec()跑文件，把呼叫逻辑封装成可接收参数的函数，方便传参调用：

# call.py
# 保留你原来的client初始化逻辑
def make_call(to_number: str, from_number: str):
    call = client.calls.create(
        machine_detection='Enable',
        url='https://ngrok.io/main',
        to=to_number,
        from_=from_number
    )
    return call

第二步：用ConversationHandler实现分步输入逻辑 #

核心逻辑是给对话定义不同状态，每收到一条用户消息就匹配当前状态做处理，存好临时数据后流转到下一个状态，直到所有参数收集完成再执行呼叫：

• 先定义对话状态常量，一共两个待收集的参数，对应两个状态
• 写入口handler，触发后提示用户输入第一个号码（呼出方号码/from_字段），进入第一个等待状态
• 写第一个输入的处理逻辑，校验号码格式合法后存到用户级临时存储context.user_data，提示用户输入第二个号码（接听方号码/to字段），进入第二个等待状态
• 写第二个输入的处理逻辑，校验通过后取出之前存的呼出号码，两个参数一起传给make_call()执行呼叫，结束对话流程
• 可选加取消逻辑，用户发送/cancel随时退出对话

完整参考代码如下：

from telegram import Update
from telegram.ext import (
    ApplicationBuilder,
    CommandHandler,
    MessageHandler,
    filters,
    ConversationHandler,
    ContextTypes
)
# 导入你封装好的呼叫函数
from call import make_call

# 定义对话状态
WAIT_FROM, WAIT_TO = range(2)

# 对话入口：用户发送/call触发
async def call_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        '请输入呼出号码:\n'
        '示例: 18004585478\n'
        '号码必须以1开头'
    )
    return WAIT_FROM

# 处理用户输入的呼出号码
async def get_from_number(update: Update, context: ContextTypes.DEFAULT_TYPE):
    from_num = update.message.text.strip()
    # 基础校验，不符合就提示重输，停在当前状态
    if not from_num.isdigit() or not from_num.startswith('1'):
        await update.message.reply_text('号码格式错误，请重新输入以1开头的纯数字号码')
        return WAIT_FROM
    # 号码合法就存到临时用户数据里
    context.user_data['from_number'] = from_num
    await update.message.reply_text(
        '请输入接听号码:\n'
        '示例: 18004585478\n'
        '号码必须以1开头'
    )
    return WAIT_TO

# 处理用户输入的接听号码，执行呼叫
async def get_to_number(update: Update, context: ContextTypes.DEFAULT_TYPE):
    to_num = update.message.text.strip()
    if not to_num.isdigit() or not to_num.startswith('1'):
        await update.message.reply_text('号码格式错误，请重新输入以1开头的纯数字号码')
        return WAIT_TO
    from_num = context.user_data['from_number']
    await update.message.reply_text('正在发起呼叫...')
    # 执行呼叫逻辑
    call_state = make_call(to_num, from_num)
    # 这里可以根据call_state的返回结果给用户发成功/失败提示
    await update.message.reply_text('呼叫请求已提交')
    # 清空临时数据，结束对话
    context.user_data.clear()
    return ConversationHandler.END

# 取消对话的处理
async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('已取消呼叫操作')
    context.user_data.clear()
    return ConversationHandler.END

if __name__ == '__main__':
    # 替换成你自己的bot token
    app = ApplicationBuilder().token('YOUR_BOT_TOKEN').build()

    # 注册对话处理器
    call_conv_handler = ConversationHandler(
        entry_points=[CommandHandler('call', call_start)],
        states={
            WAIT_FROM: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_from_number)],
            WAIT_TO: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_to_number)]
        },
        fallbacks=[CommandHandler('cancel', cancel)]
    )
    app.add_handler(call_conv_handler)
    app.run_polling()

注意事项 #

• 不要用全局变量存用户输入的号码，多用户同时使用时会出现数据串号的问题，context.user_data是框架为每个用户单独隔离的临时存储，适合存对话过程中的中间数据
• 原来的写法之所以不会等输入，是因为单条消息触发handler后，函数内两次读取update.message.text拿到的都是触发这次函数的同一条消息内容，函数执行完就会释放，下一条用户消息会被框架重新分发给匹配的handler，不会回到之前的函数里
• 可以根据自己的需求补充更严格的号码校验、呼叫结果查询反馈逻辑

内容的提问来源于stack exchange，提问作者Ace