# -*- coding: utf-8 -*- import json from datetime import datetime from typing import Any, Dict, List, Optional from loguru import logger from db.base import BaseDBOperator from db.connection import DBConnectionManager class PluginScheduleDBOperator(BaseDBOperator): """插件定时任务配置与日志表操作。""" def __init__(self, db_manager: DBConnectionManager): super().__init__(db_manager) def init_tables(self) -> bool: try: self.execute_update( """ CREATE TABLE IF NOT EXISTS t_plugin_schedules ( id BIGINT PRIMARY KEY AUTO_INCREMENT, plugin_name VARCHAR(128) NOT NULL, action_key VARCHAR(64) NOT NULL, action_name VARCHAR(128) NOT NULL, description VARCHAR(255) DEFAULT '', trigger_type VARCHAR(64) NOT NULL, trigger_config JSON NOT NULL, target_scope VARCHAR(64) NOT NULL DEFAULT 'all_enabled_groups', target_config JSON DEFAULT NULL, payload JSON DEFAULT NULL, enabled TINYINT(1) NOT NULL DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_plugin_action (plugin_name, action_key) ) """ ) self.execute_update( """ CREATE TABLE IF NOT EXISTS t_plugin_schedule_logs ( id BIGINT PRIMARY KEY AUTO_INCREMENT, schedule_id BIGINT NOT NULL, triggered_at DATETIME DEFAULT CURRENT_TIMESTAMP, status VARCHAR(32) NOT NULL, summary VARCHAR(255) DEFAULT '', detail_json JSON DEFAULT NULL, INDEX idx_schedule_time (schedule_id, triggered_at) ) """ ) return True except Exception as e: logger.error(f"初始化插件调度表失败: {e}") return False @staticmethod def _parse_json_field(row: Dict[str, Any], key: str): value = row.get(key) if isinstance(value, str): try: row[key] = json.loads(value) except json.JSONDecodeError: row[key] = {} elif value is None: row[key] = {} def list_schedules(self) -> List[Dict[str, Any]]: rows = self.execute_query("SELECT * FROM t_plugin_schedules ORDER BY plugin_name, action_name") or [] for row in rows: self._parse_json_field(row, "trigger_config") self._parse_json_field(row, "target_config") self._parse_json_field(row, "payload") return rows def list_enabled_schedules(self) -> List[Dict[str, Any]]: rows = self.execute_query( "SELECT * FROM t_plugin_schedules WHERE enabled = 1 ORDER BY plugin_name, action_name" ) or [] for row in rows: self._parse_json_field(row, "trigger_config") self._parse_json_field(row, "target_config") self._parse_json_field(row, "payload") return rows def get_schedule(self, schedule_id: int) -> Optional[Dict[str, Any]]: row = self.execute_query( "SELECT * FROM t_plugin_schedules WHERE id = %s", (schedule_id,), fetch_one=True, ) if not row: return None self._parse_json_field(row, "trigger_config") self._parse_json_field(row, "target_config") self._parse_json_field(row, "payload") return row def get_schedule_by_plugin_action(self, plugin_name: str, action_key: str) -> Optional[Dict[str, Any]]: """按插件名+动作键查询调度配置。""" row = self.execute_query( """ SELECT * FROM t_plugin_schedules WHERE plugin_name = %s AND action_key = %s LIMIT 1 """, (plugin_name, action_key), fetch_one=True, ) if not row: return None self._parse_json_field(row, "trigger_config") self._parse_json_field(row, "target_config") self._parse_json_field(row, "payload") return row def upsert_default_schedule(self, data: Dict[str, Any]) -> bool: try: sql = """ INSERT INTO t_plugin_schedules ( plugin_name, action_key, action_name, description, trigger_type, trigger_config, target_scope, target_config, payload, enabled ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE action_name = VALUES(action_name), description = VALUES(description) """ params = ( data["plugin_name"], data["action_key"], data["action_name"], data.get("description", ""), data["trigger_type"], json.dumps(data.get("trigger_config", {}), ensure_ascii=False), data.get("target_scope", "all_enabled_groups"), json.dumps(data.get("target_config", {}), ensure_ascii=False), json.dumps(data.get("payload", {}), ensure_ascii=False), 1 if data.get("enabled", False) else 0, ) return self.execute_update(sql, params) except Exception as e: logger.error(f"upsert 插件默认调度失败: {e}, data={data}") return False def update_schedule(self, schedule_id: int, updates: Dict[str, Any]) -> bool: fields = [] values = [] for key in ( "action_name", "description", "trigger_type", "target_scope", "enabled", ): if key in updates: fields.append(f"{key} = %s") if key == "enabled": values.append(1 if updates[key] else 0) else: values.append(updates[key]) for key in ("trigger_config", "target_config", "payload"): if key in updates: fields.append(f"{key} = %s") values.append(json.dumps(updates.get(key, {}), ensure_ascii=False)) if not fields: return True values.append(schedule_id) sql = f"UPDATE t_plugin_schedules SET {', '.join(fields)} WHERE id = %s" return self.execute_update(sql, tuple(values)) def create_log(self, schedule_id: int, status: str, summary: str, detail: Dict[str, Any]) -> bool: sql = """ INSERT INTO t_plugin_schedule_logs (schedule_id, status, summary, detail_json) VALUES (%s, %s, %s, %s) """ params = ( schedule_id, status, summary, json.dumps(detail or {}, ensure_ascii=False), ) return self.execute_update(sql, params) def get_logs(self, schedule_id: int, limit: int = 100) -> List[Dict[str, Any]]: rows = self.execute_query( """ SELECT * FROM t_plugin_schedule_logs WHERE schedule_id = %s ORDER BY triggered_at DESC LIMIT %s """, (schedule_id, int(limit)), ) or [] for row in rows: self._parse_json_field(row, "detail_json") return rows def get_latest_log_time(self, schedule_id: int) -> Optional[datetime]: """获取调度任务最新一次执行日志时间。""" row = self.execute_query( """ SELECT triggered_at FROM t_plugin_schedule_logs WHERE schedule_id = %s ORDER BY triggered_at DESC LIMIT 1 """, (int(schedule_id),), fetch_one=True, ) or {} return row.get("triggered_at") @staticmethod def _clean_schedule_ids(schedule_ids: List[int]) -> List[int]: """清洗批量查询用的调度 ID 列表。""" clean_ids: List[int] = [] seen = set() for item in schedule_ids or []: text = str(item or "").strip() if not text.isdigit(): continue schedule_id = int(text) if schedule_id in seen: continue clean_ids.append(schedule_id) seen.add(schedule_id) return clean_ids def get_latest_logs_map(self, schedule_ids: List[int]) -> Dict[int, Dict[str, Any]]: """批量获取每个调度任务最新一条执行日志。 设计说明： 1. 后台任务页展示“上次执行时间/状态”时，不能只依赖内存态； 2. 进程重启后，async_job 的运行时计数会重置，但数据库日志仍完整； 3. 这里提供批量查询接口，让上层可用日志数据兜底回填展示字段。 """ clean_ids = self._clean_schedule_ids(schedule_ids) if not clean_ids: return {} placeholders = ",".join(["%s"] * len(clean_ids)) sql = f""" SELECT l.* FROM t_plugin_schedule_logs l INNER JOIN ( SELECT schedule_id, MAX(id) AS max_id FROM t_plugin_schedule_logs WHERE schedule_id IN ({placeholders}) GROUP BY schedule_id ) t ON l.id = t.max_id """ rows = self.execute_query(sql, tuple(clean_ids)) or [] result: Dict[int, Dict[str, Any]] = {} for row in rows: self._parse_json_field(row, "detail_json") schedule_id = int(row.get("schedule_id") or 0) if schedule_id > 0: result[schedule_id] = row return result def get_schedule_history_summary_map(self, schedule_ids: List[int]) -> Dict[int, Dict[str, Any]]: """批量汇总调度任务的历史执行摘要。""" clean_ids = self._clean_schedule_ids(schedule_ids) if not clean_ids: return {} placeholders = ",".join(["%s"] * len(clean_ids)) summary_sql = f""" SELECT schedule_id, MAX(CASE WHEN status = 'success' THEN triggered_at ELSE NULL END) AS latest_success_at, MAX(CASE WHEN status = 'failed' THEN triggered_at ELSE NULL END) AS latest_failed_at, SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success_count, SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS fail_count, COUNT(*) AS total_count FROM t_plugin_schedule_logs WHERE schedule_id IN ({placeholders}) GROUP BY schedule_id """ latest_failed_sql = f""" SELECT l.* FROM t_plugin_schedule_logs l INNER JOIN ( SELECT schedule_id, MAX(id) AS max_id FROM t_plugin_schedule_logs WHERE status = 'failed' AND schedule_id IN ({placeholders}) GROUP BY schedule_id ) t ON l.id = t.max_id """ summary_rows = self.execute_query(summary_sql, tuple(clean_ids)) or [] latest_failed_rows = self.execute_query(latest_failed_sql, tuple(clean_ids)) or [] result: Dict[int, Dict[str, Any]] = {} for row in summary_rows: schedule_id = int(row.get("schedule_id") or 0) if schedule_id <= 0: continue result[schedule_id] = { "latest_success_at": row.get("latest_success_at"), "latest_failed_at": row.get("latest_failed_at"), "latest_failure_summary": "", "latest_failure_detail": {}, "history_success_count": int(row.get("success_count") or 0), "history_fail_count": int(row.get("fail_count") or 0), "history_total_count": int(row.get("total_count") or 0), } for row in latest_failed_rows: schedule_id = int(row.get("schedule_id") or 0) if schedule_id <= 0: continue detail = row.get("detail_json") if isinstance(detail, str): try: detail = json.loads(detail) except json.JSONDecodeError: detail = {} elif detail is None: detail = {} history = result.setdefault( schedule_id, { "latest_success_at": None, "latest_failed_at": row.get("triggered_at"), "latest_failure_summary": "", "latest_failure_detail": {}, "history_success_count": 0, "history_fail_count": 0, "history_total_count": 0, }, ) history["latest_failed_at"] = row.get("triggered_at") history["latest_failure_summary"] = str(row.get("summary") or "").strip() history["latest_failure_detail"] = detail or {} return result