支持多渠道消息推送

Author: TommyMerlin

README.md

@@ -447,7 +447,7 @@ A: 在 `notifier.py` 中添加新的发送方法，并在 `models.py` 中添加
 - [x] 移动端响应式适配与触摸优化
 - [x] 拖拽调整任务时间（日历视图）
 - [x] 日历订阅及同步功能
-- [ ] 通知群组/广播，允许将多个渠道打包成一个群组，实现一次任务多渠道分发
+- [x] 支持多渠道消息推送
 - [ ] 失败自动重试
 - [ ] 重复任务暂停功能
 - [ ] 数据导入/导出

app.py

@@ -123,17 +123,29 @@ def create_task():
     """
     创建通知任务
 
-    请求体示例:
+    支持单渠道和多渠道两种模式：
+    
+    单渠道模式 (向后兼容):
     {
         "title": "测试通知",
         "content": "这是一条测试通知",
         "channel": "wecom_webhook",
+        "channel_config": {"webhook_url": "..."},
         "scheduled_time": "2024-12-01T10:00:00",
-        "channel_config": {
-            "webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
+        "is_recurring": false
+    }
+    
+    多渠道模式:
+    {
+        "title": "测试通知",
+        "content": "这是一条测试通知",
+        "channels": ["wecom_webhook", "pushplus"],
+        "channels_config": {
+            "wecom_webhook": {"webhook_url": "..."},
+            "pushplus": {"token": "..."}
         },
-        "is_recurring": false,
-        "cron_expression": null
+        "scheduled_time": "2024-12-01T10:00:00",
+        "is_recurring": false
     }
     """
     try:
@@ -143,11 +155,20 @@ def create_task():
         is_recurring = bool(data.get('is_recurring', False))
         cron_expression = data.get('cron_expression')
 
+        # 检测是多渠道模式还是单渠道模式
+        is_multi_channel = 'channels' in data
+        
         # 验证必填字段
-        required_fields = ['title', 'content', 'channel', 'channel_config']
+        required_fields = ['title', 'content']
+        if is_multi_channel:
+            required_fields.extend(['channels', 'channels_config'])
+        else:
+            required_fields.extend(['channel', 'channel_config'])
+        
         # 非重复任务必须提供 scheduled_time
         if not is_recurring:
             required_fields.append('scheduled_time')
+        
         for field in required_fields:
             if field not in data:
                 return jsonify({'error': f'缺少必填字段: {field}'}), 400
@@ -172,25 +193,48 @@ def create_task():
             except ValueError:
                 return jsonify({'error': '时间格式错误，请使用 ISO 格式，如: 2024-12-01T10:00:00'}), 400
 
-        # 验证通知渠道
-        try:
-            channel = NotifyChannel(data['channel'])
-        except ValueError:
-            valid_channels = [c.value for c in NotifyChannel]
-            return jsonify({'error': f'无效的通知渠道，支持的渠道: {valid_channels}'}), 400
-
         # 创建任务
         with get_db() as db:
             task = NotifyTask(
                 user_id=request.current_user.id,
                 title=data['title'],
                 content=data['content'],
-                channel=channel,
                 scheduled_time=scheduled_time,
-                channel_config=json.dumps(data['channel_config'], ensure_ascii=False),
                 is_recurring=is_recurring,
                 cron_expression=cron_expression if is_recurring else None
             )
+            
+            if is_multi_channel:
+                # 多渠道模式
+                channels = data['channels']
+                channels_config = data['channels_config']
+                
+                # 验证所有渠道类型
+                if not isinstance(channels, list) or len(channels) == 0:
+                    return jsonify({'error': 'channels 必须是非空数组'}), 400
+                
+                valid_channels = [c.value for c in NotifyChannel]
+                for ch in channels:
+                    if ch not in valid_channels:
+                        return jsonify({'error': f'无效的通知渠道: {ch}，支持的渠道: {valid_channels}'}), 400
+                
+                # 验证每个渠道都有配置
+                for ch in channels:
+                    if ch not in channels_config:
+                        return jsonify({'error': f'渠道 {ch} 缺少配置信息'}), 400
+                
+                task.channels_json = json.dumps(channels, ensure_ascii=False)
+                task.channels_config_json = json.dumps(channels_config, ensure_ascii=False)
+            else:
+                # 单渠道模式（向后兼容）
+                try:
+                    channel = NotifyChannel(data['channel'])
+                except ValueError:
+                    valid_channels = [c.value for c in NotifyChannel]
+                    return jsonify({'error': f'无效的通知渠道，支持的渠道: {valid_channels}'}), 400
+                
+                task.channel = channel
+                task.channel_config = json.dumps(data['channel_config'], ensure_ascii=False)
 
             db.add(task)
             db.commit()
