最近做的企微二开客户经常发文件——订单表、合同、发票。之前机器人收到文件只会回已收到。业务方要求文件能自动处理识别是什么文件、提取内容、走对应业务流程。这篇重点讲文件处理每个环节调了哪些企微接口——文件消息接收、CDN 下载、消息回复。底层用的是Eyun 平台开放的企微 API统一 POSTJSON鉴权用 App Token 加 appid响应封套{code, data, detail, message, time}code 为 0 成功。文件消息接收Webhook 拿 fileId客户发文件Webhook 回调里消息类型是file带fileId、fileName、fileSizeapp.route(/wx-api/webhook/, methods[POST]) def webhook(): payload request.json[data] if payload[msgType] ! file: return ok file_id payload[fileId] file_name payload[fileName] appid payload[appid] from_uin payload[fromUin] # 异步处理文件 process_file.delay(appid, from_uin, file_id, file_name) return ok文件处理耗时长不能在回调里同步做入队异步处理先回收到文件处理中。文件下载调 CDN 接口拿到 fileId 后调 CDN 模块下载文件def download_file(appid, file_id, save_path): resp requests.post( f{BASE}/wx-api/api/cdn/download, headersHEADERS, json{appid: appid, fileId: file_id}, streamTrue ) if resp.json().get(code) 0: with open(save_path, wb) as f: for chunk in resp.iter_content(8192): f.write(chunk) return save_path raise Exception(resp.json()[message])下载是流式的大文件不会爆内存。下载完存 OSS本地不留。CDN 接口的 fileId 是回调里给的不能自己拼。接口参数在Eyun 开发文档。文件类型识别按扩展名和内容下载完识别文件类型决定怎么解析import os def detect_file_type(file_name, file_path): ext os.path.splitext(file_name)[1].lower() if ext in [.xlsx, .xls, .csv]: return excel elif ext in [.pdf, .doc, .docx]: return document elif ext in [.jpg, .png, .jpeg]: return image else: return other类型识别按扩展名加内容双重判断防止客户改扩展名骗识别。识别错类型解析全错。Excel 解析提取订单数据Excel 用 openpyxl 读提取结构化数据。客户的表头可能不统一做别名匹配import openpyxl def parse_excel(file_path): wb openpyxl.load_workbook(file_path) ws wb.active headers [c.value for c in ws[1]] # 表头别名映射 col_map {} for i, h in enumerate(headers): if h in [订单号, 订单编号, order_no]: col_map[order_no] i elif h in [产品, 商品名称, product]: col_map[product] i elif h in [数量, qty]: col_map[quantity] i orders [] for row in ws.iter_rows(min_row2, values_onlyTrue): order {k: row[v] for k, v in col_map.items() if v len(row)} if order.get(order_no): orders.append(order) return orders解析完校验——订单号不为空、数量是数字。校验不通过的行标记错误。业务回写进订单系统解析成功的订单批量创建。回写要幂等——同一文件重复上传不重复创建。幂等键用文件名加内容哈希import hashlib def create_orders(orders, file_name, file_content): idem_key ffile:{hashlib.md5((file_namefile_content).encode()).hexdigest()} if redis.set(idem_key, 1, nxTrue, ex86400): return order_api.batch_create(orders) else: return {status: duplicate}不做幂等客户重发一次文件订单建两遍。处理结果回复调消息接口处理完调消息接口把结果回给客户def reply_result(appid, to_uin, total, success, failed_rows): text f文件处理完成共 {total} 行成功 {success} 行 if failed_rows: text f失败 {len(failed_rows)} 行\n for r in failed_rows[:5]: text f- 第{r[row]}行{r[reason]}\n requests.post( f{BASE}/wx-api/api/message/sendText, headersHEADERS, json{appid: appid, to: to_uin, content: text} )部分成功要明确告诉客户哪些成功哪些失败不能含糊说处理完成。异步队列大文件不阻塞文件解析耗时长走异步队列。回调里只下载和入队worker 处理from celery import Celery celery Celery(file_processor) celery.task def process_file(appid, from_uin, file_id, file_name): local_path download_file(appid, file_id, /tmp) ftype detect_file_type(file_name, local_path) if ftype excel: orders parse_excel(local_path) result create_orders(orders, file_name, open(local_path).read()) reply_result(appid, from_uin, len(orders), result[success], result[failed]) else: reply_result(appid, from_uin, 0, 0, [{row: 0, reason: 不支持的文件类型}]) os.remove(local_path)并发控制——同时处理太多文件解析服务扛不住加令牌桶限流。写在最后文件自动处理这套东西本质是用好企微的文件和消息接口——Webhook 收文件消息拿 fileId、CDN download 下载文件、sendText 回处理结果。文件解析类型识别、Excel 提取、业务回写在应用层做。接口路径、参数、CDN 下载方式在Eyun 开发文档里。把接口调对、解析做扎实客户发的文件真正能被系统处理——而不是躺在消息里等人手动导。凭证和接入地址在Eyun 企业微信 API 平台开通。