From a8c92fe7ae5534d35e11ef70da1dc63e79d3a374 Mon Sep 17 00:00:00 2001 From: logan Date: Mon, 25 May 2026 00:08:48 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20APScheduler=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=20+=20SQL=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E6=8C=89=E9=92=AE=20+=20=E5=AE=9A=E6=97=B6=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E9=86=92=E7=9B=AE=E7=BA=A2=E8=89=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. ticket_scheduler.py: 纯asyncio→APScheduler+SQLite持久化 - 服务重启后定时任务不丢失 - 启动时自动恢复scheduled状态工单 - shutdown时优雅关闭scheduler 2. TicketCreatePage.tsx: - 提交按钮旁新增SQL检测按钮(Shield图标) - 验证通过前提交按钮置灰+提示'请先通过SQL检测' - SQL内容变化自动重置验证状态 - 定时时间改为醒目红色字体(text-red-500 font-semibold) 3. main.py: lifespan中启动APScheduler+恢复定时工单 --- backend/app/main.py | 12 ++ backend/app/services/ticket_scheduler.py | 189 +++++++++++++++++------ frontend/src/pages/TicketCreatePage.tsx | 80 ++++++---- 3 files changed, 208 insertions(+), 73 deletions(-) diff --git a/backend/app/main.py b/backend/app/main.py index 645596c..46cb0a4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -77,8 +77,20 @@ async def lifespan(app: FastAPI): await init_db() await seed_data() await seed_workflows() + + # 启动 APScheduler + 恢复定时工单 + from app.services.ticket_scheduler import start_scheduler, recover_scheduled_tickets, shutdown_scheduler + start_scheduler() + recovered = await recover_scheduled_tickets() + if recovered > 0: + import logging + logging.getLogger(__name__).info("恢复了 %d 个定时工单", recovered) + yield + # Shutdown + shutdown_scheduler() + app = FastAPI( title="SQL-Platform API", diff --git a/backend/app/services/ticket_scheduler.py b/backend/app/services/ticket_scheduler.py index 5be07a7..7413b42 100644 --- a/backend/app/services/ticket_scheduler.py +++ b/backend/app/services/ticket_scheduler.py @@ -1,49 +1,97 @@ -"""工单定时执行调度器 — 基于 asyncio 的轻量级定时任务。""" +"""工单定时执行调度器 — 基于 APScheduler + SQLite 持久化。 + +优势（对比纯 asyncio）: +1. 服务重启后定时任务不丢失（SQLite job store 持久化） +2. 启动时自动恢复 scheduled 状态的工单 +3. 支持任务重试、misfire_grace_time 等生产级特性 +""" import asyncio import logging +import os from datetime import datetime, timezone -from typing import Dict +from typing import Optional + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore +from apscheduler.triggers.date import DateTrigger logger = logging.getLogger(__name__) -# 内存中保存活跃的定时任务句柄 -_active_tasks: Dict[int, asyncio.Task] = {} +# 全局调度器实例（延迟初始化） +_scheduler: Optional[AsyncIOScheduler] = None + +DB_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), '..', 'apscheduler.db') + + +def _get_scheduler() -> AsyncIOScheduler: + """获取或创建 APScheduler 实例（单例）。""" + global _scheduler + if _scheduler is not None and _scheduler.running: + return _scheduler + + jobstore_url = 'sqlite:///' + os.path.abspath(DB_DIR) + _scheduler = AsyncIOScheduler( + jobstores={'default': SQLAlchemyJobStore(url=jobstore_url)}, + job_defaults={ + 'coalesce': True, # 同一 job 只执行一次 + 'max_instances': 1, # 不并发 + 'misfire_grace_time': 300, # 5 分钟内迟到仍执行 + }, + ) + return _scheduler + + +def start_scheduler() -> None: + """启动调度器（应用启动时调用）。""" + scheduler = _get_scheduler() + if not scheduler.running: + scheduler.start() + logger.info("APScheduler 已启动 (SQLite 持久化)") + + +def shutdown_scheduler() -> None: + """关闭调度器。""" + global _scheduler + if _scheduler and _scheduler.running: + _scheduler.shutdown(wait=False) + logger.info("APScheduler 已关闭") async def schedule_ticket_execution(ticket_id: int, scheduled_at: datetime) -> None: """注册一个定时执行任务。 - 使用 asyncio.create_task 创建后台等待任务，到达 scheduled_at 后自动执行工单。 - Args: ticket_id: 工单 ID scheduled_at: 定时执行时间 (UTC) """ - now = datetime.now(timezone.utc) - delay = (scheduled_at - now).total_seconds() + scheduler = _get_scheduler() + job_id = f"ticket_{ticket_id}" - if delay <= 0: + now = datetime.now(timezone.utc) + run_date = scheduled_at if scheduled_at.tzinfo else scheduled_at.replace(tzinfo=timezone.utc) + + if run_date <= now: # 已到时间，立即执行 logger.info("工单 %d 定时时间已到，立即执行", ticket_id) - task = asyncio.create_task(_execute_scheduled_ticket(ticket_id)) - else: - logger.info("工单 %d 定时执行: %s (%.1f 秒后)", ticket_id, scheduled_at, delay) - task = asyncio.create_task(_wait_and_execute(ticket_id, delay)) - - _active_tasks[ticket_id] = task - - -async def _wait_and_execute(ticket_id: int, delay: float) -> None: - """等待 delay 秒后执行工单。""" - try: - await asyncio.sleep(delay) await _execute_scheduled_ticket(ticket_id) - except asyncio.CancelledError: - logger.info("工单 %d 定时执行已取消", ticket_id) - except Exception: - logger.exception("工单 %d 定时执行异常", ticket_id) - finally: - _active_tasks.pop(ticket_id, None) + return + + delay = (run_date - now).total_seconds() + logger.info("工单 %d 定时执行: %s (%.1f 秒后)", ticket_id, scheduled_at, delay) + + scheduler.add_job( + _run_async_execution, + trigger=DateTrigger(run_date=run_date), + args=[ticket_id], + id=job_id, + replace_existing=True, + name=f"定时执行工单 #{ticket_id}", + ) + + +async def _run_async_execution(ticket_id: int) -> None: + """APScheduler 调用的入口（同步 wrapper → 异步执行）。""" + await _execute_scheduled_ticket(ticket_id) async def _execute_scheduled_ticket(ticket_id: int) -> None: @@ -52,7 +100,6 @@ async def _execute_scheduled_ticket(ticket_id: int) -> None: from app.models.ticket import Ticket as TicketModel from app.models.instance import Instance from app.services.ticket_executor import TicketExecutor - from app.services.audit_log import AuditLogService from app.core.security import decrypt_password from sqlalchemy import select import json @@ -61,24 +108,20 @@ async def _execute_scheduled_ticket(ticket_id: int) -> None: async with async_session() as db: try: - # 查询工单 result = await db.execute(select(TicketModel).where(TicketModel.id == ticket_id)) ticket = result.scalar_one_or_none() if not ticket: logger.error("定时工单 %d 不存在", ticket_id) return - # 状态检查 if ticket.status != "scheduled": logger.warning("工单 %d 状态已变更 (%s)，跳过执行", ticket_id, ticket.status) return - # 更新状态为 executing ticket.status = "executing" await db.commit() await db.refresh(ticket) - # 查询实例 inst_result = await db.execute(select(Instance).where(Instance.id == ticket.instance_id)) instance = inst_result.scalar_one_or_none() if not instance: @@ -88,14 +131,10 @@ async def _execute_scheduled_ticket(ticket_id: int) -> None: await db.commit() return - # 解密密码 instance.password = decrypt_password(instance.password) - - # 执行 executor = TicketExecutor() report = await executor.execute(ticket, instance) - # 更新结果 ticket.status = "success" if report.status == "success" else "failed" ticket.execute_result = report.to_json() ticket.execute_time = datetime.now(timezone.utc) @@ -115,8 +154,6 @@ async def _execute_scheduled_ticket(ticket_id: int) -> None: await db.commit() except Exception: logger.exception("更新定时工单 %d 失败状态也失败", ticket_id) - finally: - _active_tasks.pop(ticket_id, None) def cancel_ticket_execution(ticket_id: int) -> bool: @@ -128,20 +165,78 @@ def cancel_ticket_execution(ticket_id: int) -> bool: Returns: 是否成功取消 """ - task = _active_tasks.get(ticket_id) - if task and not task.done(): - task.cancel() - _active_tasks.pop(ticket_id, None) + scheduler = _get_scheduler() + job_id = f"ticket_{ticket_id}" + try: + scheduler.remove_job(job_id) logger.info("已取消工单 %d 的定时执行", ticket_id) return True - return False + except Exception: + return False + + +async def recover_scheduled_tickets() -> int: + """启动时恢复所有 scheduled 状态的工单。 + + 扫描 DB 中 status=scheduled 的工单，重新注册到 APScheduler。 + + Returns: + 恢复的任务数量 + """ + from app.core.database import async_session + from app.models.ticket import Ticket as TicketModel + from sqlalchemy import select + + count = 0 + async with async_session() as db: + result = await db.execute( + select(TicketModel).where(TicketModel.status == "scheduled") + ) + tickets = result.scalars().all() + + now = datetime.now(timezone.utc) + for ticket in tickets: + if not ticket.scheduled_at: + continue + + scheduled = ticket.scheduled_at + if scheduled.tzinfo is None: + scheduled = scheduled.replace(tzinfo=timezone.utc) + + if scheduled <= now: + # 已过期，立即执行 + logger.info("恢复过期定时工单 %d，立即执行", ticket.id) + asyncio.create_task(_execute_scheduled_ticket(ticket.id)) + else: + await schedule_ticket_execution(ticket.id, scheduled) + + count += 1 + + if count > 0: + logger.info("恢复了 %d 个定时执行工单", count) + return count def get_active_scheduled_count() -> int: - """返回当前活跃的定时任务数量。""" - return len(_active_tasks) + """返回当前 APScheduler 中的定时任务数量。""" + scheduler = _get_scheduler() + if not scheduler.running: + return 0 + jobs = scheduler.get_jobs() + return len([j for j in jobs if j.id.startswith("ticket_")]) def get_active_scheduled_ids() -> list[int]: - """返回当前活跃的定时工单 ID 列表。""" - return list(_active_tasks.keys()) + """返回当前定时工单 ID 列表。""" + scheduler = _get_scheduler() + if not scheduler.running: + return [] + jobs = scheduler.get_jobs() + ids = [] + for j in jobs: + if j.id.startswith("ticket_"): + try: + ids.append(int(j.id.split("_")[1])) + except (ValueError, IndexError): + pass + return ids diff --git a/frontend/src/pages/TicketCreatePage.tsx b/frontend/src/pages/TicketCreatePage.tsx index c78b96a..4640269 100644 --- a/frontend/src/pages/TicketCreatePage.tsx +++ b/frontend/src/pages/TicketCreatePage.tsx @@ -111,6 +111,8 @@ export default function TicketCreatePage() { // -- Inline audit result -- const [auditResult, setAuditResult] = useState(null); + const [validating, setValidating] = useState(false); + const [validated, setValidated] = useState(false); // -- Schema Browser State -- const [tables, setTables] = useState([]); @@ -249,37 +251,53 @@ export default function TicketCreatePage() { const handleSqlChange = useCallback((val: string) => { setSqlContent(val); if (auditResult) setAuditResult(null); - }, [auditResult]); + if (validated) setValidated(false); + }, [auditResult, validated]); /* ---------------------------------------------------------------- */ - /* Submit: audit + create in one click */ + /* Validate: SQL 检测（不提交） */ + /* ---------------------------------------------------------------- */ + + const handleValidate = useCallback(async () => { + if (!sqlContent.trim()) { setError('请填写 SQL 语句'); return; } + if (!instanceId) { setError('请选择目标实例'); return; } + if (!database) { setError('请选择数据库'); return; } + + setError(''); setAuditResult(null); setValidating(true); + try { + const result = await ticketApi.audit({ + sql_content: sqlContent, instance_id: Number(instanceId), database, + }) as unknown as AuditResult; + setAuditResult(result); + + const hasErrors = result.findings?.some((f) => f.level === 'error'); + if (hasErrors) { + setError('SQL 审核未通过，请修改后重新检测'); + setValidated(false); + } else { + setValidated(true); + } + } catch (e: unknown) { + const err = e as { response?: { data?: { detail?: string } } }; + setError(err?.response?.data?.detail || 'SQL 检测失败'); + setValidated(false); + } finally { setValidating(false); } + }, [sqlContent, instanceId, database]); + + /* ---------------------------------------------------------------- */ + /* Submit: 验证通过后直接提交工单 */ /* ---------------------------------------------------------------- */ const handleSubmit = useCallback(async () => { - // 1. 表单校验 if (!title.trim()) { setError('请填写工单标题'); return; } if (!instanceId) { setError('请选择目标实例'); return; } if (!database) { setError('请选择数据库'); return; } if (!sqlContent.trim()) { setError('请填写 SQL 语句'); return; } if (!reason.trim()) { setError('请填写变更原因'); return; } - setError(''); setAuditResult(null); setSubmitting(true); + setError(''); setSubmitting(true); try { - // 2. 先审核 - const result = await ticketApi.audit({ - sql_content: sqlContent, instance_id: Number(instanceId), database, - }) as unknown as AuditResult; - setAuditResult(result); - - // 3. 有 error 级别违规则阻止提交 - const hasErrors = result.findings?.some((f) => f.level === 'error'); - if (hasErrors) { - setError('SQL 审核未通过，请修改后重新提交'); - return; - } - - // 4. 提交工单 const inst = instances.find((i) => i.id === Number(instanceId)); await ticketApi.create({ title: title.trim(), @@ -298,7 +316,7 @@ export default function TicketCreatePage() { const msg = Array.isArray(raw) ? raw.map((v) => v.msg).join('; ') : (raw || '提交失败，请稍后重试'); setError(msg); } finally { setSubmitting(false); } - }, [title, sqlContent, instanceId, database, reason, instances, navigate]); + }, [title, sqlContent, instanceId, database, reason, instances, navigate, enableSchedule, scheduledAt]); /* ---------------------------------------------------------------- */ /* Derived values */ @@ -718,7 +736,7 @@ export default function TicketCreatePage() { min={new Date().toISOString().slice(0, 16)} /> )} {enableSchedule && scheduledAt && ( - + 将于 {new Date(scheduledAt).toLocaleString('zh-CN')} 执行 )} @@ -795,13 +813,23 @@ export default function TicketCreatePage() { )} - {/* === Submit Button === */} -
- +