@@ -322,8 +366,9 @@ def update_task(task_id):
     """
     更新任务
 
-    可更新字段: title, content, scheduled_time, channel_config, status
+    可更新字段: title, content, scheduled_time, channel_config, channels_config, status
     支持重新启用已取消或已执行的任务，以及暂停/恢复重复任务
+    支持在单渠道和多渠道模式间切换
     """
     try:
         data = request.get_json()
@@ -386,12 +431,37 @@ def update_task(task_id):
                 except Exception as e:
                     return jsonify({'error': f'状态更新失败: {str(e)}'}), 400
 
-            # 更新字段
+            # 更新基本字段
             if 'title' in data:
                 task.title = data['title']
             if 'content' in data:
                 task.content = data['content']
-            if 'channel_config' in data:
+            
+            # 处理渠道配置更新（支持单渠道和多渠道模式）
+            if 'channels' in data and 'channels_config' in data:
+                # 更新为多渠道模式
+                channels = data['channels']
+                channels_config = data['channels_config']
+                
+                if not isinstance(channels, list) or len(channels) == 0:
+                    return jsonify({'error': 'channels 必须是非空数组'}), 400
+                
+                valid_channels = [c.value for c in NotifyChannel]
+                for ch in channels:
+                    if ch not in valid_channels:
+                        return jsonify({'error': f'无效的通知渠道: {ch}'}), 400
+                
+                for ch in channels:
+                    if ch not in channels_config:
+                        return jsonify({'error': f'渠道 {ch} 缺少配置信息'}), 400
+                
+                task.channels_json = json.dumps(channels, ensure_ascii=False)
+                task.channels_config_json = json.dumps(channels_config, ensure_ascii=False)
+                # 清空单渠道字段
+                task.channel = None
+                task.channel_config = None
+            elif 'channel_config' in data:
+                # 更新单渠道模式的配置
                 task.channel_config = json.dumps(data['channel_config'], ensure_ascii=False)
 
             # 处理时间更新
@@ -420,6 +490,8 @@ def update_task(task_id):
                 task.status = NotifyStatus.PENDING
                 task.sent_time = None
                 task.error_msg = None
+                # 清空多渠道发送结果
+                task.send_results = None
 
             db.commit()
 

models.py

@@ -167,11 +167,16 @@ class NotifyTask(Base):
     user_id = Column(Integer, ForeignKey('users.id'), nullable=False, comment="用户ID")
     title = Column(String(200), nullable=False, comment="通知标题")
     content = Column(Text, nullable=False, comment="通知内容")
-    channel = Column(Enum(NotifyChannel, values_callable=lambda obj: [e.value for e in NotifyChannel]), nullable=False, comment="通知渠道")
+    channel = Column(Enum(NotifyChannel, values_callable=lambda obj: [e.value for e in NotifyChannel]), nullable=True, comment="通知渠道（单渠道模式）")
     scheduled_time = Column(DateTime, nullable=False, comment="计划发送时间")
 
     # 渠道配置（JSON格式字符串）
-    channel_config = Column(Text, nullable=False, comment="渠道配置信息")
+    channel_config = Column(Text, nullable=True, comment="渠道配置信息（单渠道模式）")
+    
+    # 多渠道支持
+    channels_json = Column(Text, nullable=True, comment="通知渠道数组（JSON格式，多渠道模式）")
+    channels_config_json = Column(Text, nullable=True, comment="渠道配置映射（JSON格式，多渠道模式）")
+    send_results = Column(Text, nullable=True, comment="各渠道发送结果（JSON格式）")
 
     # 状态相关
     status = Column(Enum(NotifyStatus, values_callable=lambda obj: [e.value for e in NotifyStatus]), default=NotifyStatus.PENDING, comment="发送状态")
@@ -201,12 +206,28 @@ def to_dict(self):
                 channel_config = ast.literal_eval(self.channel_config) if self.channel_config else {}
             except:
                 channel_config = {}
+        
+        # 安全解析多渠道字段
+        try:
+            channels = json.loads(self.channels_json) if self.channels_json else None
+        except (json.JSONDecodeError, TypeError):
+            channels = None
+        
+        try:
+            channels_config = json.loads(self.channels_config_json) if self.channels_config_json else None
+        except (json.JSONDecodeError, TypeError):
+            channels_config = None
+        
+        try:
+            send_results = json.loads(self.send_results) if self.send_results else None
+        except (json.JSONDecodeError, TypeError):
+            send_results = None
 
