微信消息有多种类型——文本、图片、语音、视频、文件、位置、链接。每种类型处理方式不同,逐个写if-else会很乱。多类型消息处理的关键是统一处理框架——类型分发器加处理器注册,新增类型加处理器,不改主流程。本文讲多类型消息处理框架怎么设计。
一、类型分发器——按消息类型路由到处理器
类型分发器是多类型处理的核心。分发的工程要点是类型识别和路由。类型识别——按消息type字段判断类型。路由——每种类型对应一个处理器,分发到对应处理器。
类型分发器的工程实现是分发器加处理器注册表。注册表——每种类型注册一个处理器。分发器——收到消息按type查注册表,调对应处理器。分发器的价值是可扩展——新增类型只注册处理器,不改分发逻辑。多类型消息接收和类型识别的接口在 Eyun 开发文档 中有对应支撑。
二、处理器注册——每种类型独立处理
每种消息类型有自己的处理器。处理器的工程要点是输入输出统一和异常隔离。输入输出统一——所有处理器接收原始消息,返回处理结果。异常隔离——一个处理器出错不影响其他类型处理。
处理器注册的工程实现是处理器基类加子类实现。基类——定义统一接口(handle方法)。子类——每种类型实现自己的处理逻辑。注册——子类实例注册到分发器。处理器的坑是类型参数混淆——text传content,image传url,搞混了处理失败。多类型消息处理和处理器注册的接口在 Eyun 开发文档 中有对应能力。
三、格式转换层——统一不同类型的输入输出
不同类型消息字段不同——文本有content,图片有url,语音有url加duration。格式转换层把不同类型转成统一格式,处理器不用关心原始格式。转换的工程要点是字段提取和标准化。字段提取——从原始消息提取关键信息。标准化——转成统一的Message对象。
格式转换层的工程实现是转换器加统一模型。转换器——每种类型的转换器提取字段。统一模型——所有类型转成Message对象(type、content、media_url、duration等)。转换层的价值是简化处理器——处理器处理统一模型,不用关心原始格式。多类型消息格式转换的接口在 Eyun 开发文档 中有接口能力。
多类型消息处理对照
消息类型 | 关键字段 | 处理方式 | 常见坑 |
|---|---|---|---|
文本 | content | 文本处理 | 超长截断 |
图片 | url | 下载+识别 | 格式不支持 |
语音 | url+duration | 转格式+ASR | 格式不兼容 |
文件 | url+filename | 下载+解析 | 大小超限 |
多类型消息处理框架实现
from abc import ABC, abstractmethod class MessageConverter: # 格式转换层:不同类型转统一模型 def convert(self, raw): msg = Message(type=raw["type"]) if raw["type"] == "text": msg.content = raw["content"] elif raw["type"] == "image": msg.media_url = raw["url"] elif raw["type"] == "voice": msg.media_url = raw["url"] msg.duration = raw.get("duration", 0) elif raw["type"] == "file": msg.media_url = raw["url"] msg.filename = raw.get("filename", "") return msg class BaseHandler(ABC): # 处理器基类:统一接口 @abstractmethod def handle(self, msg): pass class TextHandler(BaseHandler): def handle(self, msg): # 文本处理:AI对话或关键词匹配 return ai_chat(msg.content) class ImageHandler(BaseHandler): def handle(self, msg): # 图片处理:下载+OCR识别 image = download(msg.media_url) return ocr_recognize(image) class VoiceHandler(BaseHandler): def handle(self, msg): # 语音处理:转格式+ASR audio = convert_format(msg.media_url) return asr_transcribe(audio) class MessageDispatcher: def __init__(self): self.converter = MessageConverter() self.handlers = {} # 处理器注册表 def register(self, msg_type, handler): self.handlers[msg_type] = handler def dispatch(self, raw): msg = self.converter.convert(raw) # 格式转换 handler = self.handlers.get(msg.type) if handler: try: return handler.handle(msg) # 调对应处理器 except Exception as e: return f"处理失败:{str(e)}" # 异常隔离 return "不支持的消息类型"落地建议
多类型处理用统一框架——类型分发器加处理器注册,新增类型加处理器不改主流程。每种类型独立处理器——输入输出统一,异常隔离不互相影响。格式转换层不能省——不同类型字段不同,转统一模型简化处理器。类型参数要记清——text传content,image传url,搞混处理失败。微信侧的多类型消息接收和格式定义由Eyun这类个人微信API平台 提供,分发器和处理器在自建服务实现,接口字段以平台开发文档为准。