技术协助请求：实现score_user函数在指定时间的定时调度运行

我看了你的定时调度代码，确实存在几个关键问题导致定时功能无法正常工作，下面逐一分析并给出修复后的代码：

核心问题分析 #

• 错误的任务注册方式：你在调用schedule.do()时直接执行了score_user(msisdn)，这会导致函数立即执行，而不是被定时调度。do()方法需要传入函数对象，参数要单独传递。
• 静态的MSISDN数据：你的数据库查询只在程序启动时执行一次，如果每日的用户数据有更新，调度任务会一直使用旧的MSISDN，无法获取最新数据。
• 重复注册任务风险：如果查询结果res有多条数据，你会循环注册多个相同时间的定时任务，导致同一时间多次执行相同逻辑。

修复后的完整代码 #

import requests
import json
import schedule
import time
# 假设dbconn是已经初始化好的数据库连接对象

def score_user(msisdn):
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': 'Bearer 54763|E2jmjRM7xGmfhuWdUk2pO4SyHeDauvVer9sdCzGe'
    }
    payload = json.dumps({
        "id_number": "000",
        "msisdn": f"{msisdn}",
        "amount": "1500"
    })
    url = "https://preprod.senti.co.ke/api/v5/scoring/query"
    try:
        r = requests.post(url, headers=headers, data=payload)
        r.raise_for_status()  # 捕获HTTP请求错误
        print(f"Successfully scored {msisdn}: {r.text}")
        return r.text
    except Exception as e:
        print(f"Error scoring {msisdn}: {str(e)}")
        return None

def daily_scoring_task():
    """封装每日执行的完整任务逻辑：查询最新数据 + 批量评分"""
    cur = dbconn.cursor(dictionary=True)
    try:
        cur.execute('''
            select c.msisdn from azima._tmp_latest_loan a
            left join _tmp_latest_grade b on a.profile_id = b.profile_id
            left join v_profiles c on a.profile_id = c.profile_id
            order by a.pay_date desc limit 1
        ''')
        res = cur.fetchall()
        for row in res:
            msisdn = f"{row['msisdn']}"
            print(f"Starting scoring for: {msisdn}")
            score_user(msisdn)
    finally:
        cur.close()  # 确保游标关闭，避免资源泄漏

# 注册每日定时任务
schedule.every().day.at("16:56").do(daily_scoring_task)

print("Scheduler started, waiting for scheduled task...")
while True:
    schedule.run_pending()
    time.sleep(1)

关键改进点说明 #

1. 封装任务逻辑：把数据库查询和评分逻辑放到daily_scoring_task()函数中，确保每次定时执行时都会查询最新的MSISDN数据。
2. 正确注册任务：do(daily_scoring_task)传入的是函数对象，而非执行结果，这样schedule会在指定时间自动调用这个函数。
3. 异常处理：在score_user中添加了异常捕获，避免单个请求失败导致整个任务崩溃。
4. 资源清理：在任务函数中使用finally块确保数据库游标关闭，防止资源泄漏。

额外注意事项 #

• 确保你的程序能长期稳定运行（比如用nohup、systemd等方式后台运行，避免终端关闭后程序终止）。
• 如果你的数据库连接可能超时，可以考虑在daily_scoring_task中每次重新创建连接，或者添加连接保活逻辑。
• 可以添加日志记录（比如使用Python的logging模块），方便排查定时任务的执行情况。

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