用Python解放双手：SAP发票自动化处理实战指南

每月底财务部门的同事总在重复相同的操作——打开SAP，输入MIR7/MIRO/MR8M事务码，逐个处理成百上千张发票的删除或冲销请求。这种机械式操作不仅效率低下，还容易因疲劳导致误操作。其实，只需几行Python代码，就能将这些重复劳动交给机器自动完成。

1. 环境准备与SAP连接配置

1.1 安装必要Python库

处理SAP自动化需要两个核心库： pyrfc 用于RFC通信， pandas 用于数据处理。建议使用conda创建独立环境：

conda create -n sap_auto python=3.8
conda activate sap_auto
pip install pyrfc pandas openpyxl

注意：pyrfc需要SAP NWRFC SDK支持，需提前从SAP官网下载对应版本

1.2 SAP连接参数配置

建立可靠的RFC连接是自动化基础。建议将连接参数保存在配置文件中：

# config.ini
[SAP_PROD]
ashost=sap.example.com
sysnr=00
client=100
user=automation_user
passwd=secure_password
lang=EN

连接测试代码：

from pyrfc import Connection

conn = Connection(config_section='SAP_PROD')
print(conn.get_system_info())  # 验证连接

2. 发票处理核心BAPI解析

2.1 MIR7预制发票删除

BAPI_INCOMINGINVOICE_DELETE 需要两个关键参数：

参数名	类型	说明
invoicedocnumber	CHAR(10)	发票凭证号
fiscalyear	CHAR(4)	会计年度

典型错误处理模式：

def delete_invoice(doc_number, fiscal_year):
    try:
        # 格式转换
        doc_number = str(doc_number).zfill(10)
        
        result = conn.call('BAPI_INCOMINGINVOICE_DELETE',
                          invoicedocnumber=doc_number,
                          fiscalyear=fiscal_year)
        
        if any(msg['TYPE'] == 'E' for msg in result['RETURN']):
            conn.call('BAPI_TRANSACTION_ROLLBACK')
            return False, [msg for msg in result['RETURN'] if msg['TYPE'] in ('E', 'A')]
        else:
            conn.call('BAPI_TRANSACTION_COMMIT', wait=True)
            return True, f"预制发票 {doc_number} 删除成功"
    except Exception as e:
        return False, str(e)

2.2 MIRO发票冲销

MR8M事务对应的 BAPI_INCOMINGINVOICE_CANCEL 更复杂：

def cancel_invoice(doc_number, fiscal_year, reason_code='03', post_date=None):
    params = {
        'invoicedocnumber': str(doc_number).zfill(10),
        'fiscalyear': fiscal_year,
        'reasonreversal': reason_code
    }
    
    if reason_code == '04' and post_date:
        params['postingdate'] = post_date.strftime('%Y%m%d')
    
    result = conn.call('BAPI_INCOMINGINVOICE_CANCEL', **params)
    
    if result.get('invoicedocnumber_reversal'):
        conn.call('BAPI_TRANSACTION_COMMIT', wait=True)
        return True, {
            'new_doc': result['invoicedocnumber_reversal'],
            'new_year': result['fiscalyear_reversal']
        }
    else:
        conn.call('BAPI_TRANSACTION_ROLLBACK')
        return False, [msg for msg in result['RETURN'] if msg['TYPE'] in ('E', 'A')]

3. 构建自动化处理流水线

3.1 从Excel读取任务清单

典型任务清单格式示例：

操作类型	凭证号	年度	冲销原因	过账日期
DELETE	5100001234	2023	-	-
CANCEL	5100005678	2023	03	-

处理脚本：

import pandas as pd

def process_batch(file_path):
    df = pd.read_excel(file_path)
    report = []
    
    for _, row in df.iterrows():
        if row['操作类型'] == 'DELETE':
            status, msg = delete_invoice(row['凭证号'], row['年度'])
        elif row['操作类型'] == 'CANCEL':
            status, msg = cancel_invoice(
                row['凭证号'], 
                row['年度'],
                row.get('冲销原因', '03'),
                row.get('过账日期')
            )
        
        report.append({
            '原始凭证': row['凭证号'],
            '状态': '成功' if status else '失败',
            '消息': msg,
            '新凭证': msg.get('new_doc') if isinstance(msg, dict) else None
        })
    
    return pd.DataFrame(report)

3.2 结果报告与异常处理

生成带格式的Excel报告：

def generate_report(result_df, output_file):
    writer = pd.ExcelWriter(output_file, engine='xlsxwriter')
    result_df.to_excel(writer, index=False)
    
    workbook = writer.book
    worksheet = writer.sheets['Sheet1']
    
    format_success = workbook.add_format({'bg_color': '#C6EFCE'})
    format_failure = workbook.add_format({'bg_color': '#FFC7CE'})
    
    for idx, status in enumerate(result_df['状态']):
        if status == '成功':
            worksheet.set_row(idx+1, None, format_success)
        else:
            worksheet.set_row(idx+1, None, format_failure)
    
    writer.close()

4. 高级应用与性能优化

4.1 并行处理加速

使用 concurrent.futures 实现并行处理：

from concurrent.futures import ThreadPoolExecutor

def parallel_process(file_path, max_workers=5):
    df = pd.read_excel(file_path)
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for _, row in df.iterrows():
            if row['操作类型'] == 'DELETE':
                futures.append(executor.submit(
                    delete_invoice, 
                    row['凭证号'], 
                    row['年度']
                ))
            # 其他操作类似...
        
        results = [f.result() for f in futures]
    
    return pd.DataFrame([{
        '原始凭证': df.iloc[idx]['凭证号'],
        '状态': '成功' if res[0] else '失败',
        '消息': res[1]
    } for idx, res in enumerate(results)])

4.2 自动重试机制

对暂时性错误实现智能重试：

from time import sleep

def robust_call(bapi_name, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            result = conn.call(bapi_name, **kwargs)
            if any(msg['TYPE'] == 'E' for msg in result.get('RETURN', [])):
                if attempt == max_retries - 1:
                    return False, result['RETURN']
                sleep(2 ** attempt)  # 指数退避
                continue
            return True, result
        except (ConnectionError, RFCError) as e:
            if attempt == max_retries - 1:
                return False, str(e)
            conn.reopen()  # 重建连接
            sleep(2 ** attempt)

5. 系统集成方案

5.1 与邮件系统联动

处理完成后自动发送结果邮件：

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

def send_report(email_to, report_file):
    msg = MIMEMultipart()
    msg['Subject'] = 'SAP发票处理报告'
    msg['From'] = 'sap_auto@example.com'
    msg['To'] = email_to
    
    with open(report_file, '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="{report_file}"')
        msg.attach(part)
    
    with smtplib.SMTP('smtp.example.com') as server:
        server.send_message(msg)

5.2 定时任务配置

使用Windows任务计划或Linux cron实现定时运行：

# 每天上午6点运行
0 6 * * * /path/to/python /scripts/sap_invoice_auto.py

对于更复杂的调度，可以集成到Airflow等平台：

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

default_args = {
    'start_date': datetime(2023, 1, 1),
}

dag = DAG('sap_invoice_processing', 
          schedule_interval='0 6 * * *',
          default_args=default_args)

run_script = PythonOperator(
    task_id='process_invoices',
    python_callable=main_processing_function,
    dag=dag
)