重复任务 cron 表达式支持六位(秒)

任务发送成功后弹出消息

Author: TommyMerlin

app.py

@@ -1,11 +1,12 @@
-from flask import Flask, request, jsonify, send_from_directory
+from flask import Flask, request, jsonify, send_from_directory, Response
 from flask_cors import CORS
 from datetime import datetime
 from models import init_db, get_db, NotifyTask, NotifyChannel, NotifyStatus, User, UserChannel
-from scheduler import scheduler
+from scheduler import scheduler, get_cron_trigger, event_manager
 from auth import login_required, admin_required, user_login, user_register, update_user_profile
 import json
 import os
+import jwt
 
 app = Flask(__name__, static_folder='static')
 CORS(app)  # 启用跨域支持
@@ -153,8 +154,7 @@ def create_task():
                 return jsonify({'error': '重复任务必须提供 cron_expression'}), 400
             # 由 cron 计算下一次运行时间（用于列表展示与排序）
             try:
-                from apscheduler.triggers.cron import CronTrigger
-                trigger = CronTrigger.from_crontab(cron_expression)
+                trigger = get_cron_trigger(cron_expression)
                 next_run = trigger.get_next_fire_time(None, datetime.now())
                 if not next_run:
                     return jsonify({'error': '无法根据 cron_expression 计算下一次执行时间'}), 400
@@ -356,8 +356,7 @@ def update_task(task_id):
                         # 恢复时重新计算下一次执行时间
                         if task.is_recurring and task.cron_expression:
                             try:
-                                from apscheduler.triggers.cron import CronTrigger
-                                trigger = CronTrigger.from_crontab(task.cron_expression)
+                                trigger = get_cron_trigger(task.cron_expression)
                                 next_run = trigger.get_next_fire_time(None, datetime.now())
                                 if next_run:
                                     task.scheduled_time = next_run
@@ -391,8 +390,7 @@ def update_task(task_id):
             # 关键：如果是重复任务，根据 cron 表达式重新计算下一次执行时间
             if task.is_recurring and task.cron_expression:
                 try:
-                    from apscheduler.triggers.cron import CronTrigger
-                    trigger = CronTrigger.from_crontab(task.cron_expression)
+                    trigger = get_cron_trigger(task.cron_expression)
                     # 以当前时间为基准，计算下一次执行时间
                     next_run = trigger.get_next_fire_time(None, datetime.now())
                     if next_run:
@@ -698,6 +696,34 @@ def health_check():
     })
 
 
+@app.route('/api/events')
+def sse_events():
+    """服务器发送事件 (SSE) 端点"""
+    token = request.args.get('token')
+    if not token:
+        return jsonify({'error': 'Missing token'}), 401
+    
+    try:
+        payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
+        # 尝试获取 user_id，兼容常见的 payload key
+        user_id = payload.get('user_id') or payload.get('id') or payload.get('sub')
+        if not user_id:
+             return jsonify({'error': 'Invalid token payload'}), 401
+    except Exception:
+        return jsonify({'error': 'Invalid token'}), 401
+
+    def stream():
+        messages = event_manager.listen(user_id)
+        try:
+            while True:
+                msg = messages.get()
+                yield f"data: {json.dumps(msg, ensure_ascii=False)}\n\n"
+        except GeneratorExit:
+            pass
+
+    return Response(stream(), mimetype='text/event-stream')
+
+
 if __name__ == '__main__':
     try:
         app.run(host='0.0.0.0', port=8080, debug=True)

scheduler.py

@@ -5,11 +5,49 @@
 from models import NotifyTask, NotifyStatus, get_db
 from notifier import NotificationSender, parse_config
 import logging
