如何用Python将格式不匹配的Excel数据导入指定PostgreSQL表

需求说明 #

现有PostgreSQL表结构如下：

CREATE TABLE public.test (
    "Test_Id" int4 NOT NULL,
    "Test_Name" text NOT NULL,
    "Test_Code" text NULL,
    "Test_Description" text NOT NULL,
    "Test_Parent_Id" numeric NOT NULL,
    "Level_Test_Id" numeric NULL,
    "Level_Test_Name" text null
    CONSTRAINT "Test_Id_pkey" PRIMARY KEY ("Test_Id")
);

需要将格式不匹配的Excel数据导入该表，其中Excel包含父测试（如TestA）和子测试（如TestA indicator name A1.1），且Level_Test_Name可选值为('Level 1', 'Level 2', 'Level 3', 'Level 4','Level 5')。

依赖准备 #

先安装所需Python库：

pip install pandas psycopg2-binary openpyxl

代码实现 #

import pandas as pd
import psycopg2
from psycopg2 import sql

# 1. 读取Excel数据（假设Excel文件名为test_data.xlsx，数据在Sheet1）
df = pd.read_excel('test_data.xlsx', sheet_name='Sheet1', engine='openpyxl')

# 2. 预处理数据：识别父子测试，填充对应字段
# 先提取父测试列表（假设父测试名称不含"indicator"关键词，可根据实际调整规则）
parent_tests = df[~df['Test_Name'].str.contains('indicator')]['Test_Name'].unique()

# 创建父测试ID映射（这里Test_Id从1开始自增，可根据实际调整）
parent_id_map = {name: idx+1 for idx, name in enumerate(parent_tests)}

# 初始化结果列表
db_data = []
current_test_id = 1

# 遍历每条数据
for _, row in df.iterrows():
    test_name = row['Test_Name']
    test_code = row.get('Test_Code', None)
    test_desc = row['Test_Description']  # 假设Excel有该列，若无则需补充默认值
    
    # 判断是父测试还是子测试
    if test_name in parent_id_map:
        # 父测试：Test_Parent_Id设为0（顶级父节点），Level设为Level 1
        test_parent_id = 0
        level_test_name = 'Level 1'
        level_test_id = 1
        db_data.append({
            'Test_Id': current_test_id,
            'Test_Name': test_name,
            'Test_Code': test_code,
            'Test_Description': test_desc,
            'Test_Parent_Id': test_parent_id,
            'Level_Test_Id': level_test_id,
            'Level_Test_Name': level_test_name
        })
        parent_id_map[test_name] = current_test_id
        current_test_id += 1
    else:
        # 子测试：匹配对应的父测试（假设子测试名称以父测试名称开头）
        parent_name = next((name for name in parent_id_map if test_name.startswith(name)), None)
        if parent_name:
            test_parent_id = parent_id_map[parent_name]
            # 根据父级Level推导子级Level（这里假设父是Level1则子是Level2，可按需调整）
            level_test_id = 2
            level_test_name = 'Level 2'
            # 若有更深层级，可扩展判断逻辑
            # ...
        else:
            # 无匹配父测试的情况，可设为默认值或抛出异常
            test_parent_id = 0
            level_test_id = None
            level_test_name = None
        
        db_data.append({
            'Test_Id': current_test_id,
            'Test_Name': test_name,
            'Test_Code': test_code,
            'Test_Description': test_desc,
            'Test_Parent_Id': test_parent_id,
            'Level_Test_Id': level_test_id,
            'Level_Test_Name': level_test_name
        })
        current_test_id += 1

# 3. 连接PostgreSQL并写入数据
conn = psycopg2.connect(
    dbname='your_db_name',
    user='your_username',
    password='your_password',
    host='your_host',
    port='your_port'
)
cur = conn.cursor()

# 批量插入数据
insert_query = sql.SQL("""
    INSERT INTO public.test (
        "Test_Id", "Test_Name", "Test_Code", "Test_Description",
        "Test_Parent_Id", "Level_Test_Id", "Level_Test_Name"
    ) VALUES (%s, %s, %s, %s, %s, %s, %s)
""")

for item in db_data:
    cur.execute(insert_query, (
        item['Test_Id'],
        item['Test_Name'],
        item['Test_Code'],
        item['Test_Description'],
        item['Test_Parent_Id'],
        item['Level_Test_Id'],
        item['Level_Test_Name']
    ))

# 提交事务并关闭连接
conn.commit()
cur.close()
conn.close()

print("数据导入完成")

关键逻辑说明 #

• 父子测试识别：通过判断Test_Name是否包含"indicator"关键词区分父/子测试，可根据Excel实际格式调整识别规则（比如前缀匹配、层级标识等）
• Test_Id生成：采用自增方式生成，若Excel已有Test_Id列，可直接使用该列值
• Level字段填充：示例中默认父测试为Level1，子测试为Level2，若有更深层级，可扩展逻辑（比如通过子测试名称中的编号如A1.1判断层级）
• 异常处理：示例中对无匹配父测试的情况做了默认值处理，实际使用中可根据需求添加异常捕获或日志记录

内容的提问来源于stack exchange，提问作者Safa edligo