使用Python Telegram Bot的job_queue.run_repeating执行函数遇参数错误

错误原因 #

job_queue.run_repeating执行回调函数时，只会传入一个CallbackContext参数，但你的check_who_likes函数定义了update: Update和context: CallbackContext两个必填参数，导致参数不匹配抛出TypeError。另外定时任务场景下不存在Update对象（非用户触发消息），不能依赖update获取用户ID。

解决方案步骤 #

• 调整check_who_likes参数，仅保留context，从job的data中获取用户ID
• 修复SQL注入风险：IN子句改用参数化查询，避免字符串拼接
• 修正finally块错误逻辑：移除依赖update的判断，确保数据库连接/游标安全关闭
• 创建定时任务时，将用户ID传入job的data参数

修正后的代码 #

import asyncio
from telegram import Update
from telegram.ext import CallbackContext

async def check_who_likes(context: CallbackContext):
    user_id = context.job.data  # 从job数据中获取用户ID
    conn = None
    cursor = None
    try:
        conn = connection_pool.get_connection()
        cursor = conn.cursor()
        # 查询点赞当前用户的ID列表
        cursor.execute("SELECT LikeUserID FROM Likes WHERE LikedUserID = %s", (user_id,))
        likes = cursor.fetchall()
        like_ids = [like[0] for like in likes]
        
        if not like_ids:
            # 无点赞情况的提示
            await context.bot.send_message(chat_id=user_id, text="Henüz kimse seni beğenmedi.")
            return
        
        # 参数化处理IN子句，避免SQL注入
        placeholders = ', '.join(['%s'] * len(like_ids))
        sql_query = f"SELECT UserName FROM Users WHERE PersonID IN ({placeholders})"
        cursor.execute(sql_query, tuple(like_ids))
        users = cursor.fetchall()
        liked_users_text = ", ".join(str(user[0]) for user in users)
        
        await context.bot.send_message(
            chat_id=user_id,
            text=f"{len(likes)} kişi sizi beğendi.Bakmak istermisin?\n\n\n1.Göster.\n2.Artık aramıyorum."
        )
    except Exception as e:
        # 记录错误方便排查
        print(f"检查点赞列表出错: {str(e)}")
    finally:
        # 确保资源安全关闭
        if cursor:
            try:
                cursor.close()
            except:
                pass
        if conn:
            try:
                conn.close()
            except:
                pass
        await asyncio.sleep(1)

async def matching(update: Update, context: CallbackContext):
    user_id = update.message.from_user.id
    # 创建定时任务并传递用户ID
    context.job_queue.run_repeating(check_who_likes, interval=1, first=0, data=user_id)

额外说明 #

1. SQL注入防护：动态生成占位符配合参数元组的方式，彻底避免字符串拼接带来的注入风险
2. 资源安全管理：在finally块中分别判断游标和连接是否存在，再尝试关闭，避免初始化失败导致的二次错误
3. 边界场景处理：添加无点赞时的提示，避免空列表引发的SQL执行错误
4. 用户ID传递：通过job的data参数传递用户ID，解决定时任务无Update对象的核心问题

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