整体脚本结构设计
Python自动化处理业务报表并上传系统的脚本通常可以划分为四个核心模块，分别是报表读取模块、数据处理模块、系统上传模块和异常处理模块，各模块职责清晰，便于后续维护和扩展。
1. 报表读取模块
该模块负责读取本地存储的业务报表文件，支持常见的Excel、CSV格式，根据文件类型选择对应的读取方式。
import pandas as pd
import os
def read_report(file_path):
# 获取文件后缀判断文件类型
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext == '.xlsx' or file_ext == '.xls':
# 读取Excel格式报表，默认读取第一个工作表
df = pd.read_excel(file_path, sheet_name=0)
elif file_ext == '.csv':
# 读取CSV格式报表，指定编码避免乱码
df = pd.read_csv(file_path, encoding='utf-8')
else:
raise ValueError("不支持的报表文件格式，仅支持xlsx、xls、csv格式")
return df
2. 数据处理模块
该模块对读取到的原始报表数据进行清洗、格式转换，使其符合目标系统的上传要求，比如统一字段名称、处理空值、转换数据类型等。
def process_data(raw_df, required_columns):
# 检查必要字段是否存在
missing_cols = [col for col in required_columns if col not in raw_df.columns]
if missing_cols:
raise ValueError(f"报表缺少必要字段：{missing_cols}")
# 提取需要的字段
processed_df = raw_df[required_columns].copy()
# 处理空值，将空值替换为默认值
processed_df = processed_df.fillna('')
# 转换日期字段格式，假设报表中有date字段
if 'date' in processed_df.columns:
processed_df['date'] = pd.to_datetime(processed_df['date']).dt.strftime('%Y-%m-%d')
return processed_df
3. 系统上传模块
该模块负责将处理后的数据按照系统要求的接口格式进行封装，调用系统提供的上传接口完成数据提交，这里以常见的HTTP接口为例。
import requests
import json
def upload_to_system(processed_df, upload_url, auth_token):
# 将DataFrame转换为系统要求的JSON格式
data_list = processed_df.to_dict(orient='records')
upload_payload = {
"report_data": data_list,
"report_type": "business_report"
}
# 设置请求头，包含认证信息
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {auth_token}"
}
# 发送POST请求上传数据
response = requests.post(upload_url, data=json.dumps(upload_payload), headers=headers, timeout=30)
# 检查响应状态
if response.status_code == 200:
result = response.json()
if result.get('code') == 0:
return True, "上传成功"
else:
return False, result.get('msg', '上传失败，系统返回错误')
else:
return False, f"请求失败，状态码：{response.status_code}"
4. 异常处理模块
该模块负责捕获脚本运行过程中出现的各类异常，记录错误日志，避免脚本直接崩溃，同时支持失败重试逻辑。
import logging
import time
# 配置日志
logging.basicConfig(
filename='report_auto_upload.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def run_with_retry(file_path, upload_url, auth_token, required_columns, max_retry=3):
retry_count = 0
while retry_count < max_retry:
try:
# 读取报表
raw_df = read_report(file_path)
logging.info(f"成功读取报表：{file_path}，共{len(raw_df)}条数据")
# 处理数据
processed_df = process_data(raw_df, required_columns)
logging.info("报表数据处理完成")
# 上传系统
success, msg = upload_to_system(processed_df, upload_url, auth_token)
if success:
logging.info("报表上传系统成功")
return True
else:
logging.error(f"上传失败：{msg}")
retry_count += 1
time.sleep(2)
except Exception as e:
logging.error(f"脚本运行异常：{str(e)}")
retry_count += 1
time.sleep(2)
logging.error(f"达到最大重试次数{max_retry}，任务执行失败")
return False
完整调用示例
将各个模块组合起来，传入对应的参数即可运行整个自动化流程，以下是完整的调用示例。
if __name__ == "__main__":
# 配置参数
report_file = "业务报表_202405.xlsx"
upload_api = "http://192.168.0.1:8080/api/upload_report"
token = "your_auth_token_here"
# 系统要求的必要字段列表
need_columns = ["date", "product_name", "sales_count", "sales_amount"]
# 执行自动化任务
run_with_retry(report_file, upload_api, token, need_columns)
注意事项
- 运行脚本前需要安装对应的依赖库，包括pandas、requests、openpyxl，可通过
pip install pandas requests openpyxl命令安装 - 上传接口的地址和认证token需要根据实际系统的要求进行调整，避免接口调用失败
- 如果报表格式有特殊要求，需要在数据处理模块中补充对应的处理逻辑，比如合并单元格、跨表关联等场景
- 日志文件会记录所有运行过程的信息，方便后续排查问题，建议定期清理日志文件避免占用过多存储空间