通用并发定时任务管理器 - 支持资源监控、自检通知的Python定时任务管理框架

• ✅ 并发执行：支持多个任务并发运行，提高效率
• 📊 资源监控：实时监控CPU、内存、硬盘使用情况
• 🔔 智能通知：任务完成后自动发送通知，支持多种通知方式
• 🔁 自动重试：任务失败后自动重试，可配置重试次数和延迟
• ✅ 自检功能：任务执行后自动验证结果
• ⏰ 灵活调度：支持配置任务执行时间窗口
• 🎯 易于扩展：简单的任务基类，快速创建自定义任务

使用 uv 安装依赖：

# 安装uv（如果还没安装）
pip install uv

# 同步依赖
uv sync

# 初始化配置文件和目录结构
python -m task_scheduler.cli.commands init

这会创建：

• config.yaml - 配置文件
• tasks/ - 任务目录
• logs/ - 日志目录
• notifications/ - 通知目录
• examples/ - 示例任务

在 tasks/ 目录下创建你的任务文件：

from datetime import time
from task_scheduler.core.task import ScheduledTask
from typing import Dict, Any

class MyTask(ScheduledTask):
    @property
    def name(self) -> str:
        return "my_task"

    @property
    def schedule_time(self) -> time:
        return time(hour=0, minute=5)  # 凌晨0:05执行

    def execute(self) -> Dict[str, Any]:
        # 任务执行逻辑
        self.logger.info("执行任务...")
        return {"status": "completed"}

    def self_check(self) -> bool:
        # 自检逻辑
        return True

# 查看已注册的任务
python -m task_scheduler.cli.commands list

# 运行所有任务
python -m task_scheduler.cli.commands run

# 使用自定义配置文件
python -m task_scheduler.cli.commands run -c /path/to/config.yaml

# 资源监控配置
resource_monitor:
  enabled: true                # 是否启用资源监控
  interval: 5.0                # 采样间隔（秒）
  save_path: "./logs/resources.json"  # 监控数据保存路径

# 通知配置
notification:
  enabled: true                # 是否启用通知
  type: "file"                 # 通知类型
  directory: "./notifications" # 通知文件目录
  filename_prefix: "notification"  # 文件名前缀

# 调度配置
schedule:
  start_time: "00:00"          # 任务开始时间
  end_time: "08:00"            # 任务结束时间
  max_concurrent: 5            # 最大并发任务数

# 日志配置
logging:
  level: "INFO"                # 日志级别
  directory: "./logs"          # 日志目录
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

# 任务目录
task_dir: "./examples"         # 任务模块目录

所有任务必须继承 ScheduledTask 并实现以下方法：

from datetime import time
from task_scheduler.core.task import ScheduledTask
from typing import Dict, Any

class YourTask(ScheduledTask):
    @property
    def name(self) -> str:
        """任务唯一名称（必须）"""
        return "task_name"

    @property
    def schedule_time(self) -> time:
        """计划执行时间（必须）"""
        return time(hour=1, minute=30)

    @property
    def max_retries(self) -> int:
        """最大重试次数（可选，默认3）"""
        return 3

    @property
    def timeout(self) -> int:
        """超时时间（可选，默认不限制）"""
        return 300

    def execute(self) -> Dict[str, Any]:
        """任务执行逻辑（必须）"""
        # 你的代码
        return {"key": "value"}

    def pre_execute(self) -> None:
        """执行前钩子（可选）"""
        pass

    def post_execute(self, result) -> None:
        """执行后钩子（可选）"""
        pass

    def self_check(self) -> bool:
        """自检功能（可选）"""
        return True

    def on_failure(self, error: Exception) -> None:
        """失败处理（可选）"""
        pass

1. pre_execute() - 执行前准备
2. execute() - 执行任务（带重试机制）
3. self_check() - 自检验证
4. post_execute() - 执行后清理
5. 通知 - 发送执行结果通知

python -m task_scheduler.cli.commands windows-help

在午夜12点启动，每天执行：

schtasks /Create /TN "TaskScheduler" /TR "python -m task_scheduler.cli.commands run -c config.yaml" /SC DAILY /ST 00:00 /RU SYSTEM /F

# 查看任务
schtasks /Query /TN "TaskScheduler" /FO LIST

# 立即运行（测试）
schtasks /Run /TN "TaskScheduler"

# 取消注册
schtasks /Delete /TN "TaskScheduler" /F

• /TN - 任务名称
• /TR - 要执行的命令
• /SC - 调度类型（DAILY=每天）
• /ST - 开始时间
• /RU - 运行用户（SYSTEM=系统账户）
• /F - 强制创建

# 初始化配置
python -m task_scheduler.cli.commands init [-c CONFIG]

# 运行任务
python -m task_scheduler.cli.commands run [-c CONFIG]

# 列出任务
python -m task_scheduler.cli.commands list [-c CONFIG]

# 查看状态
python -m task_scheduler.cli.commands status [-c CONFIG]

# Windows帮助
python -m task_scheduler.cli.commands windows-help

auto_everyday/
├── src/task_scheduler/
│   ├── core/               # 核心模块
│   │   ├── task.py        # 任务基类
│   │   ├── manager.py     # 任务管理器
│   │   └── scheduler.py   # 调度器
│   ├── monitoring/        # 监控模块
│   │   └── resource.py    # 资源监控
│   ├── notifications/     # 通知模块
│   │   ├── base.py        # 通知基类
│   │   └── file_notifier.py # 文件通知
│   ├── config/            # 配置模块
│   │   └── settings.py    # 配置管理
│   └── cli/               # 命令行接口
│       └── commands.py    # CLI命令
├── examples/              # 示例任务
│   ├── sample_task.py
│   └── config.yaml
├── tasks/                 # 你的任务目录
├── logs/                  # 日志目录
└── notifications/         # 通知目录

uv run pytest

# 使用black格式化
uv run black src/

# 使用ruff检查
uv run ruff check src/

• Python 3.12+
• uv - 包管理和项目管理
• psutil - 系统资源监控
• PyYAML - 配置文件解析
• Click - 命令行接口

1. 时间窗口：任务默认在 00:00-08:00 执行，未执行完的任务会在人工介入后处理
2. 并发控制：默认最多5个任务并发执行，可在配置文件中调整
3. 资源监控：资源监控数据会自动保存为JSON格式
4. 日志文件：日志按日期分割，便于查找

MIT