使用Python处理Excel文件并将数据存储到PostgreSQL的方法

# 定义日志记录器
log_home = '/home/xusl/log/excel'  # 请将此路径改为你自己的本地路径
log_level = logging.INFO
log_to_console = True
log_config = {
    'version': 1,
    'formatters': {
        'generic': {
            'format': '%(asctime)s %(levelname)-5.5s [%(name)s:%(lineno)s][%(threadName)s] %(message)s',
        },
        'simple': {
            'format': '%(asctime)s %(levelname)-5.5s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': os.path.join(log_home, 'excel_to_data.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file', ] if log_to_console else ['file', ],
    }
}
logging.config.dictConfig(log_config)
logger = logging.getLogger(__name__)

# 建立与PostgreSQL数据库的连接 此处需要更改为自己的数据库配置
db_user = 'dps'
db_password = 'DPS888'
db_host = '10.12.8.88'
db_port = '5432'
db_name = 'dpstest'


def get_conn():
    conn_url = 'postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}'
    engine = create_engine(conn_url.format(database=db_name, user=db_user, password=db_password, host=db_host, port=db_port),
                                 pool_size=20,
                                 pool_recycle=7200,
                                 connect_args={'connect_timeout': 30})
    try:
        with engine.connect():
            logger.info('成功连接到数据库')
    except Exception as e:
        logger.error('无法连接到数据库:', str(e))
    return engine

# 创建存储数据的表
table_name = 'public.excel_data'
ddl = """
DROP TABLE IF EXISTS public.excel_data;
CREATE TABLE IF NOT EXISTS public.excel_data (
    file_nm VARCHAR(255),
    cust_nm VARCHAR(255),
    cert_no VARCHAR(255),
    prod_nm VARCHAR(255),
    amt numeric(20,2),
    crt_dtm timestamp NOT NULL DEFAULT now() -- 创建时间
);
"""

import os
import pandas as pd
import logging.config
import shutil
import datetime

from sqlalchemy import create_engine

_tb_nm = 'excel_to_data'
_tb_nm_cn = "excel数据入库"
_service_code = _tb_nm
# 日志目录
log_home = '/home/xusl/log/excel'

# 日志level
log_level = logging.INFO

# 日志打印到控制台
log_to_console = True

# 配置日志记录器
log_config = {
    'version': 1,
    'formatters': {
        'generic': {
            'format': '%(asctime)s %(levelname)-5.5s [%(name)s:%(lineno)s][%(threadName)s] %(message)s',
        },
        'simple': {
            'format': '%(asctime)s %(levelname)-5.5s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': os.path.join(log_home, _tb_nm + '.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file', ] if log_to_console else ['file', ],
    }
}
logging.config.dictConfig(log_config)
logger = logging.getLogger(_tb_nm)


# 建立与PostgreSQL数据库的连接 39数据库
db_user = 'dps'
db_password = 'DPS888'
db_host = '10.12.8.88'
db_port = '5432'
db_name = 'dpstest'



def get_conn():
    conn_url = 'postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}'
    engine = create_engine(conn_url.format(database=db_name, user=db_user, password=db_password, host=db_host, port=db_port),
                                 pool_size=20,
                                 pool_recycle=7200,
                                 connect_args={'connect_timeout': 30})
    try:
        with engine.connect():
            print('成功连接到数据库')
    except Exception as e:
        print('无法连接到数据库:', str(e))
    return engine

# engine = create_engine(f'postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}')


# 创建存储数据的表
table_name = 'public.excel_data'

ddl = """
DROP TABLE IF EXISTS public.excel_data;
CREATE TABLE IF NOT EXISTS public.excel_data (
    file_nm VARCHAR(255),
    cust_nm VARCHAR(255),
    cert_no VARCHAR(255),
    prod_nm VARCHAR(255),
    amt numeric(20,2),
    crt_dtm timestamp NOT NULL DEFAULT now() -- 创建时间
);
"""


# 遍历指定目录下的所有Excel文件
excel_dir = '/home/xusl/data'
src_excel = '/home/xusl/data/src'
end_excel = '/home/xusl/data/end'
src_dir = 'src'
end_dir = 'end'


def deal(conn):
    for filename in os.listdir(src_excel):
        if not filename.endswith('.xlsx'):
            logging.info('没有excel文件！')
            continue
        else:
            logging.info('')
            logging.info('')
            excel_file = os.path.join(src_excel, filename)

            # 提取文件名
            file_nm = os.path.basename(excel_file)
            func_name = file_nm
            logging.info('start %s' % func_name)
            logging.info(f'Reading data from {excel_file}')
            d0 = datetime.datetime.now()

            # 读取Excel文件数据并提取前4列
            try:
                df = pd.read_excel(excel_file, usecols=[0, 1, 2, 3])
                logging.info('df读取内容：%s ' % df)
            except Exception as e:
                logging.error(f'Error reading file {excel_file}: {str(e)}')
                continue

            # 修改列名
            df.columns = ['cust_nm', 'cert_no', 'prod_nm', 'amt']
            logging.info('df修改后内容：%s ' % df)

            # 根据条件过滤末尾的空行
            if not df.empty and df.iloc[-1].isnull().all():
                df = df[:-1]
                logging.debug('df删减末尾后：%s ' % df)

            # 将数据存储到PostgreSQL表中
            df['file_nm'] = file_nm
            df = df[['file_nm', 'cust_nm', 'cert_no', 'prod_nm', 'amt']]

            try:
                # 将整个DF导入数据库中
                df.to_sql(name='excel_data', schema='public', con=conn, if_exists="append", index=False)
                d1 = datetime.datetime.now()
                s = (d1 - d0).total_seconds()
                logging.info('... end %s, 耗时: %s 秒. ' % (func_name, s))
            except Exception as e:
                logging.error(f'Error inserting data from file {excel_file}: {str(e)}')
                continue

            # 处理成功后将Excel文件移动到end目录
            src_file = os.path.join(src_excel, filename)
            end_file = os.path.join(end_excel, filename)
            try:
                shutil.move(src_file, end_file)
            except Exception as e:
                logging.error(f'Error moving file {src_file} to {end_file}: {str(e)}')

    # 关闭数据库连接
    # engine.dispose()


if __name__ == '__main__':
    engine = get_conn()
    deal(engine)