企业官网建设流程全解析

影刀RPA深度实战：多源数据采集到BI看板自动化的完整Pipeline

痛点切入

我去年帮一个年销过亿的天猫店做数据中台，他们当时的数据流是真的乱：运营每天要登录天猫、京东、拼多多、抖音四个平台后台导出Excel，手动合并到一个"老板看板"里。不光耗时，还经常出错——有时候数值对不上、有时候忘记导出、有时候公式被覆盖了。

更要命的是，老板每天早上要看的BI看板上数据总是滞后24小时以上。我当时就想：能不能把整个数据流自动化——从采集到入库到BI刷新全链路打通？

答案是能。而且整套方案用影刀RPA + MySQL + 飞书多维表格就搞定了，不用买几万块的数据中台产品。这篇把完整Pipeline拆给你。

核心方法

场景：多平台电商数据自动汇聚到BI看板

某电商公司的数据流：

1. 每日凌晨自动从4个平台采集销售、流量、转化数据
2. 标准化清洗后写入MySQL数据库
3. 飞书多维表格（Base）作为BI前端，定时刷新
4. 老板和运营每天早上看到的是昨天的最新数据

整体架构

[天猫后台] ──┐ [京东后台] ──┤ [拼多多后台]─┼── RPA采集 ──→ 数据清洗 ──→ MySQL ──→ 飞书Base(BI看板) ──→ 老板 [抖音后台] ──┘ │ └──→ Excel备份

分步操作

第一步：多源数据采集

在影刀中为每个平台建立独立的采集子流程，然后通过主流程串联调用。

天猫后台采集子流程：

1. 「打开网页」→https://myseller.taobao.com/→ 登录

2. 「点击」→「生意参谋」→「流量」→「店铺来源」

3. 设置日期筛选（昨天）

5. 「数据抓取」→ 抓取：访客数、支付金额、支付转化率、客单价等

核心XPath：

// 生意参谋数据表格 表格行：//div[contains(@class,'data-table')]//tr[contains(@class,'row')] // 各指标列（根据实际页面结构调整） - 访客数：.//td[@data-field='uv'] - 支付金额：.//td[@data-field='pay_amt'] - 转化率：.//td[@data-field='conversion_rate'] - 客单价：.//td[@data-field='avg_price'] // 日期选择器 日期输入：//input[@placeholder='选择日期'] 昨天快捷：//span[contains(text(),'昨天')]

京东商智采集（步骤同理）：

// 京东商智后台特殊点：数据经常需要切换tab才会加载 流量概况tab：//div[contains(@class,'tab-item') and contains(text(),'流量')] 数据表格：//table[contains(@class,'report-table')]//tr 常见问题：京东页面有iframe嵌套，影刀需要先「切换到iframe」再操作 iframe定位：//iframe[contains(@src,'index.jd.com')]

采集结果汇总为统一格式：

# Python节点：多源数据标准化""" 将4个平台的原始数据统一为标准格式 不同平台的字段名不一样，这里做映射和单位统一 作者：林焱 """importpandasaspdfromdatetimeimportdatetime,timedelta yesterday=(datetime.now()-timedelta(days=1)).strftime("%Y-%m-%d")# 影刀变量读取各平台数据tmall_data=GetVariable("tmall_data")# list of dictjd_data=GetVariable("jd_data")pdd_data=GetVariable("pdd_data")douyin_data=GetVariable("douyin_data")defnormalize_platform_data(raw_data,platform_name):""" 标准化函数：将不同平台的字段统一映射 核心字段：日期、平台、销售额、订单量、访客数、转化率、客单价 """normalized=[]forrowinraw_data:record={"date":yesterday,"platform":platform_name,"sales_amount":float(row.get("支付金额",row.get("gmv",0))),"order_count":int(row.get("支付订单数",row.get("order_cnt",0))),"uv":int(row.get("访客数",row.get("visitors",0))),"conversion_rate":float(row.get("转化率",row.get("cvr",0))),"avg_price":float(row.get("客单价",row.get("asp",0))),"refund_rate":float(row.get("退款率",row.get("refund_rate",0))),"collected_at":datetime.now().strftime("%Y-%m-%d %H:%M:%S")}normalized.append(record)returnnormalized# 各平台标准化all_data=[]all_data.extend(normalize_platform_data(tmall_data,"天猫"))all_data.extend(normalize_platform_data(jd_data,"京东"))all_data.extend(normalize_platform_data(pdd_data,"拼多多"))all_data.extend(normalize_platform_data(douyin_data,"抖音"))# 汇总计算df=pd.DataFrame(all_data)summary={"date":yesterday,"total_sales":round(df["sales_amount"].sum(),2),"total_orders":int(df["order_count"].sum()),"total_uv":int(df["uv"].sum()),"platform_count":len(df)}print(f"\n{yesterday}数据汇总:")print(f" 总销售额:{summary['total_sales']:.2f}")print(f" 总订单数:{summary['total_orders']}")print(f" 总访客:{summary['total_uv']}")# 回写变量SetVariable("normalized_data",all_data)SetVariable("daily_summary",summary)

