定时事件需求：如何实现每日运行函数并支持传入参数？

针对每日仅运行一次的定时函数传参需求，下面是几种实用的实现方式：

1. 定时库直接绑定参数（以Python schedule为例） #

如果用脚本内部的定时库，直接在任务注册时传递参数即可，支持位置参数和关键字参数：

import schedule
import time

def daily_task(task_name, target_date):
    print(f"执行[{task_name}]任务，目标日期：{target_date}")

# 每日10点执行，直接传入固定参数
schedule.every().day.at("10:00").do(daily_task, task_name="日度数据同步", target_date="20240520")

while True:
    schedule.run_pending()
    time.sleep(60)

如果参数是动态生成的（比如当天日期），也可以在do()里传动态值：

from datetime import datetime
schedule.every().day.at("10:00").do(daily_task, task_name="日度报表", target_date=datetime.now().strftime("%Y%m%d"))

2. 系统级定时任务（如Linux cron）通过命令行传参 #

如果用系统自带的定时任务调度（比如cron），可以直接在调用脚本时通过命令行传递参数：

首先写可接收命令行参数的脚本daily_job.py：

import sys

def main():
    # 从命令行获取参数
    task_name = sys.argv[1]
    target_date = sys.argv[2]
    print(f"执行任务：{task_name}，日期：{target_date}")

if __name__ == "__main__":
    main()

然后在crontab中配置定时任务时传入参数：

# 每日10点执行，后面跟参数
0 10 * * * /usr/bin/python3 /home/user/scripts/daily_job.py "日度备份" "$(date +\%Y\%m\%d)"

这里$(date +\%Y\%m\%d)会动态生成当天日期作为参数，不需要手动修改。

3. 配置文件驱动（适合多参数/频繁修改场景） #

如果参数较多或者需要经常调整，把参数放到配置文件里，脚本运行时读取即可，避免修改代码：

比如创建task_config.json：

{
    "daily_task": {
        "task_name": "用户行为统计",
        "target_table": "user_behavior",
        "output_path": "/data/reports/daily/"
    }
}

然后脚本中读取配置并执行任务：

import schedule
import time
import json

def load_config():
    with open("task_config.json", "r", encoding="utf-8") as f:
        return json.load(f)

def daily_task():
    config = load_config()
    task_info = config["daily_task"]
    print(f"执行任务：{task_info['task_name']}，目标表：{task_info['target_table']}，输出路径：{task_info['output_path']}")

schedule.every().day.at("10:00").do(daily_task)

while True:
    schedule.run_pending()
    time.sleep(60)

后续修改参数只需要编辑task_config.json，不用改动脚本。

4. 中间函数中转动态参数 #

如果参数需要在任务触发时实时计算（比如依赖当前系统状态、数据库数据），可以写一个中间函数来生成参数，再调用目标任务函数：

import schedule
import time
import psutil

def daily_task(cpu_threshold, memory_threshold):
    print(f"监控任务：CPU阈值{cpu_threshold}%，内存阈值{memory_threshold}%")
    # 这里可以加具体的监控逻辑

def get_system_thresholds():
    # 从数据库或系统配置中实时获取阈值
    cpu_threshold = 80
    memory_threshold = 75
    return cpu_threshold, memory_threshold

def run_daily_task():
    cpu, mem = get_system_thresholds()
    daily_task(cpu, mem)

schedule.every().day.at("10:00").do(run_daily_task)

while True:
    schedule.run_pending()
    time.sleep(60)

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