+import queue
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
+class EventManager:
+    def __init__(self):
+        # listeners: list of (queue, user_id)
+        self.listeners = []
+
+    def listen(self, user_id):
+        q = queue.Queue(maxsize=10)
+        self.listeners.append((q, user_id))
+        return q
+
+    def announce(self, user_id, msg):
+        for i in range(len(self.listeners) - 1, -1, -1):
+            q, uid = self.listeners[i]
+            if uid == user_id:
+                try:
+                    q.put_nowait(msg)
+                except queue.Full:
+                    del self.listeners[i]
+
+event_manager = EventManager()
+
+
+def get_cron_trigger(expression):
+    """根据 cron 表达式获取触发器，支持 5 位 (分时日月周) 和 6 位 (秒分时日月周)"""
+    values = expression.strip().split()
+    if len(values) == 6:
+        return CronTrigger(
+            second=values[0],
+            minute=values[1],
+            hour=values[2],
+            day=values[3],
+            month=values[4],
+            day_of_week=values[5]
+        )
+    return CronTrigger.from_crontab(expression)
+
+
 class NotifyScheduler:
     """通知调度器"""
 
@@ -27,23 +65,36 @@ def add_task(self, task: NotifyTask):
         """
         if task.is_recurring and task.cron_expression:
             # 重复任务，使用 cron 表达式
-            trigger = CronTrigger.from_crontab(task.cron_expression)
-            job_id = f"recurring_task_{task.id}"
+            try:
+                trigger = get_cron_trigger(task.cron_expression)
+                job_id = f"recurring_task_{task.id}"
+                
+                self.scheduler.add_job(
+                    func=self._execute_task,
+                    trigger=trigger,
+                    args=[task.id],
+                    id=job_id,
+                    replace_existing=True,
+                    misfire_grace_time=60  # 错过时间窗口60秒内仍执行
+                )
+                logger.info(f"任务 {task.id} 已添加到调度器，计划执行时间: {task.scheduled_time}")
+            except Exception as e:
+                logger.error(f"添加任务 {task.id} 失败，Cron 表达式无效: {e}")
         else:
             # 一次性任务，使用指定时间
             trigger = DateTrigger(run_date=task.scheduled_time)
             job_id = f"task_{task.id}"
 
-        self.scheduler.add_job(
-            func=self._execute_task,
-            trigger=trigger,
-            args=[task.id],
-            id=job_id,
-            replace_existing=True,
-            misfire_grace_time=60  # 错过时间窗口60秒内仍执行
-        )
-        
-        logger.info(f"任务 {task.id} 已添加到调度器，计划执行时间: {task.scheduled_time}")
+            self.scheduler.add_job(
+                func=self._execute_task,
+                trigger=trigger,
+                args=[task.id],
+                id=job_id,
+                replace_existing=True,
+                misfire_grace_time=60  # 错过时间窗口60秒内仍执行
+            )
+            
+            logger.info(f"任务 {task.id} 已添加到调度器，计划执行时间: {task.scheduled_time}")
 
     def remove_task(self, task_id: int, is_recurring: bool = False):
         """
@@ -108,8 +159,7 @@ def _execute_task(self, task_id: int):
                     # 关键：重复任务执行成功后，滚动更新下一次执行时间（用于列表展示）
                     if task.is_recurring and task.cron_expression:
                         try:
-                            from apscheduler.triggers.cron import CronTrigger
-                            trigger = CronTrigger.from_crontab(task.cron_expression)
+                            trigger = get_cron_trigger(task.cron_expression)
                             # 以"本次实际执行时间"为基准，计算下一次
                             base_time = datetime.now()
                             next_run = trigger.get_next_fire_time(None, base_time)
@@ -120,12 +170,30 @@ def _execute_task(self, task_id: int):
                             logger.warning(f"任务 {task_id} 更新下一次执行时间失败: {str(e)}")
 
                     logger.info(f"任务 {task_id} 执行成功")
+                    
+                    # 通知前端
+                    event_manager.announce(task.user_id, {
+                        'type': 'task_executed',
+                        'task_id': task.id,
+                        'title': task.title,
+                        'status': 'sent',
+                        'message': '发送成功'
+                    })
 
                 except Exception as e:
                     # 更新任务状态为失败
                     task.status = NotifyStatus.FAILED
                     task.error_msg = str(e)
                     logger.error(f"任务 {task_id} 执行失败: {str(e)}")
