Python+Selenium下载CSV导入Postgres字段全为字符串及整型带.0求助

技能水平 #

初次接触编程，决心完成项目，愿意主动查阅资料学习。

需求目标 #

暂时无法实现所需API与Postgres数据库的对接，因此采用Selenium模拟登录网页门户下载CSV文件，通过Python将CSV数据导入Postgres数据库，用于对接报表软件，为当前临时过渡方案。

预期与实际结果 #

• 预期：脚本可自动登录网页门户、下载CSV文件、将数据导入Postgres数据库。
• 实际：脚本可正常完成登录、CSV下载、数据入库流程，但入库后所有字段均为字符串类型，大量整数字段值末尾带.0，经纬度字段也为字符串类型，无程序报错，但使用Tableau制作报表时受影响。

已尝试方案 #

查阅Pandas相关文档，在pd.read_csv方法中添加index_col=None参数，未解决问题。

你的脚本出现这个问题是两个核心原因导致的：

1. 建表逻辑硬编码所有字段为字符串：你创建表的时候用了固定的VARCHAR(20000)作为所有字段的类型，不管原始数据是数字还是日期，入库后自然全是字符串类型
2. 数据类型处理未生效：你虽然在Pandas读取下载文件的时候做了float转int的处理，但是后续入库逻辑是直接读取CSV文件的原始内容，逐行转成字符串拼接SQL插入，之前Pandas处理的类型完全没有用到

步骤1：安装依赖 #

使用Pandas自带的to_sql方法替代手动拼SQL的入库逻辑，自动推断字段类型，性能更高也更安全，先安装需要的依赖：

pip install sqlalchemy psycopg2-binary

步骤2：优化数据清洗逻辑 #

把原脚本中bulkDownload函数里的低效循环替换为向量化操作，更稳定地处理字段类型：

def bulkDownload(url, file_format):
    driver.get(url)
    downloadBtn = waitAndGetElement(6000, EC.presence_of_element_located((By.XPATH, '//a[text()=" Download Now"]')))
    time.sleep(5)
    downloadBtn.click()
    current_url = driver.current_url
    export_num = current_url.split("/")[-1]

    file_path = file_format.format(export_num)
    while not os.path.exists(file_path):
        time.sleep(2)
    
    print(f"正在处理CSV文件：{file_path}")
    df = pd.read_csv(file_path, low_memory=False, index_col=None, header=0)
    # 去重列
    df = df.loc[:,~df.columns.duplicated()]
    # 批量处理数值列：空值填0，整数类型的float转成int
    for col in df.columns:
        # 判断列是否是float类型
        if df[col].dtype == 'float64':
            # 先填充空值为0
            df[col] = df[col].fillna(0)
            # 判断是否所有值都是整数（没有小数部分），经纬度这类小数列会保留原类型
            if (df[col] % 1 == 0).all():
                df[col] = df[col].astype('int64')
    return df

步骤3：替换原有入库逻辑 #

删除原脚本中从import psycopg2开始到cursor.close()之间的所有手动入库代码，替换为以下逻辑：

from sqlalchemy import create_engine

# 数据库连接配置
db_user = os.environ.get('DB_USER')
db_password = os.environ.get('DB_PASS')
db_name = os.environ.get('DB_NAME')
engine = create_engine(f'postgresql://{db_user}:{db_password}@127.0.0.1:5432/{db_name}')

# 遍历csvs文件夹下的所有csv文件直接入库
csv_dir = 'csvs'
for filename in os.listdir(csv_dir):
    if not filename.endswith('.csv'):
        continue
    table_name = os.path.splitext(filename)[0].replace(' ', '')
    file_path = os.path.join(csv_dir, filename)
    # 读csv的时候自动推断类型
    df = pd.read_csv(file_path, low_memory=False)
    # 自动建表+入库，if_exists='replace'表示覆盖旧表
    df.to_sql(table_name, engine, if_exists='replace', index=False)
    print(f'{table_name} 表数据入库完成')

# 原有的删除下载目录CSV文件、统计运行时间逻辑保留即可

注意事项 #

• 如果有明确的字段类型要求，可以给pd.read_csv加dtype参数手动指定每一列的类型，比如dtype={'经度': float, '订单号': int, '下单时间': 'datetime64[ns]'}
• 新的入库逻辑支持批量插入，数据量大的时候性能比原来的逐行插入高数十倍，同时避免了SQL注入风险

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