最近做的企微二開客戶經(jīng)常發(fā)文件——訂單表、合同、發(fā)票。之前機(jī)器人收到文件只會(huì)回已收到。業(yè)務(wù)方要求文件能自動(dòng)處理識(shí)別是什么文件、提取內(nèi)容、走對(duì)應(yīng)業(yè)務(wù)流程。這篇重點(diǎn)講文件處理每個(gè)環(huán)節(jié)調(diào)了哪些企微接口——文件消息接收、CDN 下載、消息回復(fù)。底層用的是Eyun 平臺(tái)開放的企微 API統(tǒng)一 POSTJSON鑒權(quán)用 App Token 加 appid響應(yīng)封套{code, data, detail, message, time}code 為 0 成功。文件消息接收Webhook 拿 fileId客戶發(fā)文件Webhook 回調(diào)里消息類型是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文件處理耗時(shí)長(zhǎng)不能在回調(diào)里同步做入隊(duì)異步處理先回收到文件處理中。文件下載調(diào) CDN 接口拿到 fileId 后調(diào) 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])下載是流式的大文件不會(huì)爆內(nèi)存。下載完存 OSS本地不留。CDN 接口的 fileId 是回調(diào)里給的不能自己拼。接口參數(shù)在Eyun 開發(fā)文檔。文件類型識(shí)別按擴(kuò)展名和內(nèi)容下載完識(shí)別文件類型決定怎么解析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類型識(shí)別按擴(kuò)展名加內(nèi)容雙重判斷防止客戶改擴(kuò)展名騙識(shí)別。識(shí)別錯(cuò)類型解析全錯(cuò)。Excel 解析提取訂單數(shù)據(jù)Excel 用 openpyxl 讀提取結(jié)構(gòu)化數(shù)據(jù)?？蛻舻谋眍^可能不統(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 [訂單號(hào), 訂單編號(hào), order_no]: col_map[order_no] i elif h in [產(chǎn)品, 商品名稱, product]: col_map[product] i elif h in [數(shù)量, 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àn)——訂單號(hào)不為空、數(shù)量是數(shù)字。校驗(yàn)不通過(guò)的行標(biāo)記錯(cuò)誤。業(yè)務(wù)回寫進(jìn)訂單系統(tǒng)解析成功的訂單批量創(chuàng)建?；貙懸獌绲取晃募貜?fù)上傳不重復(fù)創(chuàng)建。冪等鍵用文件名加內(nèi)容哈希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ā)一次文件訂單建兩遍。處理結(jié)果回復(fù)調(diào)消息接口處理完調(diào)消息接口把結(jié)果回給客戶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} )部分成功要明確告訴客戶哪些成功哪些失敗不能含糊說(shuō)處理完成。異步隊(duì)列大文件不阻塞文件解析耗時(shí)長(zhǎng)走異步隊(duì)列?；卣{(diào)里只下載和入隊(duì)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ā)控制——同時(shí)處理太多文件解析服務(wù)扛不住加令牌桶限流。寫在最后文件自動(dòng)處理這套東西本質(zhì)是用好企微的文件和消息接口——Webhook 收文件消息拿 fileId、CDN download 下載文件、sendText 回處理結(jié)果。文件解析類型識(shí)別、Excel 提取、業(yè)務(wù)回寫在應(yīng)用層做。接口路徑、參數(shù)、CDN 下載方式在Eyun 開發(fā)文檔里。把接口調(diào)對(duì)、解析做扎實(shí)客戶發(fā)的文件真正能被系統(tǒng)處理——而不是躺在消息里等人手動(dòng)導(dǎo)。憑證和接入地址在Eyun 企業(yè)微信 API 平臺(tái)開通。