-        return {
+        result = {
             'id': self.id,
             'title': self.title,
             'content': self.content,
-            'channel': self.channel.value,
+            'channel': self.channel.value if self.channel else None,
             'scheduled_time': self.scheduled_time.isoformat() if self.scheduled_time else None,
             'status': self.status.value,
             'sent_time': self.sent_time.isoformat() if self.sent_time else None,
@@ -217,6 +238,14 @@ def to_dict(self):
             'channel_config': channel_config,
             'external_uid': self.external_uid
         }
+        
+        # 添加多渠道字段（如果存在）
+        if channels:
+            result['channels'] = channels
+            result['channels_config'] = channels_config
+            result['send_results'] = send_results
+        
+        return result
 
 
 # 数据库配置
@@ -240,13 +269,82 @@ def init_db():
             except Exception:
                 print("Migrating: Adding calendar_token to users table...")
                 conn.execute(text("ALTER TABLE users ADD COLUMN calendar_token VARCHAR(64)"))
+                conn.commit()
 
             # 2. 检查 notify_tasks.external_uid
             try:
                 conn.execute(text("SELECT external_uid FROM notify_tasks LIMIT 1"))
             except Exception:
                 print("Migrating: Adding external_uid to notify_tasks table...")
                 conn.execute(text("ALTER TABLE notify_tasks ADD COLUMN external_uid VARCHAR(255)"))
+                conn.commit()
+            
+            # 3. 检查 notify_tasks.channels_json（多渠道支持）
+            try:
+                conn.execute(text("SELECT channels_json FROM notify_tasks LIMIT 1"))
+            except Exception:
+                print("Migrating: Adding multi-channel support fields to notify_tasks table...")
+                conn.execute(text("ALTER TABLE notify_tasks ADD COLUMN channels_json TEXT"))
+                conn.execute(text("ALTER TABLE notify_tasks ADD COLUMN channels_config_json TEXT"))
+                conn.execute(text("ALTER TABLE notify_tasks ADD COLUMN send_results TEXT"))
+                conn.commit()
+            
+            # 4. 移除 channel 和 channel_config 的 NOT NULL 约束（多渠道模式需要）
+            # SQLite 不支持直接修改列约束，需要重建表
+            try:
+                # 检查是否已经迁移（通过尝试插入 channel=NULL 的记录）
+                result = conn.execute(text("SELECT sql FROM sqlite_master WHERE type='table' AND name='notify_tasks'"))
+                table_schema = result.fetchone()[0]
+                
+                # 如果表结构中 channel 字段仍有 NOT NULL 约束
+                if 'channel' in table_schema and 'channel TEXT NOT NULL' in table_schema:
+                    print("Migrating: Removing NOT NULL constraint from channel and channel_config...")
+                    
+                    # 创建临时表
+                    conn.execute(text("""
+                        CREATE TABLE notify_tasks_new (
+                            id INTEGER PRIMARY KEY AUTOINCREMENT,
+                            user_id INTEGER NOT NULL,
+                            title VARCHAR(200) NOT NULL,
+                            content TEXT NOT NULL,
+                            channel TEXT,
+                            scheduled_time DATETIME NOT NULL,
+                            channel_config TEXT,
+                            channels_json TEXT,
+                            channels_config_json TEXT,
+                            send_results TEXT,
+                            status TEXT DEFAULT 'pending',
+                            sent_time DATETIME,
+                            error_msg TEXT,
+                            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                            is_recurring BOOLEAN DEFAULT 0,
+                            cron_expression VARCHAR(100),
+                            external_uid VARCHAR(255),
+                            FOREIGN KEY (user_id) REFERENCES users(id)
+                        )
+                    """))
+                    
+                    # 复制数据
+                    conn.execute(text("""
+                        INSERT INTO notify_tasks_new 
+                        SELECT id, user_id, title, content, channel, scheduled_time, 
+                               channel_config, channels_json, channels_config_json, send_results,
+                               status, sent_time, error_msg, created_at, updated_at, 
+                               is_recurring, cron_expression, external_uid
+                        FROM notify_tasks
+                    """))
+                    
+                    # 删除旧表
+                    conn.execute(text("DROP TABLE notify_tasks"))
+                    
+                    # 重命名新表
+                    conn.execute(text("ALTER TABLE notify_tasks_new RENAME TO notify_tasks"))
+                    
+                    conn.commit()
+                    print("Migration completed: channel and channel_config are now nullable")
+            except Exception as e:
+                print(f"Channel nullable migration info: {e}")
     except Exception as e:
         print(f"Migration warning: {e}")