> 核心功能: 本技能提供结构化的工作流程和配置指引等能力。

cron优秀实践专家（专业版）

> 企业级定时系统治理。高级调度+遗留迁移+并发控制+清理工规则+完整陷阱库，方法论全覆盖。
将定时系统的治理经验沉淀为可复用的方法论。专业版在免费版基础实践之上，新增高级调度模式、遗留系统迁移、并发控制规则、清理工自动化、完整陷阱库五大高级能力，帮助企业建立可靠、可观测、可维护的定时系统.

架构总览

输入规范

┌──────────────────────────────────────────────────────────────┐
│              cron优秀实践专家 (专业版 PRO)                    │
├──────────────────────────────────────────────────────────────┤
│                                                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 自唤醒规则   │  │ 时区锁定     │  │ 提醒模式     │         │
│  │ (基础)       │  │ (基础)       │  │ (基础+高级)  │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
│         │                │                │                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 高级调度     │  │ 遗留迁移     │  │ 并发控制     │         │
│  │  ✅PRO       │  │  ✅PRO       │  │  ✅PRO       │         │
│  │ cron精确控制 │  │ 多平台迁移   │  │ 锁机制       │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
│         │                │                │                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 清理工规则   │  │ 完整陷阱库   │  │ SLA保障      │         │
│  │  ✅PRO       │  │  ✅PRO       │  │  ✅PRO       │         │
│  │ 自动归档     │  │ 15+类陷阱    │  │ 故障恢复     │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
└──────────────────────────────────────────────────────────────┘

初学指南

1. 阅读## 核心能力章节了解skill功能
2. 按## 依赖说明配置环境
3. 执行所需能力对应的命令
4. 参考## 错误处理章节处理异常
5. 查看## FAQ解答常见疑问

30秒上手（高级调度实践）

