自动化任务调度器
监控文件夹变化并自动执行任务,失败自动重试与告警。
目标场景
生产与个人脚本中,文件落地即触发处理(转码、压缩、上传),或按时间表执行批处理任务。
核心功能
- 按文件事件触发(新增/修改/删除)
- 内置调度表与失败重试、并发限制
- 执行日志落盘并可选邮件/Webhook 告警
技术栈
开发代码示例
文件事件监听主循环
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class Handler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
print(f"[created] {event.src_path}")
dispatch(event.src_path)
observer = Observer()
observer.schedule(Handler(), watch_dir, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
定时任务调度
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
@sched.scheduled_job("interval", minutes=30)
def backup_job():
run_backup() # 你的业务函数
@sched.scheduled_job("cron", hour=3, minute=0)
def daily_report():
send_report()
sched.start()
来源与参考
- watchdog:跨平台文件系统事件监听库,本例事件循环直接基于它。
- APScheduler:Python 定时任务调度库,支持 cron / interval / date 触发。
开发操作步骤与流程
阶段:环境准备
- pip install watchdog apscheduler
- 创建任务函数与日志目录
阶段:联调
- 先监听测试目录确认事件触发
- 再叠加调度器,观察两个任务并行运行不互相阻塞