关注公众号：weelinking | 访问官网：weelinking.com

📅 发布日期：2026年02月14日
🏷️ 标签：自动化 | 定时任务 | 工作流
⏱️ 阅读时长：20分钟
📚 系列文章：手机指挥电脑干活系列 - 第8篇

---

📄 文章摘要

前面几篇都是"你说，AI 做"的模式，但真正的高级玩法是让 AI 自己知道什么时候该做什么。本文从零搭建一套基于 Python + APScheduler 的自动化任务调度系统，实现每日晨间日报自动生成、代码质量定时检查、系统健康监控、Git 提交周报等 5 大高频自动化场景。还包含条件触发、链式任务、文件监控等进阶玩法，以及邮件/钉钉/Webhook 多渠道结果推送。读完本文，你的 AI 助手将从"被动执行"进化为"主动工作"。

关键词： #自动化 #定时任务 #APScheduler #工作流 #weelinking

---

一、前言:从"指挥 AI"到"AI 自己干"

前面几篇都是"你说,AI 做"的模式。

但更高级的玩法是:让 AI 自己知道什么时候该做什么。

比如:

• 每天早上 8 点自动生成工作日报
• 每小时检查服务器状态,异常时告警
• 每周五自动整理本周代码提交
• 文件变更时自动触发测试

真正的被动收入(划掉)——被动办公!

💡 本文所有操作均在 weelinking 上完成——国内可丝滑直连 Claude，无需魔法。

二、定时任务系统设计

2.1 实现方案

我们使用 Python + APScheduler 实现定时任务调度。

架构:

⏰ APScheduler 调度器
  ├─ Cron 触发器(定时执行)
  ├─ Interval 触发器(间隔执行)
  └─ Date 触发器(一次性任务)
      ↓
💻 调用 Claude API
      ↓
📱 结果推送到手机

2.2 创建调度服务

创建文件 scheduler_service.py:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
import anthropic
import os
from datetime import datetime
import json

# Claude 配置
API_KEY = "你的API_KEY"
BASE_URL = "https://api.weelinking.com/v1"

client = anthropic.Anthropic(api_key=API_KEY, base_url=BASE_URL)

WORK_DIR = os.path.expanduser("~/claude_workspace")

# 创建调度器
scheduler = BackgroundScheduler()

def call_claude(prompt):
    """调用 Claude 执行任务"""
    try:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text
    except Exception as e:
        return f"错误: {str(e)}"

