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}")returnconn, cursor = None, Nonetry: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()