字段对比清洗

import pandas as pd
import psycopg2
from psycopg2 import sql

# 数据库连接配置
DB_CONFIG = {
    "host": "",
    "user": "",
    "password": "",
    "dbname": "",
    "port": ,
}

def get_excel_fields(excel_file, sheet_name, column_name):
    """从 Excel 读取字段列表"""
    df = pd.read_excel(excel_file, sheet_name=sheet_name)
    
    # 确保列名一致
    if column_name not in df.columns:
        raise ValueError(f"列 '{column_name}' 未在 Excel 文件中找到")
    
    # 移除空值，并转换为小写，避免大小写问题
    return set(df[column_name].dropna().astype(str).str.lower())

def get_db_fields(cursor, table_name):
    """获取数据库表的字段列表（转换为小写）"""
    query = sql.SQL("""
        SELECT column_name FROM information_schema.columns 
        WHERE LOWER(table_name) = LOWER({})
    """).format(sql.Literal(table_name))
    
    cursor.execute(query)
    
    # 确保字段名全部转换为小写
    return set(row[0].lower() for row in cursor.fetchall())

def drop_columns(cursor, table_name, columns_to_drop):
    """删除数据库表中的指定字段"""
    for column in columns_to_drop:
        query = sql.SQL("ALTER TABLE {} DROP COLUMN {} CASCADE").format(
            sql.Identifier(table_name),
            sql.Identifier(column)
        )
        
        print(f"执行 SQL: {query.as_string(cursor.connection)}")  # 打印 SQL 方便调试
        
        try:
            cursor.execute(query)
            print(f"✅ 删除成功: {column}")
        except psycopg2.Error as e:
            print(f"❌ 删除失败: {column}, 错误: {e}")

def main():
    # 读取 Excel 表结构
    excel_file = "./test.xlsx"
    sheet_name = "数据结构"
    column_name = "编码"
    table_name = "test_name"

    try:
        excel_fields = get_excel_fields(excel_file, sheet_name, column_name)
        print(f"✅ Excel 字段列表: {excel_fields}")
    except Exception as e:
        print(f"❌ 读取 Excel 失败: {e}")
        return

    conn, cursor = None, None
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cursor = conn.cursor()

        db_fields = get_db_fields(cursor, table_name)
        print(f"✅ 数据库字段列表: {db_fields}")

        columns_to_drop = db_fields - excel_fields  # 计算需要删除的字段

        if columns_to_drop:
            print(f"⚠️ 需要删除的字段: {columns_to_drop}")
            drop_columns(cursor, table_name, columns_to_drop)
            conn.commit()
            print("✅ 未使用的字段已删除")
        else:
            print("✅ 没有未使用的字段需要删除")

    except psycopg2.Error as e:
        print(f"❌ 数据库错误: {e}")
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

if __name__ == "__main__":
    main()