Python自动录入ERP系统数据

'# Python自动录入ERP系统数据

一、背景与问题

在制造业、零售业等企业信息化场景中，ERP（Enterprise Resource Planning）系统是核心业务管理系统。日常运营中，常需要将外部系统（如MES、CRM、财务系统）的数据同步至ERP系统，或通过自动化手段完成数据录入工作。传统方式需要人工输入，存在效率低、错误率高、数据不一致等问题。

本篇将深入探讨使用Python实现ERP系统数据自动录入的技术方案，包括：

• ERP系统接口对接原理
• 常见数据源类型（API/数据库/文件）
• 安全性与性能优化
• 实际工程实践中的关键问题

二、基本原理

ERP系统数据录入的核心是与系统接口的通信，主要包括以下技术栈：

1. 协议层：HTTP/HTTPS（REST API）、数据库协议（ODBC/JDBC）、文件传输协议（FTP/SFTP）
2. 数据处理：数据清洗、格式转换、校验规则
3. 安全机制：OAuth2、API Key、数据库连接字符串加密
4. 错误处理：重试机制、日志记录、异常捕获

以REST API为例，其通信流程如下：

[数据源] -> [Python程序] -> [HTTP请求] -> [ERP系统API] -> [数据入库]

三、环境准备

1. Python环境：建议使用Python 3.9+，安装以下库：

   pip install requests pyodbc pandas openpyxl

2. ERP系统对接信息：

   • API地址（如：https://erp.example.com/api/v1/inventory）
   • 接口文档（包含请求方法、参数、认证方式）
   • 数据库连接参数（如：DRIVER={SQL Server};SERVER=erpdb;DATABASE=erp;UID=sa;PWD=secret）

3. 开发工具：

   • Postman（API调试）
   • MySQL Workbench（数据库验证）
   • VSCode（代码编辑）

四、核心实现

1. REST API数据录入（示例）

import requests
import json
import time

def batch_upload_inventory(items):
    """
    批量上传库存数据到ERP系统
    items: list of dict, 每个元素格式：
    {
        'item_code': 'ITEM001',
        'quantity': 100,
        'warehouse': 'WH01',
        'last_update': datetime.now().isoformat()
    }
    """
    # 配置参数
    url = "https://erp.example.com/api/v1/inventory"
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
    }
    batch_size = 100  # 每批次处理100条数据
    
    # 分批处理
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        payload = {
            "inventory_records": json.dumps(batch)
        }
        
        try:
            response = requests.post(url, headers=headers, json=payload, timeout=10)
            response.raise_for_status()
            
            # 记录成功条数
            success_count = len(batch)
            print(f"成功上传{success_count}条数据")
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            # 记录错误日志并重试
            if "429" in str(e):  # 超过速率限制
                print("触发速率限制，等待10秒后重试")
                time.sleep(10)
                continue
            else:
                raise

    return True

关键代码解释：

• batch_size控制批量处理大小，避免单次请求过大导致超时
• timeout=10设置请求超时时间，防止程序卡死
• 429错误处理：针对HTTP 429（Too Many Requests）进行重试
• json.dumps()将Python对象转换为JSON格式

2. 数据库直连录入（示例）

import pyodbc
import pandas as pd

def sync_inventory_from_excel(file_path):
    """
    从Excel文件同步库存数据到ERP数据库
    """
    # 读取Excel文件
    df = pd.read_excel(file_path, engine='openpyxl')
    
    # 数据预处理
    df['last_update'] = pd.to_datetime(df['last_update'])
    df['quantity'] = pd.to_numeric(df['quantity'])
    
    # 建立数据库连接
    conn = pyodbc.connect(
        "DRIVER={SQL Server};SERVER=erpdb;DATABASE=erp;UID=sa;PWD=secret"
    )
    
    # 创建SQL语句
    insert_sql = """
        INSERT INTO inventory (item_code, quantity, warehouse, last_update)
        VALUES (?, ?, ?, ?)
        ON DUPLICATE KEY UPDATE
        quantity = VALUES(quantity),
        last_update = VALUES(last_update)
    """
    
    # 批量插入
    cursor = conn.cursor()
    for _, row in df.iterrows():
        cursor.execute(insert_sql, (
            row['item_code'], 
            row['quantity'], 
            row['warehouse'], 
            row['last_update']
        ))
    
    conn.commit()
    cursor.close()
    conn.close()

关键代码解释：

• 使用ON DUPLICATE KEY UPDATE实现UPSERT操作
• pyodbc支持多种数据库，需配置对应驱动
• 使用pandas进行数据预处理和批量操作
• 数据类型转换确保数据一致性

3. 文件传输录入（示例）

import os
import paramiko
from datetime import datetime

def upload_file_to_erp(file_path, server_ip, server_path):
    """
    通过SFTP上传文件到ERP服务器
    """
    # 创建SSH客户端
    transport = paramiko.Transport((server_ip, 22))
    transport.connect(username='admin', password='secure_password')
    
    # 创建SFTP会话
    sftp = paramiko.SFTPClient.from_transport(transport)
    
    # 生成文件名（带时间戳）
    file_name = f"inventory_{datetime.now().strftime('%Y%m%d')}.csv"
    remote_path = os.path.join(server_path, file_name)
    
    # 上传文件
    sftp.put(file_path, remote_path)
    print(f"文件上传至：{remote_path}")
    
    # 关闭连接
    sftp.close()
    transport.close()

关键代码解释：

• 使用paramiko实现SFTP协议通信
• 生成带时间戳的文件名避免覆盖
• 需要配置服务器IP、端口、认证信息
• 文件传输前建议进行完整性校验

五、完整案例

场景：库存数据同步

假设ERP系统有以下需求：

1. 每日从MES系统获取库存数据
2. 将数据通过API同步至ERP系统
3. 遇到网络问题时进行重试
4. 记录操作日志

完整代码如下：

import requests
import json
import time
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO)

# ERP系统配置
ERP_API_URL = "https://erp.example.com/api/v1/inventory"
API_HEADERS = {
    "Content-Type": "application/json",
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

def get_inventory_data_from_mes():
    """
    模拟从MES系统获取库存数据
    返回格式：list of dict
    """
    return [
        {'item_code': 'ITEM001', 'quantity': 150, 'warehouse': 'WH01', 'last_update': datetime.now().isoformat()},
        {'item_code': 'ITEM002', 'quantity': 200, 'warehouse': 'WH02', 'last_update': datetime.now().isoformat()}
    ]

def batch_upload_inventory(items):
    """
    批量上传库存数据到ERP系统
    """
    try:
        response = requests.post(
            ERP_API_URL,
            headers=API_HEADERS,
            json={'inventory_records': json.dumps(items)},
            timeout=10
        )
        response.raise_for_status()
        return True
    except requests.exceptions.RequestException as e:
        logging.error(f"上传失败: {str(e)}")
        if "429" in str(e):
            logging.warning("触发速率限制，等待10秒后重试")
            time.sleep(10)
            return batch_upload_inventory(items)
        else:
            raise

def main():
    try:
        # 获取数据
        inventory_data = get_inventory_data_from_mes()
        
        # 上传数据
        if batch_upload_inventory(inventory_data):
            logging.info("数据同步成功")
        else:
            logging.error("数据同步失败")
    
    except Exception as e:
        logging.error(f"发生异常: {str(e)}")

if __name__ == "__main__":
    main()

六、源码解析

1. 异常处理机制

• 使用try...except捕获网络请求异常
• 针对特定错误码（如429）进行重试处理
• 记录日志信息用于问题排查

2. 批量处理策略

• 分批次发送数据，避免单次请求过大
• 使用json.dumps()确保数据格式正确
• 设置超时时间防止程序卡死

3. 日志系统

• 使用logging模块记录不同级别的日志
• 便于后续监控和问题追溯

七、进阶使用

1. 异步处理

使用concurrent.futures实现异步处理：

from concurrent.futures import ThreadPoolExecutor

def async_upload(items):
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(batch_upload_inventory, [items]))
    return all(results)

2. 安全增强

• 使用requests.Session()保持会话
• 实现OAuth2认证流程
• 使用http.client.HTTPSConnection进行SSL验证

3. 性能优化

• 使用grequests库实现异步HTTP请求
• 对数据库操作添加索引
• 对文件传输使用压缩算法

八、性能与工程实践

1. 性能优化方案

2. 异常处理规范

• 网络异常：重试机制+断路器模式
• 数据异常：校验规则+数据回滚
• 系统异常：熔断机制+降级处理

3. 安全实践

• API密钥应使用加密存储（如Vault）
• 数据传输必须加密（HTTPS/SFTP）
• 敏感信息应使用环境变量存储
• 设置严格的访问控制（RBAC）

九、常见问题与踩坑

1. 常见错误及解决

2. 常见陷阱

• 忽略数据校验导致数据污染
• 忽视错误日志记录导致问题排查困难
• 忽略安全配置导致数据泄露
• 忽视接口版本差异导致兼容性问题

十、最佳实践

1. 接口对接规范

• 严格遵循API文档
• 使用Swagger进行接口管理
• 建立接口测试用例库

2. 数据处理规范

• 建立数据校验规则库
• 使用数据转换模板
• 建立数据清洗流程

3. 安全实践规范

• 使用OAuth2进行认证
• 实施最小权限原则
• 定期更新密钥
• 使用加密传输

十一、总结

Python自动录入ERP系统数据是企业信息化建设的重要环节，需要综合考虑接口协议、数据处理、安全机制和性能优化等多个方面。本文深入分析了REST API、数据库直连和文件传输三种常见实现方式，提供了完整的代码示例和工程实践建议。

在实际开发中，应根据具体业务场景选择合适的方案：REST API适合需要安全性的场景，数据库直连适合大数据量处理，文件传输适合一次性数据同步。同时要注意安全性、错误处理和性能优化，避免常见陷阱。

建议开发者在实施前进行充分的测试验证，建立完善的日志和监控体系，确保数据同步的可靠性。对于复杂的业务场景，可以结合消息队列、分布式任务处理等技术进行扩展，构建健壮的ERP数据同步系统。