最近做的企微二開客戶經常發(fā)文件——訂單表、合同、發(fā)票。之前機器人收到文件只會回已收到。業(yè)務方要求文件能自動處理識別是什么文件、提取內容、走對應業(yè)務流程。這篇重點講文件處理每個環(huán)節(jié)調了哪些企微接口——文件消息接收、CDN 下載、消息回復。底層用的是Eyun 平臺開放的企微 API統(tǒng)一 POSTJSON鑒權用 App Token 加 appid響應封套{code, data, detail, message, time}code 為 0 成功。文件消息接收Webhook 拿 fileId客戶發(fā)文件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 開發(fā)文檔。文件類型識別按擴展名和內容下載完識別文件類型決定怎么解析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 讀提取結構化數據?？蛻舻谋眍^可能不統(tǒng)一做別名匹配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解析完校驗——訂單號不為空、數量是數字。校驗不通過的行標記錯誤。業(yè)務回寫進訂單系統(tǒng)解析成功的訂單批量創(chuàng)建。回寫要冪等——同一文件重復上傳不重復創(chuàng)建。冪等鍵用文件名加內容哈希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}不做冪等客戶重發(fā)一次文件訂單建兩遍。處理結果回復調消息接口處理完調消息接口把結果回給客戶def reply_result(appid, to_uin, total, success, failed_rows): text f文件處理完成共 {total} 行成功 {success} 行 if failed_rows: text f失敗 {len(failed_rows)} 行\(zhòng)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)并發(fā)控制——同時處理太多文件解析服務扛不住加令牌桶限流。寫在最后文件自動處理這套東西本質是用好企微的文件和消息接口——Webhook 收文件消息拿 fileId、CDN download 下載文件、sendText 回處理結果。文件解析類型識別、Excel 提取、業(yè)務回寫在應用層做。接口路徑、參數、CDN 下載方式在Eyun 開發(fā)文檔里。把接口調對、解析做扎實客戶發(fā)的文件真正能被系統(tǒng)處理——而不是躺在消息里等人手動導。憑證和接入地址在Eyun 企業(yè)微信 API 平臺開通。