系列里有多篇文章分别讲了数据采集、数据清洗、飞书写入、企业微信通知。但都是“单一功能”，没有一篇把数据汇总 → 日报生成 → 多渠道推送 → 异常告警串成一条完整的链路。
今天把这条链路打通——用一套完整的“日报自动生成与推送系统”做案例，把所有环节串起来。
先说这套系统能做什么
每天定时自动完成：
- 从多个数据源采集当日数据
- 按业务维度汇总（销售额、订单量、TOP商品）
3. 生成格式化的日报内容（文本版 + 表格版）
- 推送到飞书/企业微信群
- 自动检测异常数据并单独告警
整个过程，不需要任何人手动操作。
系统架构
【定时触发】每天早上 09:00
↓
【数据采集】
├─ 淘宝数据（昨日销售额/订单量/TOP商品）
├─ 拼多多数据（昨日销售额/订单量/TOP商品）
└─ 内部系统数据（昨日总订单/退款/客诉）
↓
【数据汇总】（Python）
├─ 合并三平台数据
├─ 计算汇总指标（总计、同比、环比）
└─ 检测异常（销售额骤降、退款率飙升）
↓
【内容生成】（Python）
├─ 生成Markdown格式日报
└─ 生成飞书多维表格格式数据
↓
【多渠道推送】
├─ 飞书群（日报消息）
├─ 企业微信群（日报消息）
└─ 飞书多维表格（明细数据）
↓
【异常告警】
└─ 如有异常 → 单独发告警消息（@负责人）
第一步：数据采集（多平台统一格式）
各平台采集子流程输出统一格式的数据：
# ===== 统一输出格式 =====
# 每个平台返回的列表，每行包含：
[
"平台名称", # 淘宝/拼多多/内部
"商品名", # 商品名称
"销售额", # 当日销售额（数字）
"销量", # 当日销量（数字）
"订单量", # 当日订单数（数字）
"退款金额", # 当日退款金额（数字）
"客诉数", # 当日客诉数（数字）
"采集时间" # 采集时间戳
]
# ===== 淘宝采集子流程示例（只展示清洗后的输出）=====
# 子流程：Fetch_Taobao
# 输出：淘宝数据列表（统一格式）
设置变量(淘宝数据 = [])
# ... 采集逻辑（详见第30篇）...
# 每条数据按统一格式组装
按列表循环(采集结果, 循环项=商品):
一行 = [
"淘宝",
商品["标题"],
商品["销售额"], # 已清洗为数字
商品["销量"], # 已清洗为数字
商品["订单数"],
商品["退款金额"],
商品["客诉数"],
获取当前时间()
]
将 一行 加入列表(淘宝数据)
结束循环
返回结果(淘宝数据)
调度流程中合并数据：
# ===== 主流程：采集所有平台 =====
调用子流程(Fetch_Taobao, 输出=淘宝数据)
调用子流程(Fetch_Pinduoduo, 输出=拼多多数据)
调用子流程(Fetch_Internal, 输出=内部数据)
# 合并（所有平台数据格式统一）
合并列表(淘宝数据, 拼多多数据, 内部数据, 存入列表=全部原始数据)
输出日志("采集完成，共 " + 获取列表长度(全部原始数据) + " 条")
第二步：数据汇总与指标计算（Python）
# ============================================
# Python段：数据汇总与日报生成
# ============================================
import pandas as pd
import json
from datetime import datetime
# ---- 输入：全部原始数据 ----
# raw_data: [[平台, 商品名, 销售额, 销量, 订单量, 退款, 客诉, 时间], ...]
df = pd.DataFrame(原始数据, columns=[
'平台', '商品名', '销售额', '销量', '订单量', '退款', '客诉', '采集时间'
])
# ----- 1. 按平台汇总 -----
平台汇总 = df.groupby('平台').agg({
'销售额': 'sum',
'销量': 'sum',
'订单量': 'sum',
'退款': 'sum',
'客诉': 'sum'
}).reset_index()
# ----- 2. 全平台汇总 -----
总销售额 = df['销售额'].sum()
总销量 = df['销量'].sum()
总订单量 = df['订单量'].sum()
总退款 = df['退款'].sum()
总客诉 = df['客诉'].sum()
# 退款率
退款率 = (总退款 / 总销售额 * 100) if 总销售额 > 0 else 0
# ----- 3. TOP商品（全平台）-----
TOP商品 = df.nlargest(5, '销售额')[['平台', '商品名', '销售额', '销量']]
# ----- 4. 异常检测 -----
异常列表 = []
# 4.1 销售额为0的平台
零销售额平台 = 平台汇总[平台汇总['销售额'] == 0]['平台'].tolist()
if 零销售额平台:
异常列表.append({
'类型': '零销售额',
'说明': f"以下平台销售额为0：{', '.join(零销售额平台)}"
})
# 4.2 退款率过高（>5%）
if 退款率 > 5:
异常列表.append({
'类型': '退款率过高',
'说明': f"退款率为 {退款率:.1f}%，超过 5% 阈值"
})
# 4.3 客诉过多（>10）
if 总客诉 > 10:
异常列表.append({
'类型': '客诉过多',
'说明': f"今日客诉 {总客诉} 件，超过 10 件阈值"
})
# 4.4 销量为0的商品（前10条）
零销量商品 = df[df['销量'] == 0]['商品名'].head(10).tolist()
if 零销量商品:
异常列表.append({
'类型': '零销量商品',
'说明': f"以下商品销量为0：{', '.join(零销量商品[:5])}" + ("..." if len(零销量商品) > 5 else "")
})
# ----- 5. 组装汇总结果 -----
汇总结果 = {
'日期': datetime.now().strftime('%Y-%m-%d'),
'总销售额': round(总销售额, 2),
'总销量': int(总销量),
'总订单量': int(总订单量),
'退款率': round(退款率, 2),
'总客诉': int(总客诉),
'平台明细': 平台汇总.to_dict('records'),
'TOP商品': TOP商品.to_dict('records'),
'异常列表': 异常列表,
'是否有异常': len(异常列表) > 0
}
# ---- 输出 ----
汇总JSON = json.dumps(汇总结果, ensure_ascii=False)
第三步：生成日报内容
# ============================================
# 生成飞书/企业微信日报（Markdown格式）
# ============================================
def 生成日报内容(汇总结果):
日期 = 汇总结果['日期']
总销售额 = 汇总结果['总销售额']
总销量 = 汇总结果['总销量']
总订单量 = 汇总结果['总订单量']
退款率 = 汇总结果['退款率']
总客诉 = 汇总结果['总客诉']
平台明细 = 汇总结果['平台明细']
TOP商品 = 汇总结果['TOP商品']
异常列表 = 汇总结果['异常列表']
是否有异常 = 汇总结果['是否有异常']
内容 = f"""
📊 **{日期} 数据日报**
━━━━━━━━━━━━━━━━━━━━━━━━━━━
💰 **核心指标**
• 总销售额：¥{总销售额:,.2f}
• 总销量：{总销量} 件
• 总订单量：{总订单量} 单
• 退款率：{退款率:.1f}%
• 客诉数：{总客诉} 件
📋 **各平台明细**
"""
# 添加平台明细
for item in 平台明细:
内容 += f"\n• {item['平台']}：销售额 ¥{item['销售额']:,.2f}，销量 {item['销量']}，订单 {item['订单量']}"
内容 += "\n\n🏆 **TOP5 热销商品**"
for i, item in enumerate(TOP商品):
内容 += f"\n{i+1}. {item['商品名']}（{item['平台']}）¥{item['销售额']:,.2f}，销量 {item['销量']}"
# 异常信息
if 是否有异常:
内容 += "\n\n⚠️ **异常告警**"
for 异常 in 异常列表:
内容 += f"\n• {异常['类型']}：{异常['说明']}"
else:
内容 += "\n\n✅ 今日无异常"
内容 += f"\n\n━━━━━━━━━━━━━━━━━━━━━━━━━━━\n🕐 更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}"
return 内容
第四步：推送飞书/企业微信
# ============================================
# 子流程：发送日报
# ============================================
# 输入参数：汇总JSON
# ----- 1. 生成日报内容 -----
执行Python代码(
输入: 汇总JSON,
输出: 日报文本,
代码=生成日报内容的Python代码
)
# ----- 2. 发送飞书 -----
调用子流程(Send_FeishuNotification,
输入参数:
msg_type="info",
title="数据日报 - " + 当前日期,
content=日报文本,
mention_all=是否有异常 # 有异常时@所有人
)
# ----- 3. 发送企业微信 -----
调用子流程(Send_WeworkNotification,
输入参数:
msg_type="info",
title="数据日报 - " + 当前日期,
content=日报文本,
mention_all=是否有异常
)
# ----- 4. 如果有异常，单独发送告警（更醒目）-----
如果 是否有异常 == True:
调用子流程(Send_Alert_Only,
输入参数:
异常列表=异常列表,
mention_all=True # 异常告警必须@所有人
)
结束If
第五步：告警子流程（单独推送异常）
# ============================================
# 子流程：Send_Alert_Only
# 职责：只推送异常信息（更醒目，@所有人）
# ============================================
告警内容 = "🚨 **【异常告警】** " + 当前日期 + "\n\n"
按列表循环(异常列表, 循环项=异常):
告警内容 = 告警内容 + "• " + 异常["类型"] + "：" + 异常["说明"] + "\n"
结束循环
告警内容 = 告警内容 + "\n请尽快处理！"
# 发送飞书（@所有人）
调用子流程(Send_FeishuNotification,
输入参数:
msg_type="error",
title="【紧急】数据异常",
content=告警内容,
mention_all=True
)
# 发送企业微信（@所有人）
调用子流程(Send_WeworkNotification,
输入参数:
msg_type="error",
title="【紧急】数据异常",
content=告警内容,
mention_all=True
)
输出日志("异常告警已发送")
第六步：写入飞书多维表格（明细数据）
# ============================================
# 子流程：Write_Detail_To_Feishu
# 职责：将明细数据写入飞书多维表格
# ============================================
# 1. 获取Token
调用子流程(Get_Feishu_Token, 输出=token)
# 2. 组装数据（飞书API格式）
设置变量(记录列表 = [])
按列表循环(清洗后数据, 循环项=行):
记录 = {
"fields": {
"日期": 行[7], # 采集时间
"平台": 行[0],
"商品名": 行[1],
"销售额": 行[2],
"销量": 行[3],
"订单量": 行[4],
"退款": 行[5],
"客诉": 行[6]
}
}
将 记录 加入列表(记录列表)
结束循环
# 3. 批量写入（每批500条）
设置变量(批次大小 = 500)
设置变量(总条数 = 获取列表长度(记录列表))
按次数循环(总条数 / 批次大小 + 1):
起始 = 循环索引 * 批次大小
结束 = 最小值(起始 + 批次大小, 总条数)
批次数据 = 取列表子集(记录列表, 起始, 结束)
HTTP请求(
方法="POST",
URL="https://open.feishu.cn/open-apis/bitable/v1/apps/AppID/tables/TableID/records/batch_create",
请求头={
"Authorization": "Bearer " + token,
"Content-Type": "application/json"
},
请求体=JSON序列化({"records": 批次数据})
)
等待(0.5)
结束循环
输出日志("写入完成，共 " + 总条数 + " 条")
第七步：定时任务配置
影石实操路径： 客户端 → 定时任务 → 新建任务
| 配置项 | 设置值 |
|---|---|
| 任务名称 | 每日数据日报 |
| 选择流程 | Daily_Report_Main |
| 触发时间 | 每天 09:00 |
| 重复 | 每天 |
| 超时时间 | 30分钟 |
| 失败重试 | 1次 |
完整主流程代码
# ============================================
# 主流程：Daily_Report_Main
# ============================================
输出日志("===== 日报系统启动 =====")
开始时间 = 获取当前时间()
# ----- 1. 加载配置 -----
调用子流程(Load_Config)
# ----- 2. 采集数据 -----
输出日志("【1/5】采集数据")
调用子流程(Fetch_Taobao, 输出=淘宝数据)
调用子流程(Fetch_Pinduoduo, 输出=拼多多数据)
调用子流程(Fetch_Internal, 输出=内部数据)
# 合并
合并列表(淘宝数据, 拼多多数据, 内部数据, 存入列表=全部数据)
输出日志("采集完成，共 " + 获取列表长度(全部数据) + " 条")
# ----- 3. 数据汇总（Python）-----
输出日志("【2/5】数据汇总")
调用子流程(Summary_Data, 输入: 全部数据, 输出: 汇总结果)
# ----- 4. 生成日报内容（Python）-----
输出日志("【3/5】生成日报")
调用子流程(Generate_Report_Content, 输入: 汇总结果, 输出: 日报文本, 异常列表)
# ----- 5. 推送消息 -----
输出日志("【4/5】推送消息")
调用子流程(Send_Feishu_Report, 输入: 日报文本, 是否有异常)
调用子流程(Send_Wework_Report, 输入: 日报文本, 是否有异常)
如果 是否有异常 == True:
调用子流程(Send_Alert_Only, 输入: 异常列表)
结束If
# ----- 6. 写入多维表格 -----
输出日志("【5/5】写入明细")
调用子流程(Write_Detail_To_Feishu, 输入: 全部数据)
# ----- 7. 记录日志 -----
结束时间 = 获取当前时间()
总耗时 = 计算时间差(开始时间, 结束时间)
输出日志("日报系统完成，耗时 " + 总耗时 + " 秒")
输出日志("===== 日报系统结束 =====")
日报效果示例
飞书/企业微信收到的消息：
📊 2026-07-01 数据日报
━━━━━━━━━━━━━━━━━━━━━━━━━━━
💰 核心指标
• 总销售额：¥156,780.50
• 总销量：1,280 件
• 总订单量：320 单
• 退款率：2.3%
• 客诉数：5 件
📋 各平台明细
• 淘宝：销售额 ¥89,200.00，销量 580，订单 180
• 拼多多：销售额 ¥45,300.00，销量 450，订单 100
• 内部系统：销售额 ¥22,280.50，销量 250，订单 40
🏆 TOP5 热销商品
1. iPhone 15（淘宝）¥32,000.00，销量 32
2. 华为Mate 60（淘宝）¥28,000.00，销量 28
3. 蓝牙耳机 Pro（拼多多）¥12,800.00，销量 64
4. 充电宝 20000mAh（拼多多）¥8,500.00，销量 85
5. 数据线三合一（淘宝）¥6,400.00，销量 128
✅ 今日无异常
━━━━━━━━━━━━━━━━━━━━━━━━━━━
🕐 更新时间：2026-07-01 09:05
常见问题速查
| 问题 | 原因 | 解决方法 |
|---|---|---|
| 平台数据缺失 | 采集失败 | 检查采集子流程日志 |
| 汇总金额不对 | 数据格式不统一 | 检查清洗逻辑 |
| 飞书消息发不出 | Token过期 | 每次发送前刷新Token |
| 异常未告警 | 阈值设置太高 | 调整异常检测阈值 |
| 日报推送延迟 | 采集耗时太长 | 优化采集速度，提前触发 |
推荐资源
- 本系列第24篇《企业微信机器人通知封装》
- 本系列第21篇《飞书多维表格读写与消息通知》
- 本系列第38篇《从0到1搭建多平台数据日报系统》
- 本系列第40篇《数据采集之后的清洗、去重与格式转换》
#影刀RPA #RPA自动化 #日报系统 #飞书 #企业微信 #数据汇总 #异常告警
作者：林焱
本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。