收到的多媒体消息回调里一般只有一个临时下载地址，不下载就过期。下载存本地这套流程要解决四个问题：下载时机、格式转换、存储组织、可检索索引。这篇把从回调到落库的完整管道讲清楚。
一、下载时机：回调到达即触发，别懒加载
最常见的错误是"用到才下载"——图片消息存了 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(target=self.worker, daemon=True).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_path=local, **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_ok=True) raw_path = f"{self.root}/{subdir}/{task['msg_id']}.raw" with requests.get(task["url"], stream=True, timeout=30) 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(f"silk-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平台 提供，下载存储管道在自建服务实现，具体字段以平台开发文档为准。
