影刀RPA进阶教程_飞书表格自动读写与消息通知联动-程序员充电站

影刀RPA进阶教程：飞书表格自动读写与消息通知联动实战

影刀采集完数据、写进Excel后，如果你想把结果同步到飞书表格、再通知飞书群里的人——这个"采集→存云端→通知"的链路，用一个流程就能跑通。

飞书和影刀的协同是店群自动化里的高频需求。这篇专门讲怎么把飞书表格和飞书消息跟影刀串起来。

一、你需要准备的

飞书开放平台应用：在飞书开放平台（open.feishu.cn）创建一个应用，获取 App ID 和 App Secret
飞书表格（多维表格/Bitable）：创建一个表格，拿到表格的app_token和table_id
飞书群聊机器人：在群里添加自定义机器人，拿到 Webhook URL

这三个信息加一起，就能让影刀完成"采数据→写飞书表→发飞书消息"的完整流程。

二、影刀写飞书表格

飞书的 API 是 RESTful 格式。在影刀里调用飞书 API，用"HTTP请求"指令或"Python代码"指令都可以。建议用 Python 代码指令，逻辑更灵活。

第一步：获取飞书访问令牌

店群矩阵自动化突破运营极限！

# 在Python代码指令中importrequestsimportjson# 飞书应用凭证（从飞书开放平台获取）APP_ID="cli_xxxxxxxxxxxx"APP_SECRET="xxxxxxxxxxxxxxxxxxxxxx"# 获取 tenant_access_tokendefget_tenant_token():url="https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"payload={"app_id":APP_ID,"app_secret":APP_SECRET}resp=requests.post(url,json=payload)returnresp.json()["tenant_access_token"]TOKEN=get_tenant_token()

第二步：往飞书多维表格追加数据

# 飞书多维表格信息APP_TOKEN="bascnxxxxxxxxxxxx"# 表格的app_tokenTABLE_ID="tblxxxxxxxxxxxx"# 数据表的table_id# 追加一条记录到飞书表格defadd_record(token,fields):url=f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records"headers={"Authorization":f"Bearer{token}","Content-Type":"application/json"}payload={"fields":fields# 列名和值的字典}resp=requests.post(url,headers=headers,json=payload)returnresp.json()# 示例：追加一条商品数据商品数据={"商品名称":"夏季连衣裙","价格":128.00,"采集时间":"2026-06-09 14:30","来源平台":"拼多多","备注":"销量10万+"}result=add_record(TOKEN,商品数据)print(result)

第三步：批量追加数据

真实场景里是从Excel逐行读到飞书：

importpandasaspd# 读取影刀刚才导出的Exceldf=pd.read_excel(r"D:\拼多多_连衣裙_清洗后.xlsx")TOKEN=get_tenant_token()成功数=0失败数=0foridx,rowindf.iterrows():fields={"商品名称":str(row['商品名称']),"价格":float(row['价格(元)']),"已拼件数":str(row['已拼件数']),"店铺名":str(row['店铺名']),"采集时间":"2026-06-09","来源平台":"拼多多"}result=add_record(TOKEN,fields)ifresult.get("code")==0:成功数+=1else:失败数+=1print(f"写入失败：{row['商品名称']}，错误：{result}")print(f"写入完成！成功{成功数}条，失败{失败数}条")

三、飞书群消息通知

采集完成或写入飞书表格后，用飞书群机器人发一条汇总消息。

# 飞书群机器人Webhook URL（在群设置中添加机器人获取）WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxxxxxx"defsend_feishu_message(title,content):payload={"msg_type":"interactive","card":{"header":{"title":{"tag":"plain_text","content":title},"template":"blue"},"elements":[{"tag":"div","text":{"tag":"lark_md","content":content}}]}}resp=requests.post(WEBHOOK_URL,json=payload)returnresp.json()# 发送采集完成通知汇总信息=f""" **采集任务完成** - 搜索关键词：连衣裙 - 采集页数：10页 - 采集商品数：{成功数+失败数}条 - 写入飞书表格：{成功数}条成功 /{失败数}条失败 - 完成时间：2026-06-09 14:35 """send_feishu_message("拼多多数据采集完成",汇总信息)

四、完整联动流程

把上面所有代码串联起来——一个完整的"采集→清洗→写飞书→发通知"流程：

# ===== 影刀指令部分 =====打开网页("https://mobile.yangkeduo.com/search_result.html?search_key=连衣裙")等待元素出现("商品列表容器",5秒)新建Excel->原始数据表 写入行数据(原始数据表,["商品名称","价格(元)","已拼件数","店铺名"])# 翻页采集逻辑（参考拼多多采集文章）# ... 采集代码 ...导出表格(原始数据表,"D:\拼多多_连衣裙_原始.xlsx")# ===== Python代码指令部分 =====importpandasaspdimportrequests# 1. 读取影刀导出的数据df=pd.read_excel(r"D:\拼多多_连衣裙_原始.xlsx")# 2. 清洗价格df['价格(元)']=df['价格(元)'].str.replace(r'[¥￥\s]','',regex=True).astype(float)# 3. 获取飞书tokenAPP_ID="cli_xxxxxxxxxxxx"APP_SECRET="xxxxxxxxxxxxxxxxxxxxxx"resp=requests.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",json={"app_id":APP_ID,"app_secret":APP_SECRET})TOKEN=resp.json()["tenant_access_token"]# 4. 写入飞书表格APP_TOKEN="bascnxxxxxxxxxxxx"TABLE_ID="tblxxxxxxxxxxxx"WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxxxxxx"成功=0失败=0foridx,rowindf.iterrows():url=f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records"headers={"Authorization":f"Bearer{TOKEN}","Content-Type":"application/json"}body={"fields":{"商品名称":str(row['商品名称']),"价格":float(row['价格(元)']),"已拼件数":str(row['已拼件数']),"店铺名":str(row['店铺名']),"采集时间":"2026-06-09"}}r=requests.post(url,headers=headers,json=body)ifr.json().get("code")==0:成功+=1else:失败+=1# 5. 发送飞书通知requests.post(WEBHOOK_URL,json={"msg_type":"interactive","card":{"header":{"title":{"tag":"plain_text","content":"拼多多数据采集完成"},"template":"green"},"elements":[{"tag":"div","text":{"tag":"lark_md","content":f"写入飞书表格：成功{成功}条 / 失败{失败}条"}}]}})print(f"全流程完成！成功{成功}条，失败{失败}条")

五、API调用频率限制

飞书 API 有频率限制（QPS）。对于多维表格的写入：

普通应用：每秒最多20次
批量写入接口：可以用batch_create_records一次写最多500条

如果你的数据超过几十条，建议用批量接口：

# 批量创建记录（一次最多500条）defbatch_add_records(token,records_list):url=f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records/batch_create"headers={"Authorization":f"Bearer{token}","Content-Type":"application/json"}# records_list 是列表，每个元素是 {"fields": {...}}payload={"records":records_list}resp=requests.post(url,headers=headers,json=payload)returnresp.json()