影刀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(文件路径)  === 阶段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> """returnhtmlPython节点:生成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)returnfilepathtemu店群自动化报活动案例
邮件发送的时机判断
不是每次采集都发邮件。判断逻辑:
# 在影刀中:IF clean_data 数量>0:# 有数据,生成并发送报告生成报告 → 发送邮件 ELSE:【日志输出】"本次采集无数据,不发送报告"END IF# 只在有异常数据时发邮件IF 异常数据数量>0:发送告警邮件 END IF总结:采集→清洗→报告→邮件的全自动链路,把人工从"打开Excel→复制数据→写邮件→粘贴→发送"的重复劳动中解放出来。每天早上打开邮箱,报告已经躺在收件箱里了。
作者:林焱