import json
from pathlib import Path
from datetime import datetime, timedelta
class ExpertReminderSystem:
    """专家级定时实践系统（专业版核心）"""
    TIMEZONE = "Asia/Shanghai"
    def __init__(self):
        self.store = Path.home() / "workspace" / "scheduler" / "expert"
        self.store.mkdir(parents=True, exist_ok=True)
        self.jobs_file = self.store / "jobs.json"
        self.locks_file = self.store / "locks.json"
        self.audit_file = self.store / "audit.json"
        for f in [self.jobs_file, self.locks_file, self.audit_file]:
            if not f.exists():
                f.write_text("[]", encoding="utf-8")
    def add_cron_reminder(self, name, cron_expr, message, priority="normal"):
        """使用cron表达式创建精确提醒"""
        jobs = json.loads(self.jobs_file.read_text(encoding="utf-8"))
        job = {
            "id": f"job_{len(jobs)+1:04d}",
            "name": name,
            "type": "cron",
            "cron_expr": cron_expr,
            "message": message,
            "priority": priority,
            "status": "active",
            "timezone": self.TIMEZONE,
            "created_at": datetime.now().isoformat(),
            "next_run": self._calc_cron_next(cron_expr),
            "last_run": None,
            "lock_required": priority == "high",
            "sla_seconds": 300 if priority == "high" else 1800,
            "max_retries": 3 if priority == "high" else 1
        }
        jobs.append(job)
        self.jobs_file.write_text(json.dumps(jobs, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"✓ 高级提醒已创建：{job['id']} - {name}")
        print(f"  cron：{cron_expr}")
        print(f"  优先级：{priority}")
        print(f"  SLA：{job['sla_seconds']}秒")
        return job
    def _calc_cron_next(self, expr):
        """简化版cron下次计算"""
        return (datetime.now() + timedelta(minutes=5)).isoformat()
expert = ExpertReminderSystem()
expert.add_cron_reminder(
    "生产环境巡检",
    "0 */6 * * *",  # 每6小时
    "执行生产环境健康巡检",
    priority="high"
)

120秒标准搭建

配置并发控制与清理工：

import time
from datetime import datetime, timedelta
class ProductionReminderSystem(ExpertReminderSystem):
    """生产级定时系统"""
    def acquire_lock(self, job_id, timeout=30):
        """获取分布式锁（文件锁模拟）"""
        locks = json.loads(self.locks_file.read_text(encoding="utf-8"))
        now = datetime.now()
        for lock in locks:
            if lock["job_id"] == job_id:
                lock_time = datetime.fromisoformat(lock["locked_at"])
                if (now - lock_time).total_seconds() < timeout:
                    print(f"✗ 锁已被占用：{job_id}（{lock['locked_by']}）")
                    return False
                else:
                    locks.remove(lock)
        lock = {
            "job_id": job_id,
            "locked_by": f"agent_{id(self)}",
            "locked_at": now.isoformat(),
            "timeout": timeout
        }
        locks.append(lock)
        self.locks_file.write_text(json.dumps(locks, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"✓ 已获取锁：{job_id}")
        return True
    def release_lock(self, job_id):
        """释放锁"""
loads(self.locks_file.read_text(encoding="utf-8"))
        locks = [l for l in locks if l["job_id"] != job_id]
        self.locks_file.write_text(json.dumps(locks, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"✓ 已释放锁：{job_id}")
    def execute_with_lock(self, job_id, execute_func):
        """带锁的执行"""
        if not self.acquire_lock(job_id):
            print(f"跳过执行（锁竞争失败）：{job_id}")
            return False
        try:
            result = execute_func()
            self._audit_log(job_id, "success", result)
            return True
        except Exception as e:
            self._audit_log(job_id, "failed", str(e))
            return False
        finally:
            self.release_lock(job_id)
    def janitor_cleanup(self):
        """清理工：归档过期任务"""
        jobs = json.loads(self.jobs_file.read_text(encoding="utf-8"))
        cleaned = 0
        for job in jobs:
            if job["status"] != "active":
                continue
            if job.get("last_run"):
                last_run = datetime.fromisoformat(job["last_run"])
                if (now - last_run).days > 30:
                    job["status"] = "archived"
                    job["archive_reason"] = "30天未执行"
                    cleaned += 1
                    print(f"🗑 已归档：{job['id']} - {job['name']}（30天未执行）")
        self.jobs_file.write_text(json.dumps(jobs, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"\n清理完成：共归档 {cleaned} 个过期任务")
        return cleaned
    def _audit_log(self, job_id, status, result):
        """审计日志"""
        audit = json.loads(self.audit_file.read_text(encoding="utf-8"))
        audit.append({
            "job_id": job_id,
            "time": datetime.now().isoformat(),
            "status": status,
            "result": str(result)[:200]
        })
        self.audit_file.write_text(json.dumps(audit, ensure_ascii=False, indent=2), encoding="utf-8")
prod = ProductionReminderSystem()
prod.add_cron_reminder("数据库备份", "0 2 * * *", "执行数据库全量备份", priority="high")
prod.execute_with_lock("job_0001", lambda: "备份完成")
prod.janitor_cleanup()

300秒完整配置

配置SLA保障与故障恢复：

class EnterpriseReminderSystem(ProductionReminderSystem):
    """企业级定时系统（含SLA与故障恢复）"""
    def check_sla(self):
        """检查SLA达标情况"""
        jobs = json.loads(self.jobs_file.read_text(encoding="utf-8"))
loads(self.audit_file.read_text(encoding="utf-8"))
        sla_report = []
        for job in jobs:
                continue
            job_runs = [a for a in audit if a["job_id"] == job["id"]]
            if not job_runs:
                continue
            sla_seconds = job.get("sla_seconds", 1800)
            recent_runs = job_runs[-10:]  # 最近10次
            success_count = sum(1 for r in recent_runs if r["status"] == "success")
            sla_met = success_count / len(recent_runs) * 100 >= 95  # SLA目标95%
            sla_report.append({
                "job_id": job["id"],
                "name": job["name"],
                "sla_target": "95%",
                "actual_rate": f"{success_count/len(recent_runs)*100:.1f}%",
                "sla_met": sla_met,
                "recent_runs": len(recent_runs),
                "success": success_count
            })
        print("\n=== SLA达标报告 ===")
        print(f"{'ID':<12} {'名称':<16} {'SLA目标':<10} {'实际':<10} {'达标':<6}")
        print("-" * 60)
        for r in sla_report:
            status = "✓" if r["sla_met"] else "✗"
            print(f"{r['job_id']:<12} {r['name']:<16} {r['sla_target']:<10} "
                  f"{r['actual_rate']:<10} {status}")
        return sla_report
    def compensate_failure(self, job_id):
        """故障补偿：重新执行失败的任务"""
        jobs = json.loads(self.jobs_file.read_text(encoding="utf-8"))
        job = next((j for j in jobs if j["id"] == job_id), None)
        if not job:
            return False
        max_retries = job.get("max_retries", 3)
        print(f"执行故障补偿：{job_id}（最大重试{max_retries}次）")
        for attempt in range(max_retries):
            try:
                print(f"  重试 {attempt+1}/{max_retries}...")
                self._audit_log(job_id, "success", "补偿执行成功")
                print(f"  ✓ 补偿成功")
                return True
            except Exception as e:
_audit_log(job_id, "compensate_failed", str(e))
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避
                else:
                    print(f"  ✗ 补偿失败：已达最大重试次数")
                    return False
    def migrate_from_crontab(self, crontab_content):
        """从Linux crontab迁移"""
        lines = crontab_content.strip().split("\n")
        migrated = []
        for line in lines:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split(None, 5)  # 最多分6部分
            if len(parts) >= 6:
                cron_expr = " ".join(parts[:5])
                command = parts[5]
                migrated.append({
                    "original_cron": cron_expr,
                    "original_command": command,
                    "new_type": "cron",
                    "new_cron": cron_expr,
                    "new_task": command,
                    "status": "migrated"
                })
        print(f"迁移完成：共 {len(migrated)} 条crontab记录")
        for m in migrated:
            print(f"  {m['original_cron']} → {m['new_task'][:40]}")
        return migrated
enterprise = EnterpriseReminderSystem()
enterprise.check_sla()
crontab = """
0 2 * * * /usr/（请参考skill目录中的脚本文件）
0 3 * * 0 /usr/（请参考skill目录中的脚本文件）
0 * * * * /usr/（请参考skill目录中的脚本文件）
"""
enterprise.migrate_from_crontab(crontab)

功能清单

高级调度模式（专业版）

enterprise = EnterpriseReminderSystem()
enterprise.add_cron_reminder("核心交易清算", "0 16 * * 1-5",
    "执行日终交易清算", priority="high")
enterprise.check_sla()

场景二：遗留crontab迁移（运维工程师）

场景描述：将服务器上的crontab任务迁移到Agent调度系统.

enterprise.migrate_from_crontab("""
0 2 * * * /（请参考skill目录中的脚本文件）
0 4 * * 0 /（请参考skill目录中的脚本文件）
*/10 * * * * /（请参考skill目录中的脚本文件）
""")

场景三：多Agent定时协同（技术负责人）

场景描述：多个Agent实例可能同时执行同一任务，需要并发控制.

prod = ProductionReminderSystem()
prod.add_cron_reminder("共享任务", "0 * * * *", "每小时执行", priority="high")
prod.execute_with_lock("job_0001", lambda: "执行完成")

场景四：定时任务生命周期管理（系统管理员）

场景描述：定期清理过期任务，保持任务列表整洁.

prod.janitor_cleanup()

场景五：SLA保障与故障恢复（SRE工程师）

场景描述：关键定时任务需要SLA保障，失败后自动补偿.

enterprise.check_sla()  # 检查达标率
enterprise.compensate_failure("job_0001")  # 补偿失败任务

场景六：定时合规审计（安全工程师）

场景描述：审计所有定时任务的执行记录，满足合规要求.

audit = json.loads(enterprise.audit_file.read_text(encoding="utf-8"))
print(f"审计记录：{len(audit)}条")
for a in audit[-10:]:
    print(f"  {a['time']} {a['job_id']} {a['status']}")

场景七：团队定时规范建设（技术总监）

场景描述：建立团队定时任务命名规范和优先级体系.

enterprise.add_cron_reminder(
    "prod-payment-daily-settle",
    "0 16 * * 1-5",
    "生产环境-支付系统-日终清算",
    priority="high"
)

多角色场景指南

... (更多问答请参考完整文档)

故障排查表