自动化消息转发系统开发方案

一、项目概述

本方案实现一个自动化消息转发系统，可将聊天软件接收到的消息自动转发到微信群或钉钉群。系统支持多源消息采集、智能过滤、消息格式转换和双通道转发，满足企业办公场景中的消息同步需求。

import json
import time
import requests
import threading
import schedule
from queue import Queue
from datetime import datetime

# 核心组件初始化
MESSAGE_QUEUE = Queue(maxsize=1000)
CONFIG = {
            
    "wechat_webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key",
    "dingtalk_webhook": "https://oapi.dingtalk.com/robot/send?access_token=your_token",
    "message_sources": [
        {
            "type": "slack", "webhook": "https://slack/webhook"},
        {
            "type": "discord", "webhook": "https://discord/webhook"}
    ],
    "forward_rules": [
        {
            "source": "slack", "target": "wechat", "keywords": ["紧急", "通知"]},
        {
            "source": "discord", "target": "dingtalk", "exclude": ["垃圾邮件"]}
    ]
}

二、系统架构设计

+----------------+     +-----------------+     +---------------+     +-----------------+
| 消息源         | --> | 消息接收器      | --> | 消息处理管道  | --> | 转发适配器      |
| (Slack/Discord)|     | (Webhook/API)   |     | (过滤/转换)   |     | (微信/钉钉)     |
+----------------+     +-----------------+     +---------------+     +-----------------+

三、核心功能实现

1. 消息接收模块

class MessageReceiver:
    def __init__(self, config):
        self.sources = config["message_sources"]
        self.running = True
        
    def start_listeners(self):
        for source in self.sources:
            thread = threading.Thread(target=self._listen, args=(source,))
            thread.daemon = True
            thread.start()
    
    def _listen(self, source):
        # 模拟不同消息源的监听
        while self.running:
            # 实际项目中替换为真实API调用
            mock_msg = {
            
                "source": source["type"],
                "content": f"测试消息 from {
              source['type']} {
              datetime.now()}",
                "sender": "user123",
                "timestamp": int(time.time())
            }
            MESSAGE_QUEUE.put(mock_msg)
            time.sleep(5)  # 模拟消息间隔
            
    def stop(self):
        self.running = False

2. 消息处理管道

class MessageProcessor:
    def __init__(self, rules):
        self.rules = rules
        self.filters = {
            
            "keyword": self._keyword_filter,
            "exclude": self._exclude_filter
        }
    
    def process(self, message):
        """消息处理流水线"""
        # 应用转发规则
        for rule in self.rules:
            if rule["source"] != message["source"]:
                continue
                
            # 执行过滤条件
            if not self._apply_filters(message, rule):
                continue
                
            # 消息格式转换
            formatted = self._format_message(message, rule["target"])
            
            # 进入转发队列
            ForwardingEngine.dispatch(formatted, rule["target"])
        
    def _apply_filters(self, message, rule):
        """应用过滤规则"""
        for filter_type, values in rule.items():
            if filter_type in self.filters and not self.filters[filter_type](message, values):
                return False
        return True
    
    def _keyword_filter(self, message, keywords):
        return any(kw in message["content"] for kw in keywords)
    
    def _exclude_filter(self, message, excludes):
        return not any(ex in message["content"] for ex in excludes)
    
    def _format_message(self, message, target):
        """转换为目标平台格式"""
        if target == "wechat":
            return {
            
                "msgtype": "text",
                "text": {
            
                    "content": f"[{
              message['source']}] {
              message['sender']}:
{
              message['content']}"
                }
            }
        elif target == "dingtalk":
            return {
            
                "msgtype": "markdown",
                "markdown": {
            
                    "title": "新消息通知",
                    "text": f"**来源**: {
              message['source']}
**发送人**: {
              message['sender']}
**内容**: {
              message['content']}"
                }
            }

3. 转发引擎

class ForwardingEngine:
    @staticmethod
    def dispatch(message, target):
        if target == "wechat":
            WechatAdapter.send(message)
        elif target == "dingtalk":
            DingtalkAdapter.send(message)
            
    @staticmethod
    def retry_failed_messages():
        """定时重试失败消息"""
        # 实现存储和重试逻辑
        pass

class WechatAdapter:
    @staticmethod
    def send(message):
        try:
            resp = requests.post(
                CONFIG["wechat_webhook"],
                json=message,
                timeout=5
            )
            if resp.status_code != 200 or resp.json().get("errcode") != 0:
                raise Exception(f"微信发送失败: {
              resp.text}")
        except Exception as e:
            print(f"微信消息发送异常: {
              str(e)}")
            # 消息存储到重试队列

class DingtalkAdapter:
    @staticmethod
    def send(message):
        try:
            resp = requests.post(
                CONFIG["dingtalk_webhook"],
                json=message,
                headers={
            "Content-Type": "application/json"},
                timeout=5
            )
            if resp.status_code != 200 or resp.json().get("errcode") != 0:
                raise Exception(f"钉钉发送失败: {
              resp.text}")
        except Exception as e:
            print(f"钉钉消息发送异常: {
              str(e)}")
            # 消息存储到重试队列

