当前位置：首页 > news >正文

SAP全自动化工具开发：Excel自动上传与邮件通知系统

news 2025/7/27 0:59:05

SAP全自动化工具开发：Excel自动上传与邮件通知系统

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家，觉得好请收藏。点击跳转到网站。

1. 项目概述

在现代企业资源规划(ERP)系统中，SAP作为行业领先的解决方案，其自动化处理能力对企业运营效率至关重要。本文将详细介绍如何使用Python开发一个全自动化工具，实现Excel数据自动上传至SAP系统，并在操作完成后发送邮件通知的功能。

1.1 项目背景

许多企业日常运营中需要将大量数据从Excel导入SAP系统，传统的手动操作方式存在以下问题：

效率低下，重复性工作耗时
人为错误风险高
缺乏操作记录和通知机制
无法实现非工作时间的自动化处理

1.2 解决方案

本自动化工具将实现以下功能：

监控指定目录下的Excel文件
自动解析Excel数据
通过SAP GUI脚本或SAP RFC接口上传数据
处理完成后发送邮件通知
记录操作日志

1.3 技术栈选择

编程语言：Python 3.8+
SAP接口：SAP GUI Scripting或PyRFC
Excel处理：openpyxl/pandas
邮件通知：smtplib/email
日志管理：logging
调度工具：APScheduler/Windows Task Scheduler

2. 系统架构设计

2.1 整体架构

+-------------------+    +-------------------+    +-------------------+
|   Excel文件监控    | -> | SAP数据上传模块   | -> | 邮件通知模块      |
+-------------------+    +-------------------+    +-------------------+^                                                ||                                                v
+-------------------+                            +-------------------+
|   配置文件管理     | <--------------------------|    日志记录模块    |
+-------------------+                            +-------------------+

2.2 模块分解

主控制模块：协调各模块运行
文件监控模块：检测新Excel文件
Excel解析模块：读取并验证数据
SAP接口模块：与SAP系统交互
邮件通知模块：发送处理结果
日志记录模块：记录系统运行情况
配置管理模块：管理系统配置

2.3 数据流程

用户将Excel文件放入指定目录
系统检测到新文件并验证格式
解析Excel数据为SAP可接受格式
通过SAP接口上传数据
记录操作结果
发送邮件通知相关人员

3. 环境配置与准备工作

3.1 Python环境配置

# 建议使用虚拟环境
python -m venv sap_auto_env
source sap_auto_env/bin/activate  # Linux/Mac
sap_auto_env\Scripts\activate     # Windows# 安装依赖包
pip install pandas openpyxl pynput pywin32 pyrfc python-dotemail logging

3.2 SAP连接配置

3.2.1 SAP GUI Scripting方式

在SAP GUI中启用脚本功能：
- 事务码RZ11，设置参数sapgui/user_scripting为TRUE
- 重启SAP GUI
注册SAP GUI Scripting API：
- 运行命令：regsvr32 "C:\Program Files (x86)\SAP\FrontEnd\SAPgui\sapfewse.ocx"

3.2.2 SAP RFC方式

安装SAP NW RFC SDK

配置连接参数：

{"user": "your_username","passwd": "your_password","ashost": "sap_server_host","sysnr": "00","client": "100","lang": "EN"
}

3.3 邮件服务器配置

在config.ini中配置SMTP服务器信息：

[EMAIL]
server = smtp.office365.com
port = 587
username = your_email@company.com
password = your_password
sender = your_email@company.com
use_tls = yes

4. 核心代码实现

4.1 主控制模块

import os
import time
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from sap_uploader import SAPUploader
from email_sender import EmailSender
from config_manager import ConfigManagerclass SAPAutomationTool:def __init__(self):self.config = ConfigManager.load_config()self.setup_logging()self.sap_uploader = SAPUploader(self.config)self.email_sender = EmailSender(self.config)self.monitor_path = self.config.get('MONITORING', 'folder_path')def setup_logging(self):logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('sap_automation.log'),logging.StreamHandler()])self.logger = logging.getLogger(__name__)def run(self):self.logger.info("Starting SAP Automation Tool")event_handler = FileHandler(self)observer = Observer()observer.schedule(event_handler, path=self.monitor_path, recursive=False)observer.start()try:while True:time.sleep(1)except KeyboardInterrupt:observer.stop()observer.join()class FileHandler(FileSystemEventHandler):def __init__(self, parent):self.parent = parentself.processed_files = set()def on_created(self, event):if not event.is_directory and event.src_path.endswith('.xlsx'):file_path = event.src_pathif file_path not in self.processed_files:self.processed_files.add(file_path)self.parent.process_file(file_path)def process_file(self, file_path):try:self.logger.info(f"Processing new file: {file_path}")# Step 1: Validate Excel fileif not self.validate_excel(file_path):raise ValueError("Invalid Excel file format")# Step 2: Upload to SAPresult = self.sap_uploader.upload_to_sap(file_path)# Step 3: Send notificationsubject = "SAP Upload Result"body = f"File {os.path.basename(file_path)} processed with result: {result}"recipients = self.config.get('NOTIFICATION', 'recipients').split(',')self.email_sender.send_email(recipients, subject, body)self.logger.info(f"File processed successfully: {file_path}")except Exception as e:self.logger.error(f"Error processing file {file_path}: {str(e)}")# Send error notificationsubject = "SAP Upload Error"body = f"Error processing file {os.path.basename(file_path)}: {str(e)}"self.email_sender.send_email(['admin@company.com'], subject, body)

4.2 SAP上传模块（GUI Scripting方式）

import win32com.client
import time
import pythoncom
from typing import Dict, Listclass SAPGUIConnector:def __init__(self, config):self.config = configself.session = Noneself.connection = Nonedef connect(self):"""Establish connection to SAP GUI"""try:pythoncom.CoInitialize()sap_gui = win32com.client.GetObject("SAPGUI")if not sap_gui:raise Exception("SAP GUI not running")application = sap_gui.GetScriptingEngineif not application:raise Exception("Could not get scripting engine")self.connection = application.OpenConnection(self.config.get('SAP', 'connection_name'), True)self.session = self.connection.Children(0)self.session.findById("wnd[0]").maximize()return Trueexcept Exception as e:raise Exception(f"SAP connection failed: {str(e)}")def upload_excel_data(self, file_path: str, mapping: Dict) -> bool:"""Upload data from Excel to SAP using GUI scripting"""if not self.session:self.connect()try:# Step 1: Open transactionself.session.StartTransaction(self.config.get('SAP', 'transaction_code'))# Step 2: Navigate to upload screenself.session.findById("wnd[0]/tbar[0]/okcd").text = "/n" + self.config.get('SAP', 'transaction_code')self.session.findById("wnd[0]").sendVKey(0)# Step 3: Open Excel upload dialogself.session.findById("wnd[0]/mbar/menu[0]/menu[3]/menu[1]").select()# Step 4: Select Excel filefile_selection = self.session.findById("wnd[1]/usr/ctxtDY_PATH").text = os.path.dirname(file_path)self.session.findById("wnd[1]/usr/ctxtDY_FILENAME").text = os.path.basename(file_path)self.session.findById("wnd[1]/tbar[0]/btn[0]").press()# Step 5: Map Excel columns to SAP fieldsfor excel_col, sap_field in mapping.items():self.session.findById(f"wnd[1]/usr/tblSAPLSPO4TC_VIEW/txtMSGV1[0,{excel_col}]").text = sap_field# Step 6: Execute uploadself.session.findById("wnd[1]/tbar[0]/btn[0]").press()# Step 7: Verify upload successstatus = self.session.findById("wnd[0]/sbar").Textif "successfully" in status.lower():return Trueelse:raise Exception(f"Upload failed: {status}")except Exception as e:raise Exception(f"Error during SAP upload: {str(e)}")finally:self.close_session()def close_session(self):"""Close SAP session"""if self.session:self.session.findById("wnd[0]/tbar[0]/btn[15]").press()self.session = Noneif self.connection:self.connection.CloseSession()self.connection = None

4.3 SAP上传模块（RFC方式）

from pyrfc import Connection, ABAPApplicationError, ABAPRuntimeError, LogonError, CommunicationError
import pandas as pdclass SAPRFCConnector:def __init__(self, config):self.config = configself.conn = Nonedef connect(self):"""Establish RFC connection to SAP"""try:self.conn = Connection(ashost=self.config.get('SAP_RFC', 'ashost'),sysnr=self.config.get('SAP_RFC', 'sysnr'),client=self.config.get('SAP_RFC', 'client'),user=self.config.get('SAP_RFC', 'user'),passwd=self.config.get('SAP_RFC', 'passwd'),lang=self.config.get('SAP_RFC', 'lang'))return Trueexcept CommunicationError:raise Exception("Could not connect to SAP server")except LogonError:raise Exception("Could not logon to SAP")except (ABAPApplicationError, ABAPRuntimeError) as e:raise Exception(f"SAP error: {str(e)}")except Exception as e:raise Exception(f"RFC connection error: {str(e)}")def upload_via_rfc(self, file_path: str, bapi_name: str) -> dict:"""Upload data via RFC using BAPI"""if not self.conn:self.connect()try:# Read Excel datadf = pd.read_excel(file_path)data = df.to_dict('records')# Prepare BAPI parameterstable_data = {'DATA': data,'FIELDS': list(df.columns)}# Call BAPIresult = self.conn.call(bapi_name, **table_data)# Check return messageif result['RETURN']['TYPE'] == 'E':raise Exception(f"BAPI error: {result['RETURN']['MESSAGE']}")return {'success': True,'processed_rows': len(data),'message': result['RETURN']['MESSAGE']}except Exception as e:raise Exception(f"RFC upload failed: {str(e)}")finally:self.close_connection()def close_connection(self):"""Close RFC connection"""if self.conn:self.conn.close()self.conn = None

4.4 Excel处理模块

import pandas as pd
from openpyxl import load_workbook
from typing import Dict, List, Tupleclass ExcelProcessor:def __init__(self, config):self.config = configself.required_columns = self.config.get('EXCEL', 'required_columns').split(',')self.sheet_name = self.config.get('EXCEL', 'sheet_name', fallback='Sheet1')def validate_excel(self, file_path: str) -> bool:"""Validate Excel file structure"""try:# Check file existsif not os.path.exists(file_path):raise FileNotFoundError(f"File not found: {file_path}")# Check file extensionif not file_path.lower().endswith('.xlsx'):raise ValueError("Only .xlsx files are supported")# Check file not emptyif os.path.getsize(file_path) == 0:raise ValueError("Empty Excel file")# Check required sheets and columnswb = load_workbook(file_path)if self.sheet_name not in wb.sheetnames:raise ValueError(f"Sheet '{self.sheet_name}' not found")sheet = wb[self.sheet_name]headers = [cell.value for cell in sheet[1]]missing_columns = [col for col in self.required_columns if col not in headers]if missing_columns:raise ValueError(f"Missing required columns: {', '.join(missing_columns)}")return Trueexcept Exception as e:raise ValueError(f"Excel validation failed: {str(e)}")def read_excel_data(self, file_path: str) -> Tuple[List[Dict], List[str]]:"""Read data from Excel file"""try:df = pd.read_excel(file_path,sheet_name=self.sheet_name,header=0,dtype=str,na_values=['', 'NULL', 'NaN'])# Replace NaN with empty stringdf = df.fillna('')# Convert to list of dictionariesdata = df.to_dict('records')return data, list(df.columns)except Exception as e:raise Exception(f"Error reading Excel file: {str(e)}")def transform_data(self, data: List[Dict], mapping: Dict) -> List[Dict]:"""Transform Excel data to SAP format"""try:transformed = []for row in data:new_row = {}for excel_col, sap_field in mapping.items():if excel_col in row:new_row[sap_field] = row[excel_col]else:new_row[sap_field] = ''transformed.append(new_row)return transformedexcept Exception as e:raise Exception(f"Data transformation failed: {str(e)}")

4.5 邮件通知模块

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.utils import formatdate
import ssl
from typing import Listclass EmailSender:def __init__(self, config):self.config = configself.smtp_server = self.config.get('EMAIL', 'server')self.smtp_port = self.config.getint('EMAIL', 'port')self.username = self.config.get('EMAIL', 'username')self.password = self.config.get('EMAIL', 'password')self.sender = self.config.get('EMAIL', 'sender')self.use_tls = self.config.getboolean('EMAIL', 'use_tls')def send_email(self, recipients: List[str], subject: str, body: str, is_html=False):"""Send email notification"""try:# Create messagemsg = MIMEMultipart()msg['From'] = self.sendermsg['To'] = ', '.join(recipients)msg['Date'] = formatdate(localtime=True)msg['Subject'] = subject# Attach bodyif is_html:msg.attach(MIMEText(body, 'html'))else:msg.attach(MIMEText(body, 'plain'))# Connect to SMTP server and sendcontext = ssl.create_default_context()with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:if self.use_tls:server.starttls(context=context)server.login(self.username, self.password)server.sendmail(self.sender, recipients, msg.as_string())except Exception as e:raise Exception(f"Failed to send email: {str(e)}")def create_html_report(self, file_name: str, success: bool, details: str):"""Create HTML formatted email report"""status = "SUCCESS" if success else "FAILED"color = "green" if success else "red"html = f"""<html><head></head><body><h2>SAP Automation Tool - Processing Result</h2><table border="1" cellpadding="5"><tr><th>File Name</th><td>{file_name}</td></tr><tr><th>Status</th><td style="color:{color};font-weight:bold">{status}</td></tr><tr><th>Details</th><td>{details}</td></tr></table><p>This is an automated message. Please do not reply.</p></body></html>"""return html

4.6 配置管理模块

import configparser
import os
from typing import Dictclass ConfigManager:DEFAULT_CONFIG = {'MONITORING': {'folder_path': './input','processed_folder': './processed','error_folder': './error'},'SAP': {'connection_name': 'SAP_PROD','transaction_code': 'ME21N','method': 'GUI'  # or 'RFC'},'SAP_RFC': {'ashost': 'sap.company.com','sysnr': '00','client': '100','user': 'RFC_USER','passwd': 'password','lang': 'EN'},'EXCEL': {'sheet_name': 'Data','required_columns': 'Material,Quantity,Plant'},'EMAIL': {'server': 'smtp.office365.com','port': '587','username': 'automation@company.com','password': 'email_password','sender': 'automation@company.com','use_tls': 'yes'},'NOTIFICATION': {'recipients': 'user1@company.com,user2@company.com','admin_recipients': 'admin@company.com'},'MAPPING': {'Material': 'MATNR','Quantity': 'MENGE','Plant': 'WERKS'}}@staticmethoddef load_config(config_path='config.ini') -> configparser.ConfigParser:"""Load configuration from file or create default"""config = configparser.ConfigParser()if os.path.exists(config_path):config.read(config_path)else:# Create default config filefor section, options in ConfigManager.DEFAULT_CONFIG.items():config.add_section(section)for key, value in options.items():config.set(section, key, value)with open(config_path, 'w') as f:config.write(f)return config@staticmethoddef validate_config(config: configparser.ConfigParser) -> Dict:"""Validate configuration and return errors if any"""errors = {}# Check required sectionsrequired_sections = ['MONITORING', 'SAP', 'EMAIL']for section in required_sections:if not config.has_section(section):errors[section] = f"Missing section: {section}"# Check monitoring folder existsif config.has_option('MONITORING', 'folder_path'):folder = config.get('MONITORING', 'folder_path')if not os.path.exists(folder):errors['MONITORING'] = f"Folder does not exist: {folder}"# Check SAP connection methodif config.has_option('SAP', 'method'):method = config.get('SAP', 'method')if method not in ['GUI', 'RFC']:errors['SAP'] = "Invalid method, must be 'GUI' or 'RFC'"return errors

5. 系统集成与测试

5.1 集成步骤

将各模块组合到主程序中
配置config.ini文件
设置监控文件夹
测试SAP连接
测试邮件发送功能

5.2 测试用例设计

5.2.1 单元测试

import unittest
import os
import shutil
from excel_processor import ExcelProcessor
from email_sender import EmailSender
from config_manager import ConfigManagerclass TestExcelProcessor(unittest.TestCase):@classmethoddef setUpClass(cls):cls.config = ConfigManager.load_config()cls.processor = ExcelProcessor(cls.config)os.makedirs('test_data', exist_ok=True)def test_validate_excel(self):# Create test Excel filetest_file = 'test_data/valid.xlsx'df = pd.DataFrame({'Material': ['M001', 'M002'],'Quantity': [10, 20],'Plant': ['P100', 'P200']})df.to_excel(test_file, index=False)# Test validationself.assertTrue(self.processor.validate_excel(test_file))def test_invalid_excel(self):# Test missing columntest_file = 'test_data/invalid.xlsx'df = pd.DataFrame({'Material': ['M001', 'M002'],'Qty': [10, 20]  # Wrong column name})df.to_excel(test_file, index=False)with self.assertRaises(ValueError):self.processor.validate_excel(test_file)class TestEmailSender(unittest.TestCase):def setUp(self):self.config = ConfigManager.load_config()self.sender = EmailSender(self.config)def test_html_report(self):html = self.sender.create_html_report("test.xlsx", True, "All records processed")self.assertIn("SUCCESS", html)self.assertIn("test.xlsx", html)if __name__ == '__main__':unittest.main()

5.2.2 集成测试

import unittest
import time
from sap_automation_tool import SAPAutomationToolclass TestSAPAutomation(unittest.TestCase):@classmethoddef setUpClass(cls):cls.tool = SAPAutomationTool()def test_file_processing(self):# Create test file in monitored foldertest_file = os.path.join(self.tool.monitor_path, 'test_integration.xlsx')df = pd.DataFrame({'Material': ['TEST001', 'TEST002'],'Quantity': [1, 2],'Plant': ['TEST']})df.to_excel(test_file, index=False)# Wait for processingtime.sleep(10)# Check if file was moved to processed folderprocessed_folder = self.tool.config.get('MONITORING', 'processed_folder')processed_file = os.path.join(processed_folder,os.path.basename(test_file))self.assertTrue(os.path.exists(processed_file))

5.3 性能测试

import timeit
import pandas as pddef performance_test():# Test Excel reading performancesetup = '''
import pandas as pd
import os
from excel_processor import ExcelProcessor
config = ConfigManager.load_config()
processor = ExcelProcessor(config)
file_path = 'large_file.xlsx'# Create large test file
if not os.path.exists(file_path):df = pd.DataFrame({'Material': ['M' + str(i).zfill(5) for i in range(10000)],'Quantity': [i % 100 for i in range(10000)],'Plant': ['P' + str(i % 10).zfill(3) for i in range(10000)]})df.to_excel(file_path, index=False)'''stmt = '''
processor.validate_excel(file_path)
data, headers = processor.read_excel_data(file_path)'''time = timeit.timeit(stmt, setup, number=10)print(f"Average processing time: {time/10:.2f} seconds")if __name__ == '__main__':performance_test()

6. 部署与维护

6.1 部署方案

6.1.1 Windows服务部署

使用pyinstaller打包为exe:

pyinstaller --onefile --windowed sap_automation_tool.py

创建Windows服务:

import win32serviceutil
import win32service
import win32event
import servicemanagerclass SAPAutomationService(win32serviceutil.ServiceFramework):_svc_name_ = "SAPAutoTool"_svc_display_name_ = "SAP Automation Tool"def __init__(self, args):win32serviceutil.ServiceFramework.__init__(self, args)self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)self.tool = SAPAutomationTool()def SvcStop(self):self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)win32event.SetEvent(self.hWaitStop)def SvcDoRun(self):servicemanager.LogMsg(servicemanager.EVENTLOG_INFORMATION_TYPE,servicemanager.PYS_SERVICE_STARTED,(self._svc_name_, ''))self.tool.run()if __name__ == '__main__':win32serviceutil.HandleCommandLine(SAPAutomationService)

6.1.2 Linux系统部署

创建systemd服务文件 /etc/systemd/system/sapautotool.service:

[Unit]
Description=SAP Automation Tool
After=network.target[Service]
Type=simple
User=automation
WorkingDirectory=/opt/sap_auto
ExecStart=/opt/sap_auto/venv/bin/python /opt/sap_auto/sap_automation_tool.py
Restart=on-failure[Install]
WantedBy=multi-user.target

启用并启动服务:

sudo systemctl daemon-reload
sudo systemctl enable sapautotool
sudo systemctl start sapautotool

6.2 维护计划

日常维护:
- 监控日志文件大小，定期归档
- 检查处理文件夹中的残留文件
- 验证SAP连接可用性
定期检查:
- 每月检查Python依赖包更新
- 每季度验证Excel模板变更
- 每年审查SAP接口变更
备份策略:
- 每日备份配置文件
- 每周备份处理记录数据库
- 每月完整备份系统

6.3 故障排除指南

6.3.1 常见问题

SAP连接失败:
- 检查SAP服务器状态
- 验证登录凭据
- 确认网络连接
Excel处理错误:
- 检查文件是否被其他程序锁定
- 验证Excel文件格式
- 确认列名匹配配置
邮件发送失败:
- 检查SMTP服务器设置
- 验证发件人权限
- 检查网络防火墙设置

6.3.2 日志分析

日志文件示例：

2023-05-15 09:30:15,123 - sap_automation_tool - INFO - Starting SAP Automation Tool
2023-05-15 09:31:22,456 - sap_automation_tool - INFO - Processing new file: ./input/orders_20230515.xlsx
2023-05-15 09:32:45,789 - sap_automation_tool - ERROR - SAP connection failed: [Errno 11001] getaddrinfo failed
2023-05-15 09:33:10,111 - sap_automation_tool - INFO - File processed successfully: ./input/inventory_update.xlsx

7. 安全考虑

7.1 安全最佳实践

凭据管理:
- 不要将密码硬编码在脚本中
- 使用加密的配置文件或密钥库
- 考虑使用Windows凭据管理器或Linux keyring
数据保护:
- 处理完成后安全删除临时文件
- 加密敏感数据的日志记录
- 限制对处理文件夹的访问权限
网络安全:
- 使用VPN连接SAP服务器
- 启用SAP通信的SSL加密
- 限制SMTP服务器的IP访问

7.2 安全增强实现

7.2.1 加密配置文件

from cryptography.fernet import Fernet
import base64class ConfigEncryptor:def __init__(self, key_path='config.key'):self.key_path = key_pathself.key = self.load_or_create_key()def load_or_create_key(self):if os.path.exists(self.key_path):with open(self.key_path, 'rb') as f:return f.read()else:key = Fernet.generate_key()with open(self.key_path, 'wb') as f:f.write(key)return keydef encrypt_config(self, config_path, encrypted_path):cipher = Fernet(self.key)with open(config_path, 'r') as f:config_data = f.read().encode()encrypted_data = cipher.encrypt(config_data)with open(encrypted_path, 'wb') as f:f.write(encrypted_data)def decrypt_config(self, encrypted_path):cipher = Fernet(self.key)with open(encrypted_path, 'rb') as f:encrypted_data = f.read()return cipher.decrypt(encrypted_data).decode()

7.2.2 安全日志记录

import logging
from logging.handlers import RotatingFileHandler
import hashlibclass SecureLogger:def __init__(self, name, log_file, max_size=10*1024*1024, backup_count=5):self.logger = logging.getLogger(name)self.logger.setLevel(logging.INFO)# Create file handler with rotationhandler = RotatingFileHandler(log_file,maxBytes=max_size,backupCount=backup_count)# Create formatter with hashed sensitive dataformatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')handler.setFormatter(formatter)self.logger.addHandler(handler)def secure_log(self, level, message, sensitive_fields=None):if sensitive_fields:for field in sensitive_fields:if field in message:hashed = hashlib.sha256(message[field].encode()).hexdigest()[:8]message = message.replace(field, f"{field[:2]}...{hashed}")if level == 'info':self.logger.info(message)elif level == 'error':self.logger.error(message)elif level == 'warning':self.logger.warning(message)

8. 扩展与优化

8.1 功能扩展

8.1.1 数据库集成

import sqlite3
from contextlib import contextmanagerclass DatabaseManager:def __init__(self, db_path='sap_auto.db'):self.db_path = db_pathself.init_db()@contextmanagerdef get_cursor(self):conn = sqlite3.connect(self.db_path)cursor = conn.cursor()try:yield cursorconn.commit()finally:conn.close()def init_db(self):with self.get_cursor() as c:c.execute('''CREATE TABLE IF NOT EXISTS processed_files (id INTEGER PRIMARY KEY AUTOINCREMENT,file_name TEXT NOT NULL,processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,status TEXT CHECK(status IN ('SUCCESS', 'FAILED')),records_processed INTEGER,error_message TEXT)''')def log_processing(self, file_name, status, records=None, error=None):with self.get_cursor() as c:c.execute('''INSERT INTO processed_files (file_name, status, records_processed, error_message)VALUES (?, ?, ?, ?)''', (file_name, status, records, error))

8.1.2 支持多种文件格式

import pandas as pd
from abc import ABC, abstractmethodclass FileProcessor(ABC):@abstractmethoddef read_data(self, file_path):pass@abstractmethoddef validate(self, file_path):passclass ExcelProcessor(FileProcessor):def read_data(self, file_path):return pd.read_excel(file_path)def validate(self, file_path):# Excel specific validationpassclass CSVProcessor(FileProcessor):def read_data(self, file_path):return pd.read_csv(file_path)def validate(self, file_path):# CSV specific validationpassclass FileProcessorFactory:@staticmethoddef get_processor(file_path):if file_path.endswith('.xlsx'):return ExcelProcessor()elif file_path.endswith('.csv'):return CSVProcessor()else:raise ValueError("Unsupported file format")

8.2 性能优化

8.2.1 多线程处理

from concurrent.futures import ThreadPoolExecutor
import queueclass ParallelProcessor:def __init__(self, max_workers=4):self.executor = ThreadPoolExecutor(max_workers=max_workers)self.task_queue = queue.Queue()def add_task(self, file_path):future = self.executor.submit(self.process_file, file_path)future.add_done_callback(self.task_done)self.task_queue.put(future)def process_file(self, file_path):# Actual file processing logicpassdef task_done(self, future):try:result = future.result()# Handle resultexcept Exception as e:# Handle errorpassdef wait_completion(self):while not self.task_queue.empty():future = self.task_queue.get()future.result()

8.2.2 批量处理优化

class BatchProcessor:def __init__(self, batch_size=100):self.batch_size = batch_sizedef process_batch(self, data):"""Process data in batches to reduce SAP calls"""results = []for i in range(0, len(data), self.batch_size):batch = data[i:i + self.batch_size]try:result = self._process_sap_batch(batch)results.extend(result)except Exception as e:# Mark entire batch as failedresults.extend([{'status': 'failed','error': str(e)}] * len(batch))return resultsdef _process_sap_batch(self, batch):# Actual batch processing logicpass

9. 项目文档

9.1 用户手册

9.1.1 安装指南

系统要求:
- Windows 10+/Linux
- Python 3.8+
- SAP GUI 7.50+ (GUI模式)
- SAP NW RFC SDK (RFC模式)

安装步骤:

# 1. 克隆仓库
git clone https://company.com/sap-automation-tool.git
cd sap-automation-tool# 2. 创建虚拟环境
python -m venv venv
venv\Scripts\activate  # Windows
source venv/bin/activate  # Linux/Mac# 3. 安装依赖
pip install -r requirements.txt# 4. 配置SAP连接
# 编辑config.ini文件

9.1.2 使用说明

准备Excel文件:
- 确保包含所有必需列
- 数据放在第一个工作表
- 不要使用合并单元格
运行工具:
```
python sap_automation_tool.py
```
监控处理:
- 查看sap_automation.log了解处理状态
- 成功处理的文件移动到processed文件夹
- 失败的文件移动到error文件夹

9.2 技术文档

9.2.1 架构图

+-------------------+    +-------------------+    +-------------------+
|   File Monitor    | -> |  Excel Processor  | -> |  SAP Connector    |
+-------------------+    +-------------------+    +-------------------+^                      |                        ||                      v                        v
+-------------------+    +-------------------+    +-------------------+
|  Config Manager   | <- |  Error Handler    | <- |  Email Notifier   |
+-------------------+    +-------------------+    +-------------------+