+                    
+                    # 通知前端
+                    event_manager.announce(task.user_id, {
+                        'type': 'task_executed',
+                        'task_id': task.id,
+                        'title': task.title,
+                        'status': 'failed',
+                        'message': str(e)
+                    })
 
                 db.commit()
 
@@ -155,7 +223,7 @@ def load_pending_tasks(self):
                     # 如果是重复任务且计划时间已过，重新计算下一次执行时间
                     if task.is_recurring and task.cron_expression and task.scheduled_time < datetime.now():
                         try:
-                            trigger = CronTrigger.from_crontab(task.cron_expression)
+                            trigger = get_cron_trigger(task.cron_expression)
                             next_run = trigger.get_next_fire_time(None, datetime.now())
                             if next_run:
                                 task.scheduled_time = next_run

static/index.html

@@ -141,7 +141,7 @@ <h2>创建通知任务</h2>
 
                         <div class="form-group">
                             <label for="scheduledTime">计划发送时间 *</label>
-                            <input type="datetime-local" id="scheduledTime" name="scheduledTime" required>
+                            <input type="datetime-local" id="scheduledTime" name="scheduledTime" required step="1">
                         </div>
 
                         <div class="form-group">
@@ -153,9 +153,9 @@ <h2>创建通知任务</h2>
 
                         <div class="form-group" id="cronGroup" style="display: none;">
                             <label for="cronExpression">Cron 表达式</label>
-                            <input type="text" id="cronExpression" name="cronExpression" placeholder="如: 0 9 * * * (每天9点)">
+                            <input type="text" id="cronExpression" name="cronExpression" placeholder="如: 0 9 * * * (每天9点) 或 0/30 * * * * * (每30秒)">
                             <small style="color: var(--text-muted); display: block; margin-top: 5px;">
-                                示例: 0 9 * * * (每天9点) | 0 */2 * * * (每2小时) | 0 9 * * 1 (每周一9点)
+                                示例: 0 9 * * * (每天9点) | 0 */2 * * * (每2小时) | */30 * * * * * (每30秒)
                             </small>
                         </div>
 
@@ -319,7 +319,7 @@ <h2>编辑任务</h2>
 
                 <div class="form-group">
                     <label for="editScheduledTime">计划发送时间 *</label>
-                    <input type="datetime-local" id="editScheduledTime" name="scheduledTime" required>
+                    <input type="datetime-local" id="editScheduledTime" name="scheduledTime" required step="1">
                 </div>
 
                 <div class="form-group">

static/js/app.js

@@ -2,6 +2,7 @@ const API_BASE = '/api';
 let channels = [];
 let currentUser = null;
 let userChannels = [];
