Python中如何设计将Excel文件上传网站并导入数据库的功能？
tornado + celery + redis + mysql。
1.由于当 Excel 文件过大时，上传及导入数据库时需花费较长时间，所以采用 celery 异步处理，首先将 Excel 保存作为临时文件上传保存服务端本地，然后进行 Excel 文件解析操作，并将数据导入数据库。
2.导入时牵扯到数据重复问题，此时采用临时表进行数据去重，当前实现的方式是原生 sql （ orm 有点 hold 不住），创建临时表， 临时表去重， 然后临时表及主表进行 join 筛选出新数据，然后进行插入。经过一番折腾，算是达到了预期目标，然后再继续下面的坑。
3.导入后，需将导入结果返回给界面，比如 Excel 中的数据是有问题的，由于 celery 异步方式，无法将结果返回给主程， 所以请教了下别人，说是可借助数据库新建一张表，负责记录导入结果。
- 那么，下面就是新建表的问题，如何建这个表呢？ 个人觉得需要有个字段来记录导入的数据的唯一性，然后通过这个字段进行筛选查询当前导入的是哪些个数据。
我的思路就是这了，我想问一下有没有别的方式，好的方式，主要问题就是在这个异步线程结果获取这里，因为上传导入数据库是需要花费不定时时间的，无法上传后立刻判断 celery 的 task 的 id 和状态。
Python中如何设计将Excel文件上传网站并导入数据库的功能？
# 核心代码示例：Flask + pandas + SQLAlchemy实现Excel上传入库
from flask import Flask, request, jsonify
import pandas as pd
from sqlalchemy import create_engine
import os
app = Flask(__name__)
UPLOAD_FOLDER = 'uploads'
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
# 数据库配置（以MySQL为例）
DB_CONFIG = {
'host': 'localhost',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database'
}
def excel_to_database(filepath, table_name):
"""将Excel文件数据导入数据库"""
try:
# 读取Excel文件（支持xlsx和xls）
df = pd.read_excel(filepath)
# 创建数据库连接
engine = create_engine(
f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"
f"@{DB_CONFIG['host']}/{DB_CONFIG['database']}"
)
# 将DataFrame写入数据库
df.to_sql(
name=table_name,
con=engine,
if_exists='append', # 追加模式
index=False
)
return True, f"成功导入 {len(df)} 条记录"
except Exception as e:
return False, str(e)
@app.route('/upload-excel', methods=['POST'])
def upload_excel():
"""处理文件上传的API端点"""
if 'excel_file' not in request.files:
return jsonify({'error': '未找到文件'}), 400
file = request.files['excel_file']
table_name = request.form.get('table_name', 'excel_data')
if file.filename == '':
return jsonify({'error': '未选择文件'}), 400
if not file.filename.endswith(('.xlsx', '.xls')):
return jsonify({'error': '仅支持Excel文件'}), 400
# 保存上传的文件
filepath = os.path.join(app.config['UPLOAD_FOLDER'], file.filename)
file.save(filepath)
# 导入数据库
success, message = excel_to_database(filepath, table_name)
# 清理临时文件
if os.path.exists(filepath):
os.remove(filepath)
if success:
return jsonify({'message': message}), 200
else:
return jsonify({'error': message}), 500
if __name__ == '__main__':
# 确保上传目录存在
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
app.run(debug=True)
前端HTML示例（配合上述后端）：
<!DOCTYPE html>
<html>
<body>
<form action="/upload-excel" method="post" enctype="multipart/form-data">
<input type="file" name="excel_file" accept=".xlsx,.xls" required>
<input type="text" name="table_name" placeholder="表名（可选）">
<button type="submit">上传并导入</button>
</form>
</body>
</html>
关键点说明：
- 文件处理：使用Flask的
request.files接收上传文件，通过pandas.read_excel()读取数据 - 数据库操作：用SQLAlchemy创建引擎，
DataFrame.to_sql()方法直接写入 - 错误处理：包含文件类型验证和异常捕获
- 安全考虑：实际部署时需要添加文件大小限制、文件名消毒、数据库事务等
替代方案：
- 大文件处理：可用
chunksize参数分块读取 - 复杂映射：可在读取Excel后对DataFrame进行数据清洗和转换
- 异步处理：Celery处理长时间导入任务
一句话建议： 核心是用pandas做数据桥梁，配合Web框架处理文件上传。
上传时记录操作人，操作时间，生成 ID （ table_A ），传完后对服务器上的 excel 进行 rename, xxx_uid_id.xlsx，这样一旦后端导入异常时可以读取 excel 文件名中的 id，根据 id 匹配 table_A 中的 uid，发送邮件或者短信通知任务状态。
嗯，有点启发，我目前是 table_M 表示目的表（ excel 中的数据存放表），table_L 表示日志表（记录 Excel 导入过程中，哪些数据异常，给用户以提示，文中说的那个新建表来记录导入数据的问题，说的就是这个 table_L ），当然除此之外还有一个临时表用以数据去重。我想得到的方式是，在界面上，就将导入的结果显示出来（也就是第一步上传导入数据库，第二步，从 table_L 中查当前所操作的表的产生的日志，并返回给界面。如果是用邮件或者短信息的话，可以，但没办法，应该不会被领导接纳。不过我想可以，模糊匹配 table_L 中的那个具有唯一性的关键字，然后根据时间排序取最新的，也就能保证是最新插入的数据了，这样应该就可以取出来操作日志了）。大神觉得如何？
还有个问题就是，我并不能知道跑在 celery 中的这个读 Excel 并导入数据库的操作是否已经完成，如何判定这个状态呢？因为没有完成的话，我去查 table_L 中的数据的话，也是会出问题，这该怎么办才好，再支个招吧。
用 pandas 来处理 Excel 数据（标题校验等等），用 marshmallow 来校验数据，最后插入数据就看你用啥 orm 或者 sql。总的来说一次性导入 1 万条数据是可以接受的，只是内存会涨 100M 左右，超过 10 万数据导入直接就无法接受了，所以你还要限制上传文件大小。
这个已经不是转格式的问题了，利用 Python 的一些库也可以将数据导入，数据重复也做了处理，现在的问题是，我把数据处理放到了 celery 中进行处理，这样主线程无法获取到 celery 的处理结果，因为数据导入后，某些处理结果日志还是要返回给前端的，现在无法获取到 celery 的处理状态，就不好进行下一步查询结果。
这么说吧，比如有个 100m 的数据大小的文件要导入数据库，那么使用 celery 的话，主线程可以立刻返回给前端结果，比如上传成功，导入成功，（而且 celery 处理完之后，会把一些处理日志放到了 table_L 表中），实际上，celery 还在处理过程中国，当 celery 的处理状态还是 PENDING， 我就无法获取到 table_L 表中的日志情况，因为我现在要把处理日志返回给前端，告诉用户，那些数据异常，需要处理，现在是这个问题了。还有就是索引唯一性，也是去重的一个方式，不过我采用的是临时表的方式。
我曾经提供过类似的方法，基本思路是与用户交互的程序负责接收其他真正的导入或者数据计算程序的状态或者消息。
错误提示显示给用户看，或者将这些错误消息(日志)持久化，等待用户回来查询。
emmm。。
理一下逻辑？
先，上传。前端显示上传成功。后台开始异步工作。生成 taskid。和前端上传的那条记录绑定。前台可以看到这条任务是处理中。后台随时可以根据 taskid 修改这个状态。
你既然已经是异步了。前端不可能同步显示。一定也是一个异步显示。就看体验了
目前是将日志进行持久化了，然而我该什么时候去查询这些日志呢？这是个问题，所以我才想到去监听 celery 的处理结果，等状态更新为 SUCCESS 的时候，再去数据库查询这些日志，然而这就引出另一个问题，就是轮询时间等待的问题，文件稍微大点，岂不要等很久。。。
因为做了些尝试，发现可以获取到异步处理的 id 及状态，所以原本想着，前端这边做个轮询的机制，进行同步结果反馈，做出来的效果的话，就像你在页面上提交东西，页面按钮进行转圈那种等待效果，感觉是这样。
嗯，类似于“ cots_18_09_20_Sep_09_1537423865 ”这样的，记录日志的数据表里 key 字段，我就三个字段，id，key, msg，这个 key 在第一步上传处理时，已经产生，请求发出后，会将其返回，用以第二步中的日志查询。目前采用的就是这种方式了，以后有新的想法，再持续更新吧，感觉做的还是有点 low。