def save_result(task_name, result):
    """保存任务结果"""
    log_dir = os.path.join(WORK_DIR, "task_logs")
    os.makedirs(log_dir, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    log_file = os.path.join(log_dir, f"{task_name}_{timestamp}.txt")

    with open(log_file, 'w', encoding='utf-8') as f:
        f.write(f"任务: {task_name}\n")
        f.write(f"时间: {datetime.now()}\n")
        f.write("-" * 50 + "\n")
        f.write(result)

    print(f"✓ {task_name} 完成,结果已保存: {log_file}")

# ========== 任务定义 ==========

def morning_report():
    """每日晨间工作日报"""
    prompt = f"""
    生成今日工作日报(Markdown 格式):

    1. 日期: {datetime.now().strftime('%Y-%m-%d')}
    2. 查看工作目录 {WORK_DIR} 下昨天修改的文件
    3. 生成今日待办事项建议
    4. 检查是否有未完成的任务

    格式:
    # 工作日报 - YYYY-MM-DD
    ## 昨日工作
    - ...

    ## 今日计划
    - ...

    ## 注意事项
    - ...
    """

    result = call_claude(prompt)
    save_result("morning_report", result)

    # 可选: 发送到手机(通过邮件/webhook等)
    # send_notification("早安!今日工作日报已生成", result)

def code_quality_check():
    """代码质量检查"""
    prompt = f"""
    对工作目录 {WORK_DIR} 进行代码质量检查:

    1. 扫描所有 .py/.js/.java 文件
    2. 检查:
       - 代码复杂度过高的函数
       - 过长的函数(>50行)
       - 潜在的 bug
       - TODO/FIXME 注释
    3. 生成质量报告

    格式化为 Markdown 表格输出
    """

    result = call_claude(prompt)
    save_result("code_quality", result)

def system_health_check():
    """系统健康检查"""
    prompt = """
    执行系统健康检查:

    1. 磁盘使用情况
    2. 内存使用情况
    3. 运行中的进程数
    4. 网络连接状态

    如果发现异常(如磁盘>90%),标记为【警告】
    """

    result = call_claude(prompt)
    save_result("health_check", result)

    # 检查是否有告警
    if "【警告】" in result:
        print("⚠️  发现系统异常,请及时处理!")

def git_summary():
    """Git 提交周报"""
    prompt = f"""
    生成本周 Git 提交总结:

    1. 切换到 {WORK_DIR}
    2. 执行 git log --since="1 week ago"
    3. 统计:
       - 总提交次数
       - 修改文件数
       - 新增/删除代码行数
       - 主要功能模块
    4. 生成 Markdown 格式的周报
    """

    result = call_claude(prompt)
    save_result("git_weekly", result)

def file_organizer():
    """文件整理助手"""
    prompt = f"""
    整理工作目录 {WORK_DIR}:

    1. 扫描所有文件
    2. 按类型分类:
       - 文档(.md, .txt, .pdf)
       - 代码(.py, .js, .java)
       - 数据(.csv, .json, .xml)
       - 其他
    3. 建议整理方案(不要直接执行,只是建议)
    4. 标记出超过 30 天未使用的文件

    输出整理建议清单
    """

    result = call_claude(prompt)
    save_result("file_organize", result)

# ========== 任务调度配置 ==========

# 每天早上 8:00 - 生成工作日报
scheduler.add_job(
    morning_report,
    CronTrigger(hour=8, minute=0),
    id="morning_report"
)

# 每 3 小时 - 系统健康检查
scheduler.add_job(
    system_health_check,
    IntervalTrigger(hours=3),
    id="health_check"
)

# 每天晚上 18:00 - 代码质量检查
scheduler.add_job(
    code_quality_check,
    CronTrigger(hour=18, minute=0),
    id="code_quality"
)

# 每周五 17:00 - Git 周报
scheduler.add_job(
    git_summary,
    CronTrigger(day_of_week='fri', hour=17, minute=0),
    id="git_weekly"
)

# 每周日 20:00 - 文件整理建议
scheduler.add_job(
    file_organizer,
    CronTrigger(day_of_week='sun', hour=20, minute=0),
    id="file_organize"
)

# ========== 主程序 ==========

def main():
    print("=" * 60)
    print("🤖 Claude 自动化任务调度器")
    print("=" * 60)
    print("已加载任务:")
    for job in scheduler.get_jobs():
        print(f"  ✓ {job.id}: {job.next_run_time}")
    print("=" * 60)

    # 启动调度器
    scheduler.start()

    print("调度器已启动,按 Ctrl+C 退出")

    try:
        # 保持程序运行
        import time
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("\n调度器已停止")

if __name__ == "__main__":
    main()

2.3 安装依赖

pip install apscheduler anthropic

2.4 启动调度服务

python scheduler_service.py

输出:

============================================================
🤖 Claude 自动化任务调度器
============================================================
已加载任务:
  ✓ morning_report: 2026-02-15 08:00:00
  ✓ health_check: 2026-02-14 12:00:00
  ✓ code_quality: 2026-02-14 18:00:00
  ✓ git_weekly: 2026-02-16 17:00:00
  ✓ file_organize: 2026-02-18 20:00:00
============================================================
调度器已启动,按 Ctrl+C 退出

现在,AI 会自动在设定的时间执行任务!

---

🔗 开始使用 Claude

文中的所有操作，你现在就可以上手。

推荐使用 weelinking 访问 Claude：

优势	说明
🇨🇳 国内直连	无需任何网络配置，打开即用
💰 按量付费	用多少花多少，不浪费
🎁 新用户福利	注册即送体验额度
⚡ 稳定可靠	账号池技术，不掉线

👉 立即注册，免费开始

---

三、5 个高频自动化场景

场景1:每日站会助手

def daily_standup():
    """生成站会发言稿"""
    prompt = """
    基于昨天的工作,生成今日站会发言稿:

    - 昨天完成: (从 Git 提交记录获取)
    - 今天计划: (从 TODO 列表获取)
    - 遇到的问题: (如果有)

    简洁专业,3 分钟内能说完
    """
    result = call_claude(prompt)
    save_result("daily_standup", result)

# 每个工作日早上 9:00
scheduler.add_job(
    daily_standup,
    CronTrigger(day_of_week='mon-fri', hour=9, minute=0)
)

场景2:依赖安全检查

def dependency_security_scan():
    """检查依赖包安全漏洞"""
    prompt = """
    扫描项目依赖的安全漏洞:

    1. 读取 requirements.txt 或 package.json
    2. 列出所有依赖包
    3. 检查是否有已知漏洞(基于常见漏洞库)
    4. 给出升级建议

    如果发现高危漏洞,标记【严重】
    """
    result = call_claude(prompt)
    save_result("security_scan", result)

    if "【严重】" in result:
        # 发送紧急通知
        print("🚨 发现严重安全漏洞!")

# 每天凌晨 2:00
scheduler.add_job(
    dependency_security_scan,
    CronTrigger(hour=2, minute=0)
)

场景3:学习笔记汇总

def weekly_learning_summary():
    """汇总本周学习笔记"""
    prompt = """
    汇总本周学习内容:

    1. 查找所有 .md 笔记文件
    2. 提取关键知识点
    3. 分类整理:
       - 技术知识
       - 问题解决
       - 最佳实践
    4. 生成思维导图(Markdown 格式)

    输出为精美的周学习总结
    """
    result = call_claude(prompt)
    save_result("learning_summary", result)

# 每周日晚上 21:00
scheduler.add_job(
    weekly_learning_summary,
    CronTrigger(day_of_week='sun', hour=21, minute=0)
)

场景4:数据备份自动化

def auto_backup():
    """自动备份重要文件"""
    prompt = f"""
    执行数据备份:

    1. 备份目标:
       - 所有 .py .js 源码文件
       - 配置文件 (.json, .yaml, .env)
       - 文档文件 (.md, .txt)

    2. 创建备份目录: backups/backup_YYYYMMDD/

    3. 复制文件并压缩

    4. 生成备份清单

    返回备份结果报告
    """
    result = call_claude(prompt)
    save_result("auto_backup", result)

# 每天凌晨 3:00
scheduler.add_job(
    auto_backup,
    CronTrigger(hour=3, minute=0)
)

场景5:性能监控报告

def performance_monitor():
    """性能监控与分析"""
    prompt = """
    生成性能监控报告:

    1. 分析应用日志,提取:
       - API 响应时间统计
       - 慢查询记录
       - 错误率统计

    2. 与昨天对比,找出异常

    3. 性能瓶颈分析

    4. 优化建议

    如果发现性能下降>20%,标记【性能告警】
    """
    result = call_claude(prompt)
    save_result("performance", result)

# 每 6 小时
scheduler.add_job(
    performance_monitor,
    IntervalTrigger(hours=6)
)

四、进阶玩法:条件触发

4.1 文件监控触发

使用 watchdog 监控文件变化:

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class CodeChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.py'):
            print(f"检测到代码变更: {event.src_path}")
            # 触发代码质量检查
            code_quality_check()

observer = Observer()
observer.schedule(CodeChangeHandler(), WORK_DIR, recursive=True)
observer.start()

4.2 条件触发任务

def smart_health_check():
    """智能健康检查"""
    # 执行检查
    prompt = "检查系统资源使用率"
    result = call_claude(prompt)

    # 根据结果决定下一步
    if "CPU > 80%" in result:
        # 触发性能分析
        call_claude("分析CPU高占用的进程,给出优化建议")

    if "磁盘 > 85%" in result:
        # 触发磁盘清理建议
        call_claude("找出大文件和临时文件,给出清理建议")

4.3 链式任务

def deploy_pipeline():
    """自动化部署流水线"""

    # 步骤1: 运行测试
    test_result = call_claude("运行所有单元测试")

    if "All tests passed" in test_result:
        # 步骤2: 构建项目
        build_result = call_claude("执行项目构建")

        if "Build successful" in build_result:
            # 步骤3: 部署
            deploy_result = call_claude("部署到测试环境")

            # 步骤4: 通知
            print("✓ 部署成功!")
        else:
            print("✗ 构建失败,停止部署")
    else:
        print("✗ 测试失败,停止流程")

五、任务结果推送

5.1 邮件推送

import smtplib
from email.mime.text import MIMEText

def send_email(subject, content):
    """发送邮件通知"""
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = 'your_email@example.com'
    msg['To'] = 'your_phone@example.com'  # 手机邮箱

    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('your_email', 'password')
        server.send_message(msg)

# 在任务中调用
def morning_report():
    result = call_claude(...)
    send_email("每日工作日报", result)

5.2 Webhook 推送

import requests

def send_to_phone(title, content):
    """通过 webhook 推送到手机"""
    webhook_url = "http://your-phone-ip:5000/api/notification"

    requests.post(webhook_url, json={
        "title": title,
        "content": content,
        "timestamp": datetime.now().isoformat()
    })

5.3 钉钉/企业微信推送

def send_to_dingtalk(content):
    """发送到钉钉机器人"""
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"

    data = {
        "msgtype": "text",
        "text": {"content": content}
    }

    requests.post(webhook, json=data)

六、运维与管理

6.1 任务管理 Web 界面

from flask import Flask, render_template, jsonify

app = Flask(__name__)

@app.route('/jobs')
def list_jobs():
    """列出所有任务"""
    jobs = []
    for job in scheduler.get_jobs():
        jobs.append({
            "id": job.id,
            "next_run": str(job.next_run_time),
            "trigger": str(job.trigger)
        })
    return jsonify(jobs)

@app.route('/job/<job_id>/run', methods=['POST'])
def run_job_now(job_id):
    """立即执行任务"""
    job = scheduler.get_job(job_id)
    if job:
        job.func()
        return jsonify({"success": True})
    return jsonify({"error": "Job not found"}), 404

@app.route('/job/<job_id>/pause', methods=['POST'])
def pause_job(job_id):
    """暂停任务"""
    scheduler.pause_job(job_id)
    return jsonify({"success": True})

app.run(port=5001)

访问 http://localhost:5001/jobs 查看和管理任务。

6.2 日志查看

@app.route('/logs')
def view_logs():
    """查看任务日志"""
    log_dir = os.path.join(WORK_DIR, "task_logs")
    files = sorted(os.listdir(log_dir), reverse=True)[:20]

    logs = []
    for file in files:
        with open(os.path.join(log_dir, file), 'r') as f:
            logs.append({
                "filename": file,
                "content": f.read()
            })

    return jsonify(logs)

6.3 后台运行

Linux/Mac:

# 使用 nohup
nohup python scheduler_service.py > scheduler.log 2>&1 &

# 或使用 systemd
sudo systemctl start claude-scheduler

Windows:

使用任务计划程序或创建Windows服务。

七、总结与下期预告

本期实现的自动化

• ✅ 每日工作日报自动生成
• ✅ 代码质量定时检查
• ✅ 系统健康监控
• ✅ Git 提交周报
• ✅ 文件整理建议
• ✅ 条件触发任务
• ✅ 结果推送到手机

你的 AI 助手现在能

• 自己知道什么时候该干什么
• 定时执行重复性任务
• 监控异常并告警
• 生成各类报表
• 24/7 不间断工作

下期预告

第9篇:多平台制霸——微信、钉钉、飞书全渠道接入

• 接入企业微信机器人
• 接入钉钉机器人
• 接入飞书机器人
• 多渠道统一管理
• 国内环境一键部署

不只是手机 App,微信、钉钉都能指挥 AI!

---

🔗 开始使用 Claude

文中的所有操作，你现在就可以上手。

推荐使用 weelinking 访问 Claude：

优势	说明
🇨🇳 国内直连	无需任何网络配置，打开即用
💰 按量付费	用多少花多少，不浪费
🎁 新用户福利	注册即送体验额度
⚡ 稳定可靠	账号池技术，不掉线

👉 立即注册，免费开始

---

🔜 下期预告

「多平台制霸——微信、钉钉、飞书全渠道接入」

不只是手机 App，微信、钉钉、飞书都能指挥 AI！下一篇实现企业微信机器人、钉钉机器人、飞书机器人接入，一个后端多个入口，Docker 一键部署，国内环境无缝使用。

👉 点击关注，下篇第一时间通知你。

---

📖 推荐阅读

如果这篇对你有帮助，以下文章你也会喜欢：

• [Claude Skill 从入门到精通 - 第7篇：Skill 的调试与测试方法论] — 系统化的调试测试方法
• [零基础用 Claude 开发应用 - 第4篇：用 Claude 开发一个自动化脚本工具箱] — 更多自动化开发实战
• [产品经理用 Claude 实现产品 - 第4篇：用 Claude 自动生成产品文档] — 自动化文档生成

---

📦 本篇配套资源

• ✅ 完整 scheduler_service.py 源码
• ✅ 5 个高频自动化场景代码模板
• ✅ APScheduler 配置速查表
• ✅ 多渠道推送配置指南
• 📷 需补充：调度器运行效果截图

---

📌 系列导航： 手机+OpenClaw+Claude实现远程AI编程系列大纲

⏩ 上一篇： 第7篇：通勤路上修 Bug——手机远程操控 Claude 写代码实战
⏩ 下一篇： 第9篇：多平台制霸_微信钉钉飞书全渠道接入

💡 本系列全程使用 weelinking 访问 Claude，国内可丝滑上手
🚀 整个系列的核心理念：你不需要变成程序员，你只需要从"找人做"变成"自己能做"。