+let eventSource = null;
 
 // 初始化
 document.addEventListener('DOMContentLoaded', function() {
@@ -54,6 +55,7 @@ function showMainApp() {
     loadTasks();
     initAppEvents();
     setDefaultTime();
+    initSSE();
 
     // 尝试加载日历（如果存在日历脚本且在主界面显示后）
     if (typeof window.loadCalendar === 'function') {
@@ -154,9 +156,53 @@ async function handleRegister(e) {
 function logout() {
     localStorage.removeItem('token');
     currentUser = null;
+    if (eventSource) {
+        eventSource.close();
+        eventSource = null;
+    }
     checkAuthStatus();
 }
 
+// 初始化 SSE
+function initSSE() {
+    if (eventSource) {
+        eventSource.close();
+    }
+    
+    const token = localStorage.getItem('token');
+    if (!token) return;
+
+    // 使用 query param 传递 token
+    eventSource = new EventSource(`${API_BASE}/events?token=${token}`);
+    
+    eventSource.onmessage = function(event) {
+        try {
+            const data = JSON.parse(event.data);
+            if (data.type === 'task_executed') {
+                const type = data.status === 'sent' ? 'success' : 'error';
+                const msgPrefix = data.status === 'sent' ? '✅' : '❌';
+                showNotification(`${msgPrefix} 任务 "${data.title}" 执行完成: ${data.message}`, type);
+                
+                // 刷新列表
+                loadTasks();
+                
+                // 刷新日历
+                if (typeof window.loadCalendar === 'function') {
+                     delete window.__TASKS_CACHE;
+                     window.loadCalendar();
+                }
+            }
+        } catch (e) {
+            console.error('SSE parse error', e);
+        }
+    };
+    
+    eventSource.onerror = function(err) {
+        // 连接错误时，EventSource 会自动重连，这里仅记录
+        console.log('SSE connection error/closed');
+    };
+}
+
 // 初始化应用事件
 function initAppEvents() {
     document.getElementById('taskForm').addEventListener('submit', submitTaskForm);
@@ -924,7 +970,8 @@ async function submitTaskForm(e) {
         content: formData.get('content'),
         channel: channel,
         // 重复任务不提交 scheduled_time，由后端根据 cron_expression 计算
-        scheduled_time: isRecurring ? undefined : (scheduledTimeValue ? `${scheduledTimeValue}:00` : null),
+        // 如果有秒(长度19)则直接使用，否则(长度16)补:00
+        scheduled_time: isRecurring ? undefined : (scheduledTimeValue ? (scheduledTimeValue.length === 16 ? `${scheduledTimeValue}:00` : scheduledTimeValue) : null),
         channel_config: channelConfig,
         is_recurring: isRecurring,
         cron_expression: isRecurring ? cronForBackend : null
@@ -1139,13 +1186,15 @@ async function toggleTaskPause(taskId, action) {
 function setDefaultTime() {
     const now = new Date();
     now.setHours(now.getHours() + 1);
+    now.setSeconds(0);
+    now.setMilliseconds(0);
     document.getElementById('scheduledTime').value = toLocalInputValue(now);
 }
 
 // 将 Date 转为 datetime-local 可用的本地时间字符串
 function toLocalInputValue(date) {
     const local = new Date(date.getTime() - date.getTimezoneOffset() * 60000);
-    return local.toISOString().slice(0, 16);
+    return local.toISOString().slice(0, 19);
 }
 
 // 显示通知
@@ -1169,7 +1218,8 @@ function formatDateTime(dateString) {
         month: '2-digit',
         day: '2-digit',
         hour: '2-digit',
-        minute: '2-digit'
+        minute: '2-digit',
+        second: '2-digit'
     });
 }
 
@@ -1516,7 +1566,7 @@ function showEditTaskModal(task) {
     const scheduledTime = new Date(task.scheduled_time);
     // 转换为本地时间并格式化
     const localTime = new Date(scheduledTime.getTime() - scheduledTime.getTimezoneOffset() * 60000);
-    document.getElementById('editScheduledTime').value = localTime.toISOString().slice(0, 16);
+    document.getElementById('editScheduledTime').value = localTime.toISOString().slice(0, 19);
 
     // 设置重复任务信息
     const isRecurringCheckbox = document.getElementById('editIsRecurring');
@@ -1675,7 +1725,7 @@ async function handleEditTaskSubmit(e) {
         title: formData.get('title'),
         content: formData.get('content'),
         channel: channelType,
-        scheduled_time: scheduledTimeValue ? `${scheduledTimeValue}:00` : null,
+        scheduled_time: scheduledTimeValue ? (scheduledTimeValue.length === 16 ? `${scheduledTimeValue}:00` : scheduledTimeValue) : null,
         channel_config: channelConfig
         // 注意：不包含 is_recurring 和 cron_expression，因为后端不允许修改这些字段
     };

version.yml

@@ -1 +1 @@
-version: 0.4.0
+version: 0.4.1