四、系统控制中心

class MessageForwardingSystem:
    def __init__(self, config):
        self.config = config
        self.receiver = MessageReceiver(config)
        self.processor = MessageProcessor(config["forward_rules"])
        
    def start(self):
        # 启动消息接收器
        self.receiver.start_listeners()
        
        # 启动处理线程
        processing_thread = threading.Thread(target=self._processing_loop)
        processing_thread.daemon = True
        processing_thread.start()
        
        # 启动定时任务
        schedule.every(5).minutes.do(ForwardingEngine.retry_failed_messages)
        
        print("系统已启动，开始监听消息...")
        while True:
            schedule.run_pending()
            time.sleep(1)
            
    def _processing_loop(self):
        while True:
            if not MESSAGE_QUEUE.empty():
                message = MESSAGE_QUEUE.get()
                self.processor.process(message)
            time.sleep(0.1)
            
    def graceful_shutdown(self):
        self.receiver.stop()
        print("系统正在关闭，处理剩余消息...")
        while not MESSAGE_QUEUE.empty():
            message = MESSAGE_QUEUE.get()
            self.processor.process(message)
        print("系统已安全关闭")

五、高级功能实现

1. 消息存储与重试机制

class MessageStorage:
    @staticmethod
    def save_failed_message(message, target, reason):
        """存储失败消息到数据库"""
        # 实际实现应使用数据库
        with open("failed_messages.json", "a") as f:
            record = {
            
                "message": message,
                "target": target,
                "reason": reason,
                "timestamp": datetime.now().isoformat(),
                "retry_count": 0
            }
            f.write(json.dumps(record) + "
")
    
    @staticmethod
    def get_retry_messages():
        """获取待重试消息"""
        # 简化的文件读取实现
        try:
            with open("failed_messages.json", "r") as f:
                return [json.loads(line) for line in f.readlines()]
        except FileNotFoundError:
            return []

2. 流量控制模块

class RateLimiter:
    def __init__(self, max_rate, time_window):
        self.max_rate = max_rate
        self.time_window = time_window
        self.timestamps = []
        
    def allow_request(self):
        now = time.time()
        # 清除过期记录
        self.timestamps = [ts for ts in self.timestamps if ts > now - self.time_window]
        
        if len(self.timestamps) < self.max_rate:
            self.timestamps.append(now)
            return True
        return False

六、安全增强措施

1. 消息验证签名

class SecurityValidator:
    @staticmethod
    def verify_signature(payload, signature, secret):
        # 钉钉签名验证示例
        timestamp = str(round(time.time() * 1000))
        sign_str = timestamp + "
" + secret
        sign = base64.b64encode(
            hmac.new(secret.encode(), sign_str.encode(), digestmod=hashlib.sha256).digest()
        )
        return hmac.compare_digest(sign.decode(), signature)

2. 敏感信息过滤

class ContentFilter:
    def __init__(self, sensitive_words):
        self.sensitive_words = sensitive_words or []
        
    def filter_content(self, content):
        for word in self.sensitive_words:
            content = content.replace(word, "***")
        return content

七、部署方案

1. 容器化部署

# Dockerfile 示例
FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
CMD ["python", "main.py"]

# 启动命令
docker build -t message-forwarder .
docker run -d --name forwarder 
  -v ./config:/app/config 
  message-forwarder

2. 配置管理

// config.json
{
            
  "wechat_webhook": "https://qyapi.weixin.qq.com/...",
  "dingtalk_webhook": "https://oapi.dingtalk.com/...",
  "sources": [
    {
            "type": "slack", "url": "https://hooks.slack.com/..."}
  ],
  "rate_limit": {
            
    "wechat": {
            "max_rate": 20, "time_window": 60},
    "dingtalk": {
            "max_rate": 30, "time_window": 60}
  }
}

八、监控与日志

class MonitoringSystem:
    def __init__(self):
        self.metrics = {
            
            "messages_received": 0,
            "messages_forwarded": 0,
            "errors": 0
        }
    
    def increment(self, metric):
        if metric in self.metrics:
            self.metrics[metric] += 1
            
    def report_metrics(self):
        while True:
            print(f"[监控报告] {
              datetime.now()}")
            print(f"收到消息: {
              self.metrics['messages_received']}")
            print(f"转发消息: {
              self.metrics['messages_forwarded']}")
            print(f"错误次数: {
              self.metrics['errors']}")
            time.sleep(300)  # 每5分钟报告一次

九、完整系统集成

def main():
    # 加载配置
    with open("config.json") as f:
        config = json.load(f)
    
    # 初始化系统
    system = MessageForwardingSystem(config)
    
    # 设置信号处理
    import signal
    signal.signal(signal.SIGINT, lambda s, f: system.graceful_shutdown())
    signal.signal(signal.SIGTERM, lambda s, f: system.graceful_shutdown())
    
    # 启动系统
    system.start()

if __name__ == "__main__":
    main()