新增 消息定时推送功能

admin/dashboard/blueprints/message_push.py

@@ -1,3 +1,7 @@
+import asyncio
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
 from flask import Blueprint, render_template, jsonify, request, current_app, session
 from .auth import login_required
 from loguru import logger
@@ -8,6 +12,56 @@ from datetime import datetime
 # 创建消息推送管理蓝图
 message_push_bp = Blueprint('message_push', __name__, url_prefix='/message_push')
 
+# 创建线程池
+message_thread_pool = ThreadPoolExecutor(max_workers=10, thread_name_prefix="message_sender_")
+
+# 创建共享的事件循环
+shared_loop = None
+loop_lock = threading.Lock()
+
+
+def get_or_create_loop():
+    """获取或创建共享的事件循环"""
+    global shared_loop
+    with loop_lock:
+        if shared_loop is None:
+            shared_loop = asyncio.new_event_loop()
+
+            # 在新线程中运行事件循环
+            def run_loop():
+                asyncio.set_event_loop(shared_loop)
+                shared_loop.run_forever()
+
+            loop_thread = threading.Thread(target=run_loop, daemon=True)
+            loop_thread.start()
+        return shared_loop
+
+
+def send_message_in_thread(func, *args, **kwargs):
+    """使用共享事件循环发送消息"""
+
+    def run():
+        try:
+            loop = get_or_create_loop()
+
+            # 创建异步任务
+            async def send():
+                try:
+                    await func(*args, **kwargs)
+                except Exception as e:
+                    logger.error(f"发送消息失败: {e}")
+
+            # 在共享事件循环中运行任务
+            future = asyncio.run_coroutine_threadsafe(send(), loop)
+            # 等待任务完成，设置超时时间
+            future.result(timeout=10)
+        except Exception as e:
+            logger.error(f"消息发送任务执行失败: {e}")
+
+    # 使用线程池提交任务
+    message_thread_pool.submit(run)
+
+
 # 消息推送管理页面
 @message_push_bp.route('/')
 @login_required
@@ -55,9 +109,7 @@ def api_create_task():
             return jsonify({"success": False, "error": "无效的请求数据"}), 400
             
         # 获取用户名
-        logger.debug(f"Session before getting username: {dict(session)}")
         username = session.get('username')
-        logger.debug(f"Username from session: {username}")
         if not username:
             return jsonify({"success": False, "error": "未登录或会话已过期"}), 401
             
@@ -192,10 +244,38 @@ def api_preview_task(task_id):
         if not task:
             return jsonify({"success": False, "error": "任务不存在"}), 404
             
-        # 发送预览
+        # 获取机器人实例
-        message_push = current_app.dashboard_server.message_push_task.message_push
+        server = current_app.dashboard_server
-        if not message_push.send_preview(task, [session.get('user_id')]):  # 从session中获取用户ID
+        if not server or not server.client:
-            return jsonify({"success": False, "error": "发送预览失败"}), 500
+            return jsonify({"success": False, "error": "机器人未初始化"}), 500
+            
+        preview_user = session.get('user_id')
+        if not preview_user:
+            return jsonify({"success": False, "error": "未登录或会话已过期"}), 401
+            
+        # 发送文本消息
+        if task.get('content_text'):
+            send_message_in_thread(server.client.send_text_message, preview_user, task['content_text'])
+            
+        # 发送图片消息
+        if task.get('content_image'):
+            send_message_in_thread(server.client.send_image_message, preview_user, task['content_image'])
+            
+        # 发送链接消息
+        if task.get('content_link'):
+            send_message_in_thread(server.client.send_link_message, preview_user, task['content_link'])
+            
+        # 发送小程序消息
+        if task.get('content_miniprogram'):
+            miniprogram = task['content_miniprogram']
+            send_message_in_thread(
+                server.client.send_miniprogram_message,
+                preview_user,
+                miniprogram.get('title'),
+                miniprogram.get('appid'),
+                miniprogram.get('pagepath'),
+                miniprogram.get('thumb_url')
+            )
             
         return jsonify({
             "success": True,