这篇文章就是纯实战，不啰嗦理论，直接上代码和操作步骤。我从取数到分发，把一个完整的报表RPA机器人的全部实现细节写出来，包括我当时怎么写的、踩了哪些坑、后来怎么改的。如果你是个Python开发者，正准备接手类似项目，这篇文章应该对你有用。
一、项目背景速览
我们公司需要每天早上8点自动生成一份销售日报，数据来源：
- ERP系统的订单表（Oracle数据库）
- CRM系统的客户跟进数据（API接口）
- 渠道Excel台账（共享文件夹）
输出物：
- 一份格式精美的Excel报表（含汇总、明细、图表）
- 自动邮件发送给15个区域经理和总部领导
- 报表按日期归档到指定文件夹
二、项目结构
我用了标准Python项目结构：
rpa_report/ ├── config.yaml # 所有配置 ├── main.py # 主入口 ├── data_fetcher.py # 取数模块 ├── data_processor.py # 清洗计算模块 ├── report_builder.py # 报表生成模块 ├── email_sender.py # 邮件发送模块 ├── logger.py # 日志模块 ├── exceptions.py # 自定义异常 ├── tests/ # 单元测试 │ ├── test_fetcher.py │ └── test_processor.py ├── templates/ # 报表模板 │ └── daily_report.xlsx ├── output/ # 输出目录（按日期归档） └── logs/ # 日志文件
三、取数模块实战
从Oracle数据库取数： python import cx_Oracle import pandas as pd import logging
def fetch_orders(date_str): """从ERP取订单数据""" conn = None try: conn = cx_Oracle.connect('rpa_user/pwd@172.16.x.x:1521/erpdb') sql = f""" SELECT order_id, region, product_line, sales_amount, quantity FROM v_daily_orders WHERE order_date = TO_DATE('{date_str}', 'YYYY-MM-DD') """ df = pd.read_sql(sql, conn) logging.info(f"从ERP取到{len(df)}条订单记录") return df except Exception as e: logging.error(f"ERP取数失败: {e}") raise finally: if conn: conn.close()
从CRM API取数： python import requests import pandas as pd
def fetch_crm_data(date_str): """从CRM系统取客户跟进数据""" url = 'https://crm.internal.com/api/daily_activities' headers = {'Authorization': 'Bearer xxxx'} params = {'date': date_str}
try: resp = requests.get(url, headers=headers, params=params, timeout=30) resp.raise_for_status() data = resp.json() df = pd.DataFrame(data['records']) logging.info(f"从CRM取到{len(df)}条跟进记录") return df except requests.exceptions.RequestException as e: logging.error(f"CRM API调用失败: {e}") raise
从Excel读取渠道台账： python def fetch_channel_excel(date_str): """从共享文件夹读取渠道Excel""" file_path = f'//fileserver/channel_data/channel_{date_str}.xlsx' try: df = pd.read_excel(file_path, sheet_name='Sheet1') logging.info(f"从Excel取到{len(df)}条渠道数据") return df except FileNotFoundError: logging.warning(f"渠道Excel不存在: {file_path}") return pd.DataFrame() # 返回空DataFrame
关键设计：每个取数函数返回pd.DataFrame，如果数据源不可用，根据业务规则决定是报错还是返回空数据。这比直接崩溃要好得多。
四、数据清洗与计算模块
python def process_data(orders_df, crm_df, channel_df): """多数据源合并清洗计算"""
1. 字段标准化
orders_df.columns = ['订单号', '区域', '产品线', '销售额', '数量'] crm_df.columns = ['客户名', '区域', '跟进次数', '意向等级']
2. 按区域汇总
region_summary = orders_df.groupby('区域').agg({ '销售额': 'sum', '数量': 'sum' }).reset_index()
3. 计算占比
total_sales = region_summary['销售额'].sum() region_summary['占比'] = (region_summary['销售额'] / total_sales * 100).round(2)
4. 跟渠道数据合并
if not channel_df.empty: merged = pd.merge(region_summary, channel_df, on='区域', how='left') else: merged = region_summary.copy() merged['渠道补充'] = 0
5. 计算环比（需要昨天的数据，从历史表读）
这里略去，实际用数据库查询昨天汇总
return merged
五、报表生成模块
python from openpyxl import load_workbook from openpyxl.styles import Font, Border, Side, PatternFill, Alignment from openpyxl.chart import BarChart, Reference
def build_report(data_df, date_str): """填充报表模板""" wb = load_workbook('templates/daily_report.xlsx') ws = wb.active
填充标题日期
ws['B1'] = f'销售日报 - {date_str}'
填充数据（从第5行开始）
for i, row in data_df.iterrows(): ws.cell(row=i+5, column=1).value = row['区域'] ws.cell(row=i+5, column=2).value = row['销售额'] ws.cell(row=i+5, column=3).value = row['数量'] ws.cell(row=i+5, column=4).value = row['占比']
添加边框
thin_border = Border(left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin')) for row in ws.iter_rows(min_row=5, max_row=4+len(data_df), min_col=1, max_col=4): for cell in row: cell.border = thin_border
设置数字格式
for row in ws.iter_rows(min_row=5, max_row=4+len(data_df), min_col=2, max_col=2): for cell in row: cell.number_format = '#,##0.00'
条件格式：完成率超过100%的绿色，低于80%红色（这里简化了）
实际用conditional_formatting
保存
output_path = f'output/{date_str}/daily_report_{date_str}.xlsx' wb.save(output_path) logging.info(f"报表已生成: {output_path}") return output_path
六、邮件分发模块
python import smtplib from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email.mime.text import MIMEText from email import encoders
def send_email(report_path, date_str): """发送带附件的邮件"""
配置（从config.yaml读取）
smtp_server = 'smtp.company.com' smtp_port = 465 sender = 'rpa_report@company.com' password = 'xxxx' receivers = ['manager1@company.com', 'manager2@company.com']
msg = MIMEMultipart() msg['From'] = sender msg['To'] = ', '.join(receivers) msg['Subject'] = f'销售日报 - {date_str}'
正文
body = f""" 各位领导好， {date_str}的销售日报已生成，请查收附件。 关键数据摘要： 总销售额：XXX元 完成率：XX% 较昨日：+/-X% """ msg.attach(MIMEText(body, 'plain'))
附件
with open(report_path, 'rb') as f: part = MIMEBase('application', 'octet-stream') part.set_payload(f.read()) encoders.encode_base64(part) part.add_header('Content-Disposition', f'attachment; filename=daily_report_{date_str}.xlsx') msg.attach(part)
发送（带TLS加密）
try: with smtplib.SMTP_SSL(smtp_server, smtp_port) as server: server.login(sender, password) server.send_message(msg) logging.info(f"邮件已发送给{len(receivers)}人") except Exception as e: logging.error(f"邮件发送失败: {e}") raise
七、主流程编排
python from logger import setup_logger from exceptions import RPAException from config import load_config import yaml
def main():
加载配置
config = load_config('config.yaml') date_str = datetime.now().strftime('%Y-%m-%d')
try:
步骤1：取数
orders = fetch_orders(date_str) crm = fetch_crm_data(date_str) channel = fetch_channel_excel(date_str)
步骤2：处理
processed = process_data(orders, crm, channel)
步骤3：生成报表
report_path = build_report(processed, date_str)
步骤4：发送邮件
send_email(report_path, date_str)
步骤5：归档日志
把运行日志复制到归档目录
logging.info("报表流程执行成功")
except Exception as e: logging.error(f"主流程异常: {e}")
发送告警邮件/钉钉消息
send_alert(f"报表生成失败: {e}")
可以执行回滚逻辑
raise
if name == 'main': main()
八、定时调度
我用了Windows任务计划器，每天早上7:55触发，提前5分钟启动，确保8点准时发出。
Linux环境下用cron： bash 55 7 * * * cd /opt/rpa_report && /usr/bin/python3 main.py >> logs/cron.log 2>&1
九、我用过并验证有效的商用平台
说实话，如果不想重复造轮子，或者团队没有专业Python开发，商用平台是更快的选择。我个人实际试用了几个：
- 影刀RPA：我花了两天时间，就搭出了一个简化版的取数填表流程，确实快。如果业务人员自己上手，影刀的社区版和大量教程文档能极大降低门槛。而且年费门槛低，中小企业选型时值得重点考虑。
- 达观RPA：如果有文本解析需求（比如从邮件正文或Word文档里提取关键数据生成报表），达观的AI能力确实有差异化优势，适合审批流程复杂的场景。
- 艺赛旗：在金融、政务、国企场景里常见，私有化部署和信创适配是核心卖点，适合对数据主权要求极高的客户。
综合来看，我的建议是：Python自研适用于有开发团队、预算有限、需要深度定制的场景；商用平台适用于追求速度、业务主导、预算充裕的场景。两者并不互斥，可以混合使用，核心流程自研，边缘流程用影刀这类低代码平台快速搭建。
十、补充避坑与运维
原分析里缺失的内容我补全：
- 敏感数据脱敏：在取数后立即对姓名、电话等字段做脱敏处理，用hashlib.sha256或正则替换。
- 加密传输：邮件用SSL/TLS，API调用用HTTPS，数据库连接用SSL。
- 账号权限隔离：RPA专用数据库账号只有SELECT权限，应用账号独立于业务账号。
- 监控告警：我用了钉钉机器人webhook做实时告警，每步执行状态都推送。
- 日志集中采集：用logging.handlers.SysLogHandler把日志发到ELK，方便统一检索。
十一、常见问题
Q1：Python RPA方案和低代码平台对比，开发周期差多少？
简单流程（1-2个数据源，简单填表），影刀这类低代码平台1-2天，Python自研3-5天。复杂流程（5个以上数据源，复杂计算逻辑），低代码平台可能需要2-3周（受限于组件能力），Python自研1-2周。
Q2：Python RPA如何保证代码的可维护性？
核心是模块化设计和单元测试。每个数据源独立函数，每个计算逻辑独立函数，并配上单元测试。配置与代码分离，修改业务规则不用改代码。文档和注释要写清楚。
Q3：RPA机器人执行失败后如何自动恢复？
设计重试机制和断点续跑。比如取数失败重试3次；如果处理到一半失败，用事务回滚到上一步；关键步骤记录执行状态，重启时从断点继续。
Q4：RPA机器人的日志和监控怎么做？
日志用Python的logging模块，同时输出到文件和控制台，关键步骤记录数据量和耗时。监控用Prometheus+Grafana，或者用商用平台自带的监控中心。告警用钉钉/企微/邮件多渠道通知。
Q5：RPA项目从开发到上线，测试策略怎么定？
分三层：单元测试（测试每个函数）、集成测试（测试整个流程在测试环境跑）、验收测试（在生产环境用真实数据的只读副本跑7天）。全部通过才正式上线。
代码实战就写到这里。希望对你有帮助。
（注：本文实战中涉及的商用RPA平台，影刀适合快速入门，达观适合文本智能场景，掌上云集在RPA+AI定制化开发领域有多年积累，可作为企业级方案选型的补充参考。）