# tasks/backup-database.yaml
name: "database-backup"
description: "MySQL 全量备份到远程 S3"
schedule: "0 3 * * *"          # 每天凌晨 3 点
priority: high                 # 高优先级
resources:
  cpu: 0.5                     # 预计占用 50% CPU
  memory_mb: 512               # 预计占用 512MB 内存
  io_weight: high              # 高 I/O 操作
  estimated_duration_min: 30   # 预计执行 30 分钟
dependencies: []               # 无依赖
after:                         # 必须在以下任务之后执行
  - "log-cleanup"
before:                        # 必须在以下任务之前执行
  - "backup-verify"
notifications:
  on_success: "log"
  on_failure: "telegram"
  on_timeout: "telegram"
command: "docker exec db-backup mysqldump --all-databases | aws s3 cp - s3://backups/$(date +\%Y\%m\%d).sql.gz"

# tasks/cleanup-temp.yaml
name: "log-cleanup"
description: "清理 7 天前的日志文件"
schedule: "30 3 * * *"         # 每天凌晨 3:30
priority: low                  # 低优先级
resources:
  cpu: 0.1
  memory_mb: 64
  io_weight: medium
  estimated_duration_min: 10
dependencies: []
after: []
before: []
notifications:
  on_success: "log"
  on_failure: "log"
  on_timeout: "log"
command: "find /var/log -name '*.log' -mtime +7 -delete"

# scheduler/task_registry.py
import yaml
import os
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class TaskResource:
    cpu: float          # CPU 权重 (0.0-1.0)
    memory_mb: int      # 内存占用 (MB)
    io_weight: str      # low / medium / high
    estimated_duration_min: int

@dataclass
class Task:
    name: str
    description: str
    schedule: str           # cron 表达式
    priority: str           # critical / high / medium / low
    resources: TaskResource
    dependencies: List[str]
    after: List[str]
    before: List[str]
    notifications: dict
    command: str
    created_at: datetime
    last_run: Optional[datetime]
    last_status: Optional[str]  # success / failure / timeout
    execution_count: int = 0
    avg_duration_min: float = 0.0

class TaskRegistry:
    def __init__(self, tasks_dir: str = "/etc/ai-scheduler/tasks"):
        self.tasks_dir = Path(tasks_dir)
        self.tasks: dict[str, Task] = {}
    
    def load_all(self):
        """从 YAML 文件加载所有任务"""
        if not self.tasks_dir.exists():
            self.tasks_dir.mkdir(parents=True)
        
        for yaml_file in self.tasks_dir.glob("*.yaml"):
            task = self._load_task(yaml_file)
            if task:
                self.tasks[task.name] = task
    
    def _load_task(self, yaml_path: Path) -> Optional[Task]:
        try:
            with open(yaml_path) as f:
                data = yaml.safe_load(f)
            
            return Task(
                name=data["name"],
                description=data.get("description", ""),
                schedule=data["schedule"],
                priority=data.get("priority", "medium"),
                resources=TaskResource(
                    cpu=data["resources"]["cpu"],
                    memory_mb=data["resources"]["memory_mb"],
                    io_weight=data["resources"]["io_weight"],
                    estimated_duration_min=data["resources"]["estimated_duration_min"],
                ),
                dependencies=data.get("dependencies", []),
                after=data.get("after", []),
                before=data.get("before", []),
                notifications=data.get("notifications", {}),
                command=data["command"],
                created_at=datetime.fromtimestamp(yaml_path.stat().st_mtime),
                last_run=None,
                last_status=None,
            )
        except Exception as e:
            print(f"[ERROR] Failed to load {yaml_path}: {e}")
            return None
    
    def get_conflicts(self) -> List[dict]:
        """检测任务之间的时间冲突"""
        conflicts = []
        task_list = list(self.tasks.values())
        
        for i, t1 in enumerate(task_list):
            for t2 in task_list[i+1:]:
                # 检查时间重叠
                if self._times_overlap(t1.schedule, t2.schedule):
                    conflicts.append({
                        "type": "time_overlap",
                        "tasks": [t1.name, t2.name],
                        "severity": self._calculate_severity(t1, t2),
                    })
                
                # 检查资源竞争
                if self._resources_conflict(t1.resources, t2.resources):
                    conflicts.append({
                        "type": "resource_conflict",
                        "tasks": [t1.name, t2.name],
                        "resource": "io" if t1.resources.io_weight == "high" or t2.resources.io_weight == "high" else "cpu",
                        "severity": "high" if t1.resources.io_weight == "high" and t2.resources.io_weight == "high" else "medium",
                    })
                
                # 检查依赖违反
                if t2.name in t1.after and t1.name in t2.before:
                    pass  # 依赖正确
                elif t2.name in t1.after:
                    conflicts.append({
                        "type": "dependency_violation",
                        "tasks": [t1.name, t2.name],
                        "issue": f"{t1.name} should run after {t2.name}",
                        "severity": "high",
                    })
        
        return conflicts
    
    def _times_overlap(self, schedule1: str, schedule2: str) -> bool:
        """简化版：检查 cron 表达式是否有重叠时间窗口"""
        # 实际实现需要使用 croniter 等库解析 cron 表达式
        # 这里简化处理：如果两个任务在同一小时内执行，认为有重叠风险
        return True  # 简化：假设同一时间段执行的任务可能有重叠

# docker-compose.scheduler.yml
version: '3.8'
services:
  task-executor:
    image: python:3.11-slim
    volumes:
      - ./tasks:/etc/ai-scheduler/tasks
      - ./logs:/var/log/ai-scheduler
      - ./scripts:/opt/scripts
    command: >
      bash -c "
        mkdir -p /var/log/ai-scheduler
        while true; do
          for task_file in /etc/ai-scheduler/tasks/*.yaml; do
            name=$(yq '.name' $task_file)
            cmd=$(yq '.command' $task_file)
            log_file=/var/log/ai-scheduler/\${name}.log
            echo \"\$(date -Iseconds) Starting $name\" >> \$log_file
            start_time=\$(date +%s)
            eval $cmd >> \$log_file 2>&1
            exit_code=\$?
            end_time=\$(date +%s)
            duration=\$((end_time - start_time))
            echo \"\$(date -Iseconds) Completed $name (exit=\$exit_code, duration=\${duration}s)\" >> \$log_file
          done
          sleep 60
        done
      "
    restart: unless-stopped
  
  llm-analyzer:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    restart: unless-stopped
  
  scheduler-api:
    build: ./scheduler-api
    ports:
      - "8080:8080"
    volumes:
      - ./tasks:/etc/ai-scheduler/tasks
      - ./logs:/var/log/ai-scheduler
      - ./config:/etc/ai-scheduler/config
    environment:
      - OLLAMA_HOST=http://llm-analyzer:11434
      - TASKS_DIR=/etc/ai-scheduler/tasks
      - LOGS_DIR=/var/log/ai-scheduler
    depends_on:
      - llm-analyzer
    restart: unless-stopped

volumes:
  ollama_data:

# scheduler-api/analyzer.py
import json
import re
from pathlib import Path
from datetime import datetime, timedelta
import requests

class TaskBehaviorAnalyzer:
    def __init__(self, ollama_url: str = "http://localhost:11434", model: str = "qwen2.5:7b"):
        self.ollama_url = ollama_url
        self.model = model
        self.logs_dir = Path("/var/log/ai-scheduler")
    
    def analyze_task(self, task_name: str) -> dict:
        """分析单个任务的行为模式"""
        log_file = self.logs_dir / f"{task_name}.log"
        if not log_file.exists():
            return {"error": "No log file found"}
        
        # 解析执行日志
        executions = self._parse_log(log_file)
        if len(executions) < 3:
            return {"error": "Insufficient execution data"}
        
        # 计算统计信息
        stats = self._compute_stats(executions)
        
        # 调用 LLM 生成行为分析
        llm_analysis = self._llm_analyze(task_name, stats)
        
        return {
            "task_name": task_name,
            "stats": stats,
            "llm_analysis": llm_analysis,
            "updated_at": datetime.now().isoformat(),
        }
    
    def _parse_log(self, log_file: Path) -> list:
        """解析任务执行日志"""
        executions = []
        pattern = r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})\s+(Starting|Completed)\s+(\S+)\s+\(exit=(\d+),\s+duration=(\d+)s\)'
        
        with open(log_file) as f:
            lines = f.readlines()
        
        i = 0
        while i < len(lines) - 1:
            match = re.match(pattern, lines[i])
            if match and 'Starting' in lines[i]:
                start_time = match.group(1)
                # Find corresponding completion
                for j in range(i+1, min(i+5, len(lines))):
                    comp_match = re.match(pattern, lines[j])
                    if comp_match and 'Completed' in lines[j]:
                        executions.append({
                            "start_time": start_time,
                            "exit_code": int(comp_match.group(4)),
                            "duration_sec": int(comp_match.group(5)),
                        })
                        i = j
                        break
            i += 1
        
        return executions
    
    def _compute_stats(self, executions: list) -> dict:
        """计算执行统计信息"""
        durations = [e["duration_sec"] for e in executions]
        failures = sum(1 for e in executions if e["exit_code"] != 0)
        
        # 按小时统计
        hour_stats = {}
        for e in executions:
            hour = e["start_time"][11:13]
            if hour not in hour_stats:
                hour_stats[hour] = {"count": 0, "total_duration": 0, "failures": 0}
            hour_stats[hour]["count"] += 1
            hour_stats[hour]["total_duration"] += e["duration_sec"]
            if e["exit_code"] != 0:
                hour_stats[hour]["failures"] += 1
        
        return {
            "total_executions": len(executions),
            "avg_duration_sec": sum(durations) / len(durations) if durations else 0,
            "max_duration_sec": max(durations) if durations else 0,
            "min_duration_sec": min(durations) if durations else 0,
            "failure_rate": failures / len(executions) if executions else 0,
            "hour_stats": hour_stats,
        }
    
    def _llm_analyze(self, task_name: str, stats: dict) -> dict:
        """调用 LLM 生成行为分析"""
        prompt = f"""你是一个 VPS 运维专家。请分析以下定时任务的执行数据，并给出优化建议。

任务名称: {task_name}
执行统计:
- 总执行次数: {stats['total_executions']}
- 平均执行时长: {stats['avg_duration_sec']:.0f} 秒
- 最大执行时长: {stats['max_duration_sec']} 秒
- 最小执行时长: {stats['min_duration_sec']} 秒
- 失败率: {stats['failure_rate']*100:.1f}%

小时分布:
{json.dumps(stats['hour_stats'], indent=2, ensure_ascii=False)}

请分析：
1. 该任务的执行模式（是否有周期性波动）
2. 是否存在执行时长异常（某个时段特别慢）
3. 失败模式是什么（是否有规律）
4. 推荐的最佳执行时间段
5. 可能的优化建议

请用 JSON 格式返回分析结果：
{{
  "pattern": "描述执行模式",
  "anomalies": ["异常1", "异常2"],
  "optimal_time": "推荐执行时间",
  "optimization_suggestions": ["建议1", "建议2"],
  "risk_level": "low/medium/high"
}}"""
        
        try:
            response = requests.post(
                f"{self.ollama_url}/api/generate",
                json={"model": self.model, "prompt": prompt, "stream": False},
                timeout=30
            )
            return response.json().get("response", "{}")
        except Exception as e:
            return f"LLM analysis failed: {e}"

# scheduler/conflict_detector.py
from datetime import datetime, timedelta
from typing import List, Tuple

class ConflictDetector:
    def __init__(self, tasks: dict, resource_limits: dict):
        self.tasks = tasks
        self.resource_limits = resource_limits  # 系统总资源限制
        self.conflicts: List[dict] = []
    
    def detect_all(self) -> List[dict]:
        """检测所有冲突"""
        self.conflicts = []
        
        task_list = list(self.tasks.values())
        
        for i, t1 in enumerate(task_list):
            for t2 in task_list[i+1:]:
                # 时间重叠检测
                overlap = self._check_time_overlap(t1, t2)
                if overlap:
                    self.conflicts.append({
                        "type": "time_overlap",
                        "tasks": [t1.name, t2.name],
                        "overlap_window": overlap,
                        "severity": self._severity(t1, t2, "time"),
                    })
                
                # 资源竞争检测
                resource_conflict = self._check_resource_conflict(t1, t2)
                if resource_conflict:
                    self.conflicts.append({
                        "type": "resource_conflict",
                        "tasks": [t1.name, t2.name],
                        "resource": resource_conflict,
                        "severity": "high" if resource_conflict == "io" else "medium",
                    })
                
                # 依赖检测
                dep_conflict = self._check_dependency_conflict(t1, t2)
                if dep_conflict:
                    self.conflicts.append({
                        "type": "dependency_conflict",
                        "tasks": [t1.name, t2.name],
                        "issue": dep_conflict,
                        "severity": "high",
                    })
        
        return self.conflicts
    
    def _check_time_overlap(self, t1, t2) -> Optional[str]:
        """检查两个任务的时间重叠"""
        # 简化版：如果两个任务的执行窗口有重叠，返回重叠时间段
        # 实际实现需要解析 cron 表达式
        return "03:00-03:30"  # 示例：两个任务都在凌晨3点执行
    
    def _check_resource_conflict(self, t1, t2) -> Optional[str]:
        """检查资源竞争"""
        conflicts = []
        
        # I/O 竞争
        if t1.resources.io_weight == "high" and t2.resources.io_weight == "high":
            conflicts.append("io")
        
        # CPU 竞争
        total_cpu = t1.resources.cpu + t2.resources.cpu
        if total_cpu > 0.8:
            conflicts.append("cpu")
        
        # 内存竞争
        total_mem = t1.resources.memory_mb + t2.resources.memory_mb
        if total_mem > self.resource_limits.get("memory_mb", 2048) * 0.7:
            conflicts.append("memory")
        
        return conflicts[0] if conflicts else None
    
    def _check_dependency_conflict(self, t1, t2) -> Optional[str]:
        """检查依赖关系冲突"""
        if t2.name in t1.after and t1.name not in t2.before:
            return f"{t1.name} should run after {t2.name}, but {t2.name} doesn't expect {t1.name} after it"
        if t1.name in t2.after and t2.name not in t1.before:
            return f"{t2.name} should run after {t1.name}, but {t1.name} doesn't expect {t2.name} after it"
        return None
    
    def _severity(self, t1, t2, conflict_type: str) -> str:
        """计算冲突严重性"""
        if conflict_type == "time" and (t1.resources.io_weight == "high" or t2.resources.io_weight == "high"):
            return "high"
        return "medium"

# scheduler/scheduler_optimizer.py
import requests

class SchedulerOptimizer:
    def __init__(self, ollama_url: str, model: str = "qwen2.5:7b"):
        self.ollama_url = ollama_url
        self.model = model
    
    def optimize_schedule(self, tasks: dict, conflicts: List[dict]) -> dict:
        """生成优化后的调度方案"""
        if not conflicts:
            return {"optimized": False, "changes": [], "reason": "No conflicts detected"}
        
        # 构建冲突摘要
        conflict_summary = []
        for c in conflicts:
            conflict_summary.append(
                f"- {c['type']}: {', '.join(c['tasks'])} (severity: {c['severity']})"
            )
        
        prompt = f"""你是一个 VPS 运维调度专家。以下定时任务存在冲突，请给出优化建议。

当前任务列表:
{self._format_tasks(tasks)}

 detected 冲突:
{chr(10).join(conflict_summary)}

系统资源限制:
- CPU: 4 cores
- 内存: 4GB
- 磁盘 I/O: 中等

请生成优化后的调度方案，格式如下：
{{
  "optimized": true,
  "changes": [
    {{
      "task": "任务名",
      "original_schedule": "原 cron 表达式",
      "optimized_schedule": "优化后的 cron 表达式",
      "reason": "优化原因"
    }}
  ],
  "explanation": "整体优化策略说明"
}}"""
        
        try:
            response = requests.post(
                f"{self.ollama_url}/api/generate",
                json={"model": self.model, "prompt": prompt, "stream": False},
                timeout=60
            )
            return response.json().get("response", "{}")
        except Exception as e:
            return {"error": str(e)}
    
    def _format_tasks(self, tasks: dict) -> str:
        """格式化任务列表"""
        lines = []
        for name, task in tasks.items():
            lines.append(f"- {name}: {task.schedule} (priority: {task.priority}, io: {task.resources.io_weight})")
        return chr(10).join(lines)

# /etc/systemd/system/ai-scheduler@.timer
[Unit]
Description=AI Scheduler Timer for %i

[Timer]
OnCalendar=*-*-* *:00:00
Persistent=true
Unit=ai-scheduler@.service

[Install]
WantedBy=timers.target

# /etc/systemd/system/ai-scheduler@.service
[Unit]
Description=AI Scheduler Service for %i
After=network-online.target

[Service]
Type=oneshot
User=root
WorkingDirectory=/opt/ai-scheduler
Environment=TASK_NAME=%i
ExecStart=/opt/ai-scheduler/run-task.sh
TimeoutStartSec=3600

# 资源限制
CPUQuota=50%
MemoryMax=512M
IOWeight=100

# scheduler/resource_manager.py
import subprocess
import psutil
from typing import List

class ResourceManager:
    def __init__(self, tasks: dict, thresholds: dict = None):
        self.tasks = tasks
        self.thresholds = thresholds or {
            "cpu_warning": 0.7,
            "cpu_critical": 0.9,
            "memory_warning": 0.75,
            "memory_critical": 0.9,
            "io_warning": 0.7,
        }
    
    def check_resources(self) -> dict:
        """检查当前系统资源状态"""
        cpu_percent = psutil.cpu_percent(interval=1)
        memory = psutil.virtual_memory()
        disk_io = self._get_disk_io()
        
        return {
            "cpu_percent": cpu_percent,
            "memory_percent": memory.percent,
            "disk_read_mb": disk_io.read_mb,
            "disk_write_mb": disk_io.write_mb,
            "status": self._assess_status(cpu_percent, memory.percent, disk_io),
        }
    
    def _assess_status(self, cpu: float, memory: float, io) -> str:
        """评估系统状态"""
        if cpu > self.thresholds["cpu_critical"] or memory.percent > self.thresholds["memory_critical"]:
            return "critical"
        elif cpu > self.thresholds["cpu_warning"] or memory.percent > self.thresholds["memory_warning"]:
            return "warning"
        return "normal"
    
    def suspend_low_priority_tasks(self, status: str):
        """根据系统状态暂停低优先级任务"""
        if status != "critical":
            return
        
        for name, task in self.tasks.items():
            if task.priority in ("low", "medium"):
                self._suspend_task(name)
                print(f"[RESOURCE] Suspended low-priority task: {name}")
    
    def _suspend_task(self, task_name: str):
        """暂停任务"""
        subprocess.run(
            ["systemctl", "stop", f"ai-scheduler@{task_name}.timer"],
            capture_output=True
        )
    
    def resume_tasks(self):
        """恢复所有任务"""
        for name in self.tasks:
            subprocess.run(
                ["systemctl", "start", f"ai-scheduler@{name}.timer"],
                capture_output=True
            )

# docker-compose.yml
version: '3.8'
services:
  scheduler-api:
    build: ./scheduler-api
    container_name: ai-scheduler
    volumes:
      - ./tasks:/etc/ai-scheduler/tasks
      - ./logs:/var/log/ai-scheduler
      - ./config:/etc/ai-scheduler/config
      - /etc/systemd/system:/etc/systemd/system:ro
    environment:
      - OLLAMA_HOST=http://ollama:11434
      - TASKS_DIR=/etc/ai-scheduler/tasks
      - LOGS_DIR=/var/log/ai-scheduler
      - API_PORT=8080
    ports:
      - "8080:8080"
    depends_on:
      - ollama
    restart: unless-stopped
    networks:
      - scheduler-net
  
  ollama:
    image: ollama/ollama:latest
    container_name: ollama
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    restart: unless-stopped
    networks:
      - scheduler-net
  
  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    volumes:
      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    ports:
      - "9090:9090"
    restart: unless-stopped
    networks:
      - scheduler-net

volumes:
  ollama_data:
  prometheus_data:

networks:
  scheduler-net:
    driver: bridge

# 拉取适合推理的模型
docker exec -it ollama ollama pull qwen2.5:7b

# 启动调度器
docker-compose up -d

# 加载任务
curl -X POST http://localhost:8080/tasks/load

# 检测冲突
curl http://localhost:8080/conflicts

# 获取优化建议
curl -X POST http://localhost:8080/scheduler/optimize

# AI 生成的优化调度
tasks:
  - name: "log-cleanup"
    schedule: "0 2 * * *"      # 最早执行，清理日志释放空间
    priority: low
  
  - name: "docker-cleanup"
    schedule: "15 2 * * *"     # 清理 Docker，释放磁盘
    priority: medium
  
  - name: "rsync-sync"
    schedule: "30 2 * * *"     # 同步数据，此时磁盘空间已释放
    priority: high
  
  - name: "database-backup"
    schedule: "0 3 * * *"      # 最后执行备份，确保数据最新
    priority: critical