#!/usr/bin/env python3 """ Cron 管理 API 路由 """ from flask import jsonify, request from database import get_db from utils import ( log_message, sync_crontab_from_db, get_system_crons, cron_expression_to_human, get_next_run_time, guess_cron_name, guess_cron_category ) def register_cron_routes(app): """注册 Cron 相关路由""" @app.route('/api/cron/tasks') def api_cron_tasks(): """获取数据库中的 Cron 任务列表""" db = get_db() c = db.cursor() c.execute("SELECT * FROM cron_tasks ORDER BY created_at DESC") tasks = [dict(row) for row in c.fetchall()] # 添加人类可读描述和下次执行时间 for task in tasks: task['human_time'] = cron_expression_to_human(task['expression']) task['next_run'] = get_next_run_time(task['expression']) return jsonify({'tasks': tasks}) @app.route('/api/cron/tasks', methods=['POST']) def api_cron_add(): """添加新的 Cron 任务""" data = request.json if not data.get('name') or not data.get('expression') or not data.get('command'): return jsonify({'error': '缺少必要字段'}), 400 db = get_db() c = db.cursor() # 插入任务 c.execute('''INSERT INTO cron_tasks (name, expression, command, description, category, enabled) VALUES (?, ?, ?, ?, ?, ?)''', (data['name'], data['expression'], data['command'], data.get('description', ''), data.get('category', 'other'), 1)) task_id = c.lastrowid db.commit() # 记录历史（版本1） c.execute('''INSERT INTO cron_history (task_id, version, name, expression, command, description, category, enabled, operation) VALUES (?, 1, ?, ?, ?, ?, ?, ?, 'create')''', (task_id, data['name'], data['expression'], data['command'], data.get('description', ''), data.get('category', 'other'), 1)) db.commit() # 同步到系统 crontab sync_crontab_from_db(db) log_message(f"添加 Cron 任务: {data['name']} ({data['expression']})") return jsonify({'message': '任务添加成功', 'task_id': task_id}) @app.route('/api/cron/tasks/', methods=['PUT']) def api_cron_update(task_id): """更新 Cron 任务""" data = request.json db = get_db() c = db.cursor() # 获取当前任务 c.execute("SELECT * FROM cron_tasks WHERE id = ?", (task_id,)) old_task = c.fetchone() if not old_task: return jsonify({'error': '任务不存在'}), 404 # 获取当前版本号 c.execute("SELECT MAX(version) FROM cron_history WHERE task_id = ?", (task_id,)) max_version = c.fetchone()[0] or 0 new_version = max_version + 1 # 更新任务 update_fields = [] update_values = [] for field in ['name', 'expression', 'command', 'description', 'category', 'enabled']: if field in data: update_fields.append(f"{field} = ?") update_values.append(data[field]) if update_fields: update_fields.append("updated_at = CURRENT_TIMESTAMP") c.execute(f"UPDATE cron_tasks SET {','.join(update_fields)} WHERE id = ?", update_values + [task_id]) db.commit() # 记录历史 c.execute('''INSERT INTO cron_history (task_id, version, name, expression, command, description, category, enabled, operation) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'update')''', (task_id, new_version, data.get('name', old_task['name']), data.get('expression', old_task['expression']), data.get('command', old_task['command']), data.get('description', old_task['description']), data.get('category', old_task['category']), data.get('enabled', old_task['enabled']))) db.commit() # 同步到系统 crontab sync_crontab_from_db(db) log_message(f"更新 Cron 任务 ID={task_id}: 版本 {new_version}") return jsonify({'message': '任务更新成功', 'version': new_version}) @app.route('/api/cron/tasks/', methods=['DELETE']) def api_cron_delete(task_id): """删除 Cron 任务""" db = get_db() c = db.cursor() c.execute("SELECT * FROM cron_tasks WHERE id = ?", (task_id,)) task = c.fetchone() if not task: return jsonify({'error': '任务不存在'}), 404 # 记录删除历史 c.execute("SELECT MAX(version) FROM cron_history WHERE task_id = ?", (task_id,)) max_version = c.fetchone()[0] or 0 c.execute('''INSERT INTO cron_history (task_id, version, name, expression, command, description, category, enabled, operation) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'delete')''', (task_id, max_version + 1, task['name'], task['expression'], task['command'], task['description'], task['category'], task['enabled'])) db.commit() # 删除任务 c.execute("DELETE FROM cron_tasks WHERE id = ?", (task_id,)) db.commit() # 同步到系统 crontab sync_crontab_from_db(db) log_message(f"删除 Cron 任务 ID={task_id}: {task['name']}") return jsonify({'message': '任务删除成功'}) @app.route('/api/cron/tasks//toggle', methods=['POST']) def api_cron_toggle(task_id): """启用/禁用 Cron 任务""" db = get_db() c = db.cursor() c.execute("SELECT enabled FROM cron_tasks WHERE id = ?", (task_id,)) task = c.fetchone() if not task: return jsonify({'error': '任务不存在'}), 404 new_enabled = 0 if task['enabled'] == 1 else 1 c.execute("UPDATE cron_tasks SET enabled = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", (new_enabled, task_id)) db.commit() # 同步到系统 crontab sync_crontab_from_db(db) log_message(f"切换 Cron 任务 ID={task_id} 状态: {new_enabled}") return jsonify({'message': '状态已切换', 'enabled': new_enabled}) @app.route('/api/cron/tasks//history') def api_cron_history(task_id): """获取任务历史版本""" db = get_db() c = db.cursor() c.execute("SELECT * FROM cron_history WHERE task_id = ? ORDER BY version DESC", (task_id,)) history = [dict(row) for row in c.fetchall()] for h in history: if h['expression']: h['human_time'] = cron_expression_to_human(h['expression']) return jsonify({'history': history}) @app.route('/api/cron/tasks//rollback/', methods=['POST']) def api_cron_rollback(task_id, version): """回退到指定版本""" db = get_db() c = db.cursor() c.execute("SELECT * FROM cron_history WHERE task_id = ? AND version = ?", (task_id, version)) history = c.fetchone() if not history: return jsonify({'error': '版本不存在'}), 404 # 更新任务到历史版本 c.execute('''UPDATE cron_tasks SET name = ?, expression = ?, command = ?, description = ?, category = ?, enabled = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?''', (history['name'], history['expression'], history['command'], history['description'], history['category'], history['enabled'], task_id)) db.commit() # 同步到系统 crontab sync_crontab_from_db(db) log_message(f"回退 Cron 任务 ID={task_id} 到版本 {version}") return jsonify({'message': f'已回退到版本 {version}'}) @app.route('/api/cron/tasks//logs') def api_cron_logs(task_id): """获取任务执行日志""" db = get_db() c = db.cursor() c.execute("SELECT * FROM cron_logs WHERE task_id = ? ORDER BY execution_time DESC LIMIT 100", (task_id,)) logs = [dict(row) for row in c.fetchall()] return jsonify({'logs': logs}) @app.route('/api/cron/categories') def api_cron_categories(): """获取任务分类列表""" categories = [ {'id': 'monitor', 'name': '系统监控', 'icon': 'ri-heart-pulse-line', 'color': 'blue'}, {'id': 'backup', 'name': '数据备份', 'icon': 'ri-archive-line', 'color': 'green'}, {'id': 'cleanup', 'name': '清理任务', 'icon': 'ri-delete-bin-line', 'color': 'yellow'}, {'id': 'report', 'name': '报表生成', 'icon': 'ri-file-chart-line', 'color': 'purple'}, {'id': 'sync', 'name': '数据同步', 'icon': 'ri-refresh-line', 'color': 'cyan'}, {'id': 'notification', 'name': '通知提醒', 'icon': 'ri-notification-line', 'color': 'orange'}, {'id': 'other', 'name': '其他任务', 'icon': 'ri-more-line', 'color': 'gray'}, ] return jsonify({'categories': categories}) @app.route('/api/cron/sync', methods=['POST']) def api_cron_sync(): """从系统 crontab 同步到数据库（清空旧的，只保留当前系统实际存在的）""" crons = get_system_crons() user_crons = [c for c in crons if c['source'] == 'user'] db = get_db() c = db.cursor() # 清空数据库中的旧任务 c.execute("DELETE FROM cron_history WHERE task_id IN (SELECT id FROM cron_tasks)") c.execute("DELETE FROM cron_logs WHERE task_id IN (SELECT id FROM cron_tasks)") c.execute("DELETE FROM cron_tasks") synced_count = 0 for cron in user_crons: # 尝试从命令推断分类和名称 name = guess_cron_name(cron['command']) category = guess_cron_category(cron['command']) c.execute('''INSERT INTO cron_tasks (name, expression, command, description, category, enabled) VALUES (?, ?, ?, ?, ?, 1)''', (name, cron['expression'], cron['command'], '', category)) synced_count += 1 db.commit() log_message(f"从系统 crontab 同步了 {synced_count} 个任务（已清空旧任务）") return jsonify({'message': f'同步完成，共 {synced_count} 个任务', 'count': synced_count})