收到的多媒体消息回调里一般只有一个临时下载地址不下载就过期。下载存本地这套流程要解决四个问题下载时机、格式转换、存储组织、可检索索引。这篇把从回调到落库的完整管道讲清楚。一、下载时机回调到达即触发别懒加载最常见的错误是用到才下载——图片消息存了 URL一周后用户想看历史图片URL 早过期了。正确做法是回调到达时立即触发下载任务进下载队列异步执行下载完成把本地路径回写到消息记录。下载失败要重试——网络抖动常见重试 3 次指数退避还失败标记下载失败趁 URL 有效期内人工处理或等消息重推再下。多媒体下载地址的有效期说明在 Eyun 开发文档 中可以查到。二、格式转换语音 silk 转 mp3图片统一格式微信语音是 silk 格式浏览器和多数播放器不支持要用 silk-v3-decoder 转成 mp3 或 wav 再存。图片可能是 webp、gif 等格式按存储策略决定是否统一转 jpg/png。视频一般保持 mp4 不转但要提取封面图首帧和时长信息入库列表页展示用。三、存储组织按日期和类型分目录文件存储建议按类型/年/月/日组织目录比如 voice/2026/10/27/xxx.mp3。文件名用 msg_id 加后缀保证唯一且能反查消息。大图可以压一份缩略图宽度 300px列表页用缩略图、详情页用原图省流量。存储介质选择量小直接用本地磁盘加定期备份量大用对象存储OSS/COS数据库里只存路径和元数据不存二进制。四、检索索引让文件能被找到存了不用等于白存。消息记录表里多媒体消息要存msg_id、类型、本地路径、文件大小、时长语音/视频、缩略图路径图片/视频、发送人、会话 ID、时间戳。按会话 ID 和时间建索引查某个好友的历史图片直接命中。多媒体下载对照类型转换额外元数据存储建议图片webp→jpg可选宽高、缩略图原图缩略图两份语音silk→mp3必须时长秒数单文件小按月归档视频不转封面、时长、分辨率对象存储文件不转文件名、大小对象存储下载存储管道实现import os, time, queue, threading, requests class MediaArchiver: def __init__(self, api, store, root/data/media): self.api api self.store store self.root root self.q queue.Queue() for _ in range(3): # 3 个下载线程 threading.Thread(targetself.worker, daemonTrue).start() def on_media_message(self, event): # 回调到达立即入下载队列 self.q.put({ msg_id: event[msg_id], type: event[type], url: event[media_url], retry: 0, }) def worker(self): while True: task self.q.get() try: local self._download(task) local self._convert(task[type], local) meta self._extract_meta(task[type], local) self.store.attach_media(task[msg_id], local_pathlocal, **meta) except Exception as e: if task[retry] 3: task[retry] 1 time.sleep(2 ** task[retry]) self.q.put(task) else: self.store.mark_media_failed( task[msg_id], str(e)) def _download(self, task): # 按 类型/年/月/日 组织目录 now time.localtime() subdir f{task[type]}/{now.tm_year}/{now.tm_mon:02d}/{now.tm_mday:02d} os.makedirs(f{self.root}/{subdir}, exist_okTrue) raw_path f{self.root}/{subdir}/{task[msg_id]}.raw with requests.get(task[url], streamTrue, timeout30) as r: r.raise_for_status() with open(raw_path, wb) as f: for chunk in r.iter_content(8192): f.write(chunk) return raw_path def _convert(self, mtype, raw_path): if mtype voice: # silk → mp3 out raw_path.replace(.raw, .mp3) os.system(fsilk-v3-decoder {raw_path} {out}) os.remove(raw_path) return out if mtype image: out raw_path.replace(.raw, .jpg) self._gen_thumb(raw_path, out.replace(.jpg, _thumb.jpg)) os.rename(raw_path, out) return out return raw_path.replace(.raw, .mp4 if mtype video else .bin) def _extract_meta(self, mtype, path): meta {size: os.path.getsize(path)} if mtype in (voice, video): meta[duration] self._probe_duration(path) if mtype video: meta[cover] self._extract_cover(path) if mtype image: meta[thumb] path.replace(.jpg, _thumb.jpg) return meta def _gen_thumb(self, src, dst): ... # 缩略图生成 def _probe_duration(self, path): ... # ffprobe 取时长 def _extract_cover(self, path): ... # 取首帧落地建议多媒体消息回调到达即入下载队列——别懒加载URL 过期就永远拿不回来了。下载失败指数退避重试 3 次趁 URL 有效期内处理失败任务。语音 silk 必须转 mp3 才能通用播放图片压缩略图列表页用视频提取封面和时长。存储按类型/年/月/日分目录文件名用 msg_id 反查消息元数据入库建会话索引。微信侧的多媒体消息与下载接口由Eyun 这类个人微信API平台 提供下载存储管道在自建服务实现具体字段以平台开发文档为准。
收到的多媒体消息回调里一般只有一个临时下载地址不下载就过期。下载存本地这套流程要解决四个问题下载时机、格式转换、存储组织、可检索索引。这篇把从回调到落库的完整管道讲清楚。一、下载时机回调到达即触发别懒加载最常见的错误是用到才下载——图片消息存了 URL一周后用户想看历史图片URL 早过期了。正确做法是回调到达时立即触发下载任务进下载队列异步执行下载完成把本地路径回写到消息记录。下载失败要重试——网络抖动常见重试 3 次指数退避还失败标记下载失败趁 URL 有效期内人工处理或等消息重推再下。多媒体下载地址的有效期说明在 Eyun 开发文档 中可以查到。二、格式转换语音 silk 转 mp3图片统一格式微信语音是 silk 格式浏览器和多数播放器不支持要用 silk-v3-decoder 转成 mp3 或 wav 再存。图片可能是 webp、gif 等格式按存储策略决定是否统一转 jpg/png。视频一般保持 mp4 不转但要提取封面图首帧和时长信息入库列表页展示用。三、存储组织按日期和类型分目录文件存储建议按类型/年/月/日组织目录比如 voice/2026/10/27/xxx.mp3。文件名用 msg_id 加后缀保证唯一且能反查消息。大图可以压一份缩略图宽度 300px列表页用缩略图、详情页用原图省流量。存储介质选择量小直接用本地磁盘加定期备份量大用对象存储OSS/COS数据库里只存路径和元数据不存二进制。四、检索索引让文件能被找到存了不用等于白存。消息记录表里多媒体消息要存msg_id、类型、本地路径、文件大小、时长语音/视频、缩略图路径图片/视频、发送人、会话 ID、时间戳。按会话 ID 和时间建索引查某个好友的历史图片直接命中。多媒体下载对照类型转换额外元数据存储建议图片webp→jpg可选宽高、缩略图原图缩略图两份语音silk→mp3必须时长秒数单文件小按月归档视频不转封面、时长、分辨率对象存储文件不转文件名、大小对象存储下载存储管道实现import os, time, queue, threading, requests class MediaArchiver: def __init__(self, api, store, root/data/media): self.api api self.store store self.root root self.q queue.Queue() for _ in range(3): # 3 个下载线程 threading.Thread(targetself.worker, daemonTrue).start() def on_media_message(self, event): # 回调到达立即入下载队列 self.q.put({ msg_id: event[msg_id], type: event[type], url: event[media_url], retry: 0, }) def worker(self): while True: task self.q.get() try: local self._download(task) local self._convert(task[type], local) meta self._extract_meta(task[type], local) self.store.attach_media(task[msg_id], local_pathlocal, **meta) except Exception as e: if task[retry] 3: task[retry] 1 time.sleep(2 ** task[retry]) self.q.put(task) else: self.store.mark_media_failed( task[msg_id], str(e)) def _download(self, task): # 按 类型/年/月/日 组织目录 now time.localtime() subdir f{task[type]}/{now.tm_year}/{now.tm_mon:02d}/{now.tm_mday:02d} os.makedirs(f{self.root}/{subdir}, exist_okTrue) raw_path f{self.root}/{subdir}/{task[msg_id]}.raw with requests.get(task[url], streamTrue, timeout30) as r: r.raise_for_status() with open(raw_path, wb) as f: for chunk in r.iter_content(8192): f.write(chunk) return raw_path def _convert(self, mtype, raw_path): if mtype voice: # silk → mp3 out raw_path.replace(.raw, .mp3) os.system(fsilk-v3-decoder {raw_path} {out}) os.remove(raw_path) return out if mtype image: out raw_path.replace(.raw, .jpg) self._gen_thumb(raw_path, out.replace(.jpg, _thumb.jpg)) os.rename(raw_path, out) return out return raw_path.replace(.raw, .mp4 if mtype video else .bin) def _extract_meta(self, mtype, path): meta {size: os.path.getsize(path)} if mtype in (voice, video): meta[duration] self._probe_duration(path) if mtype video: meta[cover] self._extract_cover(path) if mtype image: meta[thumb] path.replace(.jpg, _thumb.jpg) return meta def _gen_thumb(self, src, dst): ... # 缩略图生成 def _probe_duration(self, path): ... # ffprobe 取时长 def _extract_cover(self, path): ... # 取首帧落地建议多媒体消息回调到达即入下载队列——别懒加载URL 过期就永远拿不回来了。下载失败指数退避重试 3 次趁 URL 有效期内处理失败任务。语音 silk 必须转 mp3 才能通用播放图片压缩略图列表页用视频提取封面和时长。存储按类型/年/月/日分目录文件名用 msg_id 反查消息元数据入库建会话索引。微信侧的多媒体消息与下载接口由Eyun 这类个人微信API平台 提供下载存储管道在自建服务实现具体字段以平台开发文档为准。
