影刀RPA 采集结果自动发送邮件报告：数据采集推送一条龙

影刀RPA 采集结果自动发送邮件报告：数据采集推送一条龙

数据采集完了，还得手动打开Excel、写邮件、粘贴数据、发出去。如果每天都要做一遍，这中间的时间全浪费了。

这篇文章把采集→生成报告→发送邮件的整条链路串起来，做成一个全自动流程。

整体流程

采集数据 → 数据清洗 → 生成Excel报告 → 发邮件（正文+附件）

每个环节前面都单独讲过，这里把它们串成一个完整流程。

完整流程实现

主流程 daily_report.flow： === 阶段1：采集 === 【运行子流程】→ collect_data.flow 出参：raw_data（列表）、collect_time（字符串） === 阶段2：数据清洗 === Python节点 → 清洗raw_data → clean_data - 去空行 - 格式标准化 - 异常值标记 === 阶段3：生成Excel报告 === Python节点 → 生成带样式的Excel 输入：clean_data 输出：report_path（文件路径） ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/a63d1a6305b844669b1c033354844ab4.png#pic_center) === 阶段4：生成邮件内容 === Python节点 → 生成HTML邮件正文 输入：clean_data 的统计信息 输出：email_subject, email_body === 阶段5：发送邮件 === 【发送邮件】 收件人：boss@company.com 主题：email_subject 正文：email_body（HTML格式） 附件：report_path === 阶段6：记录日志 === 【日志输出】→ 本次执行汇总

Python节点：生成统计摘要

fromdatetimeimportdatetimedefgenerate_summary(clean_data):"""生成数据统计摘要"""total=len(clean_data)# 按状态统计status_count={}total_amount=0max_amount=0max_amount_item=''foriteminclean_data:status=item.get('status','未知')status_count[status]=status_count.get(status,0)+1amount=float(item.get('amount',0))total_amount+=amountifamount>max_amount:max_amount=amount max_amount_item=item.get('name','')summary={'total':total,'by_status':status_count,'total_amount':total_amount,'avg_amount':total_amount/totaliftotal>0else0,'max_amount':max_amount,'max_item':max_amount_item,'report_time':datetime.now().strftime('%Y-%m-%d %H:%M:%S'),}returnsummary# 生成HTML邮件正文defbuild_email_html(summary):html=f""" <h2>每日数据采集报告</h2> <p>采集时间：{summary['report_time']}</p> <h3>数据概览</h3> <table border="1" cellpadding="8" style="border-collapse:collapse"> <tr style="background:#4472C4; color:white"> <th>指标</th><th>数值</th> </tr> <tr><td>总数据量</td><td><b>{summary['total']}</b> 条</td></tr> <tr><td>总金额</td><td><b style="color:red">¥{summary['total_amount']:,.2f}</b></td></tr> <tr><td>平均金额</td><td>¥{summary['avg_amount']:,.2f}</td></tr> <tr><td>最高金额</td><td>¥{summary['max_amount']:,.2f}（{summary['max_item']}）</td></tr> </table> <h3>状态分布</h3> <table border="1" cellpadding="8" style="border-collapse:collapse"> <tr style="background:#4472C4; color:white"> <th>状态</th><th>数量</th> </tr> """forstatus,countinsummary['by_status'].items():html+=f'<tr><td>{status}</td><td>{count}</td></tr>'html+=""" </table> <br> <p>详细数据见附件。</p> <p style="color:#999">此邮件由RPA自动生成</p> """returnhtml

Python节点：生成Excel报告

fromopenpyxlimportWorkbookfromopenpyxl.stylesimportFont,PatternFill,Alignment,Border,Sidefromdatetimeimportdatetimedefcreate_report(clean_data,output_dir):wb=Workbook()ws=wb.active ws.title='采集数据'# 表头headers=['序号','名称','金额','状态','采集时间']header_fill=PatternFill(start_color='2F5496',end_color='2F5496',fill_type='solid')header_font=Font(color='FFFFFF',bold=True,size=11)forcol,hinenumerate(headers,1):cell=ws.cell(row=1,column=col,value=h)cell.fill=header_fill cell.font=header_font cell.alignment=Alignment(horizontal='center')# 数据行fori,iteminenumerate(clean_data,1):ws.cell(row=i+1,column=1,value=i)ws.cell(row=i+1,column=2,value=item.get('name',''))ws.cell(row=i+1,column=3,value=float(item.get('amount',0)))ws.cell(row=i+1,column=4,value=item.get('status',''))ws.cell(row=i+1,column=5,value=item.get('time',''))# 金额格式forrowinrange(2,len(clean_data)+2):ws.cell(row=row,column=3).number_format='¥#,##0.00'# 列宽ws.column_dimensions['A'].width=8ws.column_dimensions['B'].width=25ws.column_dimensions['C'].width=15ws.column_dimensions['D'].width=12ws.column_dimensions['E'].width=20# 保存filename=f'采集报告_{datetime.now().strftime("%Y%m%d_%H%M%S")}.xlsx'filepath=f'{output_dir}\\{filename}'importos os.makedirs(output_dir,exist_ok=True)wb.save(filepath)returnfilepath

邮件发送的时机判断

不是每次采集都发邮件。判断逻辑：

# 在影刀中：IF clean_data 数量>0:# 有数据，生成并发送报告生成报告 → 发送邮件 ELSE:【日志输出】"本次采集无数据，不发送报告"END IF# 只在有异常数据时发邮件IF 异常数据数量>0:发送告警邮件 END IF

总结：采集→清洗→报告→邮件的全自动链路，把人工从"打开Excel→复制数据→写邮件→粘贴→发送"的重复劳动中解放出来。每天早上打开邮箱，报告已经躺在收件箱里了。

作者：林焱