第二步：数据写入MySQL数据库

""" 影刀RPA — 数据写入MySQL模块 需要先安装: pip install pymysql sqlalchemy 作者：林焱 """importpymysqlfromsqlalchemyimportcreate_engineimportpandasaspd# ========== 数据库配置 ==========DB_CONFIG={"host":"192.168.1.100",# 你的MySQL服务器地址"port":3306,"user":"rpa_user","password":"你的数据库密码","database":"ecommerce_dw",# 数据仓库库名"charset":"utf8mb4"}# ================================# 获取标准化数据normalized_data=GetVariable("normalized_data")df=pd.DataFrame(normalized_data)defwrite_to_mysql(df,table_name,if_exists='append'):""" 写入MySQL数据库 if_exists参数： - 'append': 追加（日常采集用这个） - 'replace': 替换整表（首次建表用） - 'fail': 表存在则报错 """# 构造连接字符串conn_str=(f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"f"@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"f"?charset={DB_CONFIG['charset']}")engine=create_engine(conn_str)try:# 写入数据库df.to_sql(name=table_name,con=engine,if_exists=if_exists,index=False,method='multi',# 批量插入，提速chunksize=100# 每批100行)print(f"[√] 数据已写入MySQL表{table_name}，共{len(df)}行")exceptExceptionase:print(f"[×] 写入失败:{e}")raisefinally:engine.dispose()defcheck_duplicate_dates(df,table_name):""" 去重检查：避免同一天的数据重复写入 通常采集中途断掉重跑时会出现这个问题 """conn_str=(f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"f"@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")engine=create_engine(conn_str)# 检查今日是否已有数据yesterday=df['date'].iloc[0]query=f"SELECT COUNT(*) as cnt FROM{table_name}WHERE date = '{yesterday}'"existing=pd.read_sql(query,engine)ifexisting['cnt'].iloc[0]>0:print(f"[!]{yesterday}的数据已存在（{existing['cnt'].iloc[0]}条），将删除后重新写入")# 删除旧数据withengine.connect()asconn:conn.execute(f"DELETE FROM{table_name}WHERE date = '{yesterday}'")conn.commit()# 注意：这里需要commit，pymysql默认不自动提交engine.dispose()# ========== 执行写入 ==========# 1. 写入事实表（每日平台明细）check_duplicate_dates(df,"daily_platform_sales")write_to_mysql(df,"daily_platform_sales",if_exists='append')# 2. 写入汇总表（每日总览）summary=GetVariable("daily_summary")summary_df=pd.DataFrame([summary])check_duplicate_dates(summary_df,"daily_summary")write_to_mysql(summary_df,"daily_summary",if_exists='append')print("\n数据入库完成！")

第三步：创建MySQL数据视图（可选但推荐）

-- 在MySQL中创建几个常用视图，方便BI工具直接查询-- 不需要在影刀里创建，在数据库里执行一次就行-- 视图1：近30天各平台销售趋势CREATEORREPLACEVIEWv_30days_platform_trendASSELECTdate,platform,SUM(sales_amount)astotal_sales,SUM(order_count)astotal_orders,AVG(conversion_rate)asavg_cvr,AVG(avg_price)asavg_aspFROMdaily_platform_salesWHEREdate>=DATE_SUB(CURDATE(),INTERVAL30DAY)GROUPBYdate,platformORDERBYdateDESC,platform;-- 视图2：昨日全渠道快报CREATEORREPLACEVIEWv_yesterday_briefASSELECTplatform,sales_amount,order_count,uv,CONCAT(ROUND(conversion_rate*100,2),'%')ascvr_display,avg_price,refund_rateFROMdaily_platform_salesWHEREdate=DATE_SUB(CURDATE(),INTERVAL1DAY);-- 视图3：月度汇总（用于月报）CREATEORREPLACEVIEWv_monthly_summaryASSELECTDATE_FORMAT(date,'%Y-%m')asmonth,platform,SUM(sales_amount)asmonthly_sales,SUM(order_count)asmonthly_orders,AVG(conversion_rate)asavg_cvrFROMdaily_platform_salesGROUPBYDATE_FORMAT(date,'%Y-%m'),platformORDERBYmonthDESC,platform;

第四步：飞书多维表格（Base）作为BI看板

飞书Base支持外部数据源连接，但更简单的方式是直接用API写入：

""" 影刀RPA — 飞书Base数据同步模块 将MySQL每日汇总数据同步到飞书多维表格作为BI看板 作者：林焱 """importrequests# 飞书Base配置BASE_APP_TOKEN="你的Base应用Token"TABLE_ID="tblXXXXXXXX"# 目标数据表IDFEISHU_TENANT_TOKEN="你的飞书租户AccessToken"defget_feishu_token():"""获取飞书访问凭证"""resp=requests.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",json={"app_id":"你的App_ID","app_secret":"你的App_Secret"})returnresp.json().get("tenant_access_token")defappend_to_base(token,data_rows):"""追加数据到飞书Base"""url=(f"https://open.feishu.cn/open-apis/bitable/v1/apps/"f"{BASE_APP_TOKEN}/tables/{TABLE_ID}/records/batch_create")headers={"Authorization":f"Bearer{token}","Content-Type":"application/json"}# 构造记录（根据你的Base表格字段调整）records=[]forrowindata_rows:records.append({"fields":{"日期":row["date"],"平台":row["platform"],"销售额（元）":row["sales_amount"],"订单量":row["order_count"],"访客数":row["uv"],"转化率":f"{row['conversion_rate']*100:.2f}%","客单价（元）":row["avg_price"]}})resp=requests.post(url,headers=headers,json={"records":records},timeout=30)result=resp.json()ifresult.get("code")==0:print(f"[√] 飞书Base同步成功，写入{len(records)}条记录")else:print(f"[×] 同步失败:{result}")returnresult# ========== 执行同步 ==========normalized_data=GetVariable("normalized_data")token=get_feishu_token()append_to_base(token,normalized_data)

第五步：影刀定时调度

在影刀中设置定时任务：

• 主流程：每天07:00执行（确保各平台前一天数据已生成）
• 子流程调用顺序：天猫采集 → 京东采集 → 拼多多采集 → 抖音采集 → 数据标准化 → 写入MySQL → 同步飞书Base
• 异常处理：任一子流程失败，记录错误日志，飞书通知管理员
• 重试策略：失败后30分钟重试一次，最多3次

# 失败告警Python节点""" 采集流程异常时的飞书告警 """importrequests error_info=GetVariable("error_info")# 影刀错误变量msg={"msg_type":"text","content":{"text":(f"⚠️ 数据采集Pipeline异常\n"f"时间：{error_info.get('time')}\n"f"环节：{error_info.get('step')}\n"f"错误：{error_info.get('message')}\n"f"请尽快检查！")}}WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxxxx"requests.post(WEBHOOK_URL,json=msg)

注意点

1. MySQL字符集：一定要用utf8mb4而不是utf8。否则遇到emoji和生僻字会报"Incorrect string value"错误，排查起来很痛苦。

2. 去重逻辑：定时任务可能会因为临时故障手动重跑，必须先去重检查。我最早没加这个，结果发现数据库里有同一天的三条重复数据。

3. 数据库密码安全：不要把密码硬编码在代码里。建议用影刀的环境变量功能存储敏感信息，代码里通过变量读取。

5. 网络稳定性：各平台的采集页面有时候加载很慢。关键步骤加「等待元素」节点和超时重试逻辑，不要设死时间间隔。

常见问题速查

推荐资源

• 飞书Base(多维表格)开放API文档
• MySQL 8.0官方文档
• SQLAlchemy ORM教程
• 影刀RPA子流程调用教程

内容标签：#RPA #数据仓库 #BI #MySQL #飞书Base #数据Pipeline #电商数据 #自动化

作者：林焱