Python实现Excel至Oracle的文件夹触发式自动ETL方案

前置准备 #

1. 安装依赖包：

pip install pandas cx_Oracle openpyxl

2. 确保Oracle Client已正确配置，cx_Oracle可正常连接数据库（可通过TOAD验证连接可用性）。

完整实现代码 #

import os
import pandas as pd
import cx_Oracle
from time import sleep

# -------------------------- 配置项 --------------------------
# Oracle数据库连接字符串：用户名/密码@主机:端口/服务名
oracle_connection_string = 'c##chbz/excelpass@localhost:1521/XE'
# 监控的Excel文件夹路径
folder_path = r'C:\Users\USERR\Desktop\FolderDailyExcels'
# 目标Oracle表名
TARGET_TABLE = 'Project_table'
# Excel列与Oracle字段映射：Excel列名 -> Oracle字段名
COLUMN_MAPPING = {
    'column1': 'FULLNAME',
    'column2': 'NOOFATTACK',
    'column3': 'DATEOFATTACK'
}
# 监控间隔（秒）
MONITOR_INTERVAL = 60
# -----------------------------------------------------------

# 已处理文件记录，避免重复上传
processed_files = set()

def connect_to_oracle():
    """建立Oracle数据库连接"""
    try:
        connection = cx_Oracle.connect(oracle_connection_string)
        print("数据库连接成功")
        return connection
    except cx_Oracle.DatabaseError as e:
        print(f"数据库连接失败: {e}")
        return None

def read_excel(file_path):
    """读取Excel文件返回DataFrame"""
    try:
        # 支持.xlsx格式，若需.xls需安装xlrd
        data = pd.read_excel(file_path, engine='openpyxl')
        print(f"成功读取文件: {file_path}")
        return data
    except Exception as e:
        print(f"读取文件失败 {file_path}: {e}")
        return None

def upload_to_oracle(data):
    """将DataFrame数据批量插入Oracle"""
    connection = connect_to_oracle()
    if not connection:
        return
    
    cursor = connection.cursor()
    # 构造插入SQL
    oracle_columns = ', '.join(COLUMN_MAPPING.values())
    placeholders = ', '.join([f':{i+1}' for i in range(len(COLUMN_MAPPING))])
    sql = f"INSERT INTO {TARGET_TABLE} ({oracle_columns}) VALUES ({placeholders})"
    
    try:
        # 批量插入（比逐行插入效率更高）
        rows = [tuple(row[col] for col in COLUMN_MAPPING.keys()) for _, row in data.iterrows()]
        cursor.executemany(sql, rows)
        connection.commit()
        print(f"成功上传{len(rows)}条数据至{TARGET_TABLE}")
    except Exception as e:
        connection.rollback()
        print(f"数据上传失败: {e}")
    finally:
        cursor.close()
        connection.close()

def monitor_folder():
    """监控文件夹，自动处理新增Excel文件"""
    print(f"开始监控文件夹: {folder_path}，间隔{MONITOR_INTERVAL}秒")
    while True:
        current_files = set()
        for file in os.listdir(folder_path):
            if file.lower().endswith(('.xlsx', '.xls')):
                file_path = os.path.abspath(os.path.join(folder_path, file))
                current_files.add(file_path)
        
        # 找出新增的未处理文件
        new_files = current_files - processed_files
        for file_path in new_files:
            print(f"发现新文件: {file_path}")
            data = read_excel(file_path)
            if data is not None:
                upload_to_oracle(data)
                processed_files.add(file_path)
        
        sleep(MONITOR_INTERVAL)

if __name__ == "__main__":
    monitor_folder()

关键配置说明 #

1. 连接字符串：替换为你的Oracle用户名、密码、主机地址、端口和服务名。
2. 文件夹路径：修改为你要监控的Excel文件存放目录。
3. 字段映射：确保COLUMN_MAPPING中的Excel列名与你的文件列名一致，Oracle字段名与目标表字段名匹配。
4. 目标表：提前在Oracle中创建好目标表，字段类型需与Excel数据类型兼容（比如日期字段需确保Excel中的日期格式正确）。

优化建议 #

• 若需更高效的文件夹监控（无需轮询），可使用watchdog库替代time.sleep轮询方式。
• 可添加文件处理后的归档逻辑（如移动至已处理文件夹）。
• 增加日志记录功能，替代print输出以便排查问题。

内容的提问来源于stack exchange，提问作者Chibuzo Attah