全文约2000字，阅读大约需要10分钟
在日常工作中，我们经常会遇到需要处理大量文件并将数据存储至数据库或整合到一个文件的需求。这个任务对于人力和时间来说都是一大挑战。幸运的是，我们有Python这个神奇的工具，可以帮助我们自动化这个任务，省时又省力！现在，我将向你展示如何使用Python处理Excel文件并将数据存储到PostgreSQL数据库中。
先打个底：以理解为主，不够严谨，如果看完还是不会，那一定是我讲的不够好，千万别影响你们探索Python的兴趣。
在我们的奇妙冒险中，如果你想将多个excel文件整合到一个表中，需要满足一个前置条件——每个excel文件的格式和列对应的含义顺序必须一致。但是，如果表头不一样也没关系，我们可以用程序来解决这个问题。本文将带你进入Python的魔法世界，教你如何处理Excel文件并将数据存储到PostgreSQL数据库中。在开始之前，我们需要安装一些神奇的库：
-
pandas：用于处理Excel文件中的数据
-
sqlalchemy：用于连接和操作PostgreSQL数据库
安装方法这里就不再重点讲了了，直接搜网上的教程安装即可。
1.日志记录
开局先送送你一串Python日志记录的代码，可在任何场景下复用，它能够实时监测程序的运行状态，轻松解决测试和问题排查的难题。
注意：
# 定义日志记录器
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__)
2.数据库连接
接下来，我们需要配置自己的数据库信息。
# 建立与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
3.设计及创建表结构
根据文件内容来设计和创建表结构，当然你也可以用中文
# 创建存储数据的表
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() -- 创建时间
);
"""
4.处理数据
思路如下：
- 提取文件名
- 读取Excel文件数据并提取前4列
- 列名重命名
- 根据条件过滤末尾的空行
- 将数据存储到PostgreSQL表中
- 处理成功后将Excel文件移动到end目录
重点讲下
- **name：**SQL 表名
- **con：**与数据库链接的?式，推荐使?sqlalchemy的engine类型
- **schema：**相应数据库的引擎，不设置则使?数据库的默认引擎，如mysql中的innodb引擎
- **if_exists：**当数据库中已经存在数据表时对数据表的操作，有replace替换、append追加，fail则当表存在时提?
- **index：**对DataFrame的index索引的处理，为True时索引也将作为数据写?数据表
- **index_label：**当上?个参数index为True时，设置写?数据表时index的列名称
- **chunsize：**设置整数，如20000，?次写?数据时的数据?数量，当数据量很?时，需要设置，否则会链接超时写?失败。
- **dtype：**列名到 SQL 类型的字典，默认无;可选地指定列的数据类型
完整代码如下：
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)
以上就是“懒人福音：用Python让Excel数据躲猫猫！”的全部内容，希望对你有所帮助。
?
?关于Python技术储备
学好 Python 不论是就业还是做副业赚钱都不错，但要学会 Python 还是要有一个学习规划。最后大家分享一份全套的 Python 学习资料，给那些想学习 Python 的小伙伴们一点帮助！
一、Python所有方向的学习路线
Python所有方向的技术点做的整理，形成各个领域的知识点汇总，它的用处就在于，你可以按照上面的知识点去找对应的学习资源，保证自己学得较为全面。
二、Python必备开发工具
三、Python视频合集
观看零基础学习视频，看视频学习是最快捷也是最有效果的方式，跟着视频中老师的思路，从基础到深入，还是很容易入门的。
四、实战案例
光学理论是没用的，要学会跟着一起敲，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战案例来学习。
五、Python练习题
检查学习结果。
六、面试资料
我们学习Python必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有阿里大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。
最后祝大家天天进步！！
上面这份完整版的Python全套学习资料已经上传至CSDN官方，朋友如果需要可以直接微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】。
