三、处理文件上传

一个基本的文件上传表

  • application/x-www-form-urlencoded：这是默认格式，也是除包含文件字段的表单之外的任何表单的最佳格式。

  • multipart/form-data：当表单中至少有一个字段是文件字段时，需要此格式。

  • text/plain: 这种格式没有实际用途，所以你应该忽略它。

<!doctype html>
<html>
  <head>
    <title>File Upload</title>
  </head>
  <body>
    <h1>File Upload</h1>
    <form method="POST" action="" enctype="multipart/form-data">
      <p><input type="file" name="file"></p>
      <p><input type="submit" value="Submit"></p>
    </form>
  </body>
</html>

  • multiple可用于允许在单个文件字段中上传多个文件。例子：

        <input type="file" name="file" multiple>

  • accept可用于过滤可以选择的允许文件类型，可以是文件扩展名，也可以是媒体类型。例子：

        <input type="file" name="doc_file" accept=".doc,.docx">
    <input type="file" name="image_file" accept="image/*">

使用 Flask 接受文件提交

暂时忽略验证和安全等重要方面，下面显示的简短 Flask 应用程序接受使用上一节所示表单上传的文件，并将提交的文件写入当前目录：

from flask import Flask, render_template, request, redirect, url_for

app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        uploaded_file = request.files['file']
        if uploaded_file.filename != '':
            uploaded_file.save(uploaded_file.filename)
        return redirect(url_for('index'))
    return render_template('index.html')

多个文件上传

# getlist() 方法您可以在 for 循环中访问所有文件
    for uploaded_file in request.files.getlist('file'):
        if uploaded_file.filename != '':
            uploaded_file.save(uploaded_file.filename)

使用Flask-WTF扩展来处理表单

from flask_wtf import FlaskForm
from flask_wtf.file import FileField
from wtforms import SubmitField

class MyForm(FlaskForm):
    file = FileField('File')
    submit = SubmitField('Submit')

限制上传文件的大小

app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024

验证文件名

app.config['UPLOAD_EXTENSIONS'] = ['.jpg', '.png', '.gif']
    filename = uploaded_file.filename
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        if file_ext not in current_app.config['UPLOAD_EXTENSIONS']:
            abort(400)

生成上传文件名

有时候上传的文件名会重复，那么就会覆盖之前的文件名；最安全方法是忽略客户端提供的文件名并生成您自己的文件名，然后将其传递给该save()方法。这种技术运作良好的示例用例是头像图像上传。每个用户的头像可以以用户id作为文件名保存，这样客户端提供的文件名就可以丢弃了。如果您的应用程序使用 Flask-Login，您可以实现以下save()调用：

uploaded_file.save(os.path.join('static/avatars', current_user.get_id()))

在其他情况下，最好保留客户端提供的文件名，因此必须首先清理文件名。对于这些情况，Werkzeug 提供了secure_filename()函数。让我们通过在 Python 会话中运行一些测试来看看这个函数是如何工作的：

>>> from werkzeug.utils import secure_filename
>>> secure_filename('foo.jpg')
'foo.jpg'
>>> secure_filename('/some/path/foo.jpg')
'some_path_foo.jpg'
>>> secure_filename('../../../.bashrc')
'bashrc'

保存到指定路径

import os
from flask import Flask, render_template, request, redirect, url_for, abort
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024
app.config['UPLOAD_EXTENSIONS'] = ['.jpg', '.png', '.gif']
app.config['UPLOAD_PATH'] = 'uploads'

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/', methods=['POST'])
def upload_files():
    uploaded_file = request.files['file']
    filename = secure_filename(uploaded_file.filename)
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        if file_ext not in app.config['UPLOAD_EXTENSIONS']:
            abort(400)
        uploaded_file.save(os.path.join(app.config['UPLOAD_PATH'], filename))
    return redirect(url_for('index'))

验证文件内容

编写一个validate_image()对图像执行内容验证的函数

import imghdr

def validate_image(stream):
    header = stream.read(512)
    stream.seek(0)
    format = imghdr.what(None, header)
    if not format:
        return None
    return '.' + (format if format != 'jpeg' else 'jpg')

它首先从流中读取 512 个字节，然后将流指针重置回来，因为稍后save()调用该函数时，我们希望它看到整个流。图像数据的前 512 个字节将足以识别图像的格式。

imghdr.what()如果第一个参数是文件名，则该函数可以查看存储在磁盘上的文件，否则，如果第一个参数是None并且数据在第二个参数中传递，则它可以查看存储在内存中的数据。该FileStorage对象为我们提供了一个流，因此最方便的选择是从中读取安全数量的数据并将其作为第二个参数中的字节序列传递。

imghdr.what()是检测到的图像格式。该函数支持多种格式，其中流行的jpeg,png和gif. 如果检测到未知图像格式，则返回值为None。如果检测到格式，则返回格式的名称。最方便的是将格式作为文件扩展名返回，因为应用程序可以确保检测到的扩展名与文件扩展名匹配，因此该validate_image()函数将检测到的格式转换为文件扩展名。这就像为所有图像格式添加一个点作为前缀一样简单，除了jpeg，它通常使用.jpg扩展名

综合以上功能

import imghdr
import os
from flask import Flask, render_template, request, redirect, url_for, abort
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024
app.config['UPLOAD_EXTENSIONS'] = ['.jpg', '.png', '.gif']
app.config['UPLOAD_PATH'] = 'uploads'

def validate_image(stream):
    header = stream.read(512)
    stream.seek(0) 
    format = imghdr.what(None, header)
    if not format:
        return None
    return '.' + (format if format != 'jpeg' else 'jpg')

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/', methods=['POST'])
def upload_files():
    uploaded_file = request.files['file']
    filename = secure_filename(uploaded_file.filename)
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        # 这种扩展检查首先确保文件扩展名在允许列表中，然后确保通过查看数据流检测到的文件扩展名与文件扩展名相同。
        if file_ext not in app.config['UPLOAD_EXTENSIONS'] or \
                file_ext != validate_image(uploaded_file.stream):
            abort(400)
        uploaded_file.save(os.path.join(app.config['UPLOAD_PATH'], filename))
    return redirect(url_for('index'))
# if .... app.run ...

上传并展示

# 建议在保存上传的头像图像时使用用户 ID 作为文件名
uploaded_file.save(os.path.join('static/avatars', current_user.get_id()))

# URL 路径
url_for('static', filename='avatars/' + str(user_id))

或者，可以将上传的内容保存到静态文件夹之外的目录

from flask import send_from_directory

@app.route('/uploads/<filename>')
def upload(filename):
    return send_from_directory(app.config['UPLOAD_PATH'], filename)

如果您只想向登录用户提供对上传的访问权限，您可以将 Flask-Login 的@login_required装饰器添加到此路由，或任何其他用于正常路由的身份验证或角色检查机制

完整代码：

import imghdr
import os
from flask import Flask, render_template, request, redirect, url_for, abort, \
    send_from_directory
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024
app.config['UPLOAD_EXTENSIONS'] = ['.jpg', '.png', '.gif']
app.config['UPLOAD_PATH'] = 'uploads'

def validate_image(stream):
    header = stream.read(512)  # 512 bytes should be enough for a header check
    stream.seek(0)  # reset stream pointer
    format = imghdr.what(None, header)
    if not format:
        return None
    return '.' + (format if format != 'jpeg' else 'jpg')

@app.route('/')
def index():
    files = os.listdir(app.config['UPLOAD_PATH'])
    return render_template('index.html', files=files)

@app.route('/', methods=['POST'])
def upload_files():
    uploaded_file = request.files['file']
    filename = secure_filename(uploaded_file.filename)
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        if file_ext not in app.config['UPLOAD_EXTENSIONS'] or \
                file_ext != validate_image(uploaded_file.stream):
            abort(400)
        uploaded_file.save(os.path.join(app.config['UPLOAD_PATH'], filename))
    return redirect(url_for('index'))

@app.route('/uploads/<filename>')
def upload(filename):
    return send_from_directory(app.config['UPLOAD_PATH'], filename)

除了新upload()函数之外，index()视图函数使用获取上传位置的文件列表os.listdir()并将其发送到模板进行渲染。更新为显示上传的index.html模板如下所示：

<!doctype html>
<html>
  <head>
    <title>File Upload</title>
  </head>
  <body>
    <h1>File Upload</h1>
    <form method="POST" action="" enctype="multipart/form-data">
      <p><input type="file" name="file"></p>
      <p><input type="submit" value="Submit"></p>
    </form>
    <hr>
    {% for file in files %}
      <img src="{{ url_for('upload', filename=file) }}" style="width: 64px">
    {% endfor %}
  </body>
</html>

上传并私有

当用户将私人文件上传到应用程序时，需要进行额外的检查，以防止一个用户与未经授权的方共享文件。这些情况的解决方案需要upload()上面显示的视图函数的变体，以及额外的访问检查。

一个常见的要求是只与其所有者共享上传的文件。存在此要求时存储上传的一种便捷方法是为每个用户使用单独的目录。

例如，可以将给定用户的上传内容保存到该uploads/<user_id>目录中，然后uploads()可以修改该功能以仅提供来自用户自己上传目录的上传内容，从而使一个用户无法查看另一个用户的文件。您可以在下面看到此技术的可能实现，再次假设使用 Flask-Login：

@app.route('/uploads/<filename>')
@login_required
def upload(filename):
    return send_from_directory(os.path.join(
        app.config['UPLOAD_PATH'], current_user.get_id()), filename)

显示上传进度

<html>
  <head>
    <title>File Upload</title>
    <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/dropzone/5.7.1/min/dropzone.min.css">
  </head>
  <body>
    <h1>File Upload</h1>
    <form action="{{ url_for('upload_files') }}" class="dropzone">
    </form>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/dropzone/5.7.1/min/dropzone.min.js"></script>
  </body>
</html>

如果文件上传失败，无论是由于文件太大还是无效，dropzone 都希望显示错误消息。因为我们的服务器当前正在返回 413 和 400 错误的标准 Flask 错误页面，所以您会在错误弹出窗口中看到一些 HTML 乱码。为了解决这个问题，我们可以更新服务器以将其错误响应作为文本返回。

当请求负载大于配置中设置的大小时，Flask 会生成文件过大条件的 413 错误。要覆盖默认错误页面，我们必须使用app.errorhandler装饰器：

@app.errorhandler(413)
def too_large(e):
    return "File is too large", 413

当任何验证检查失败时，应用程序会生成第二个错误条件。在这种情况下，错误是通过abort(400)调用生成的。相反，可以直接生成响应：

        if file_ext not in app.config['UPLOAD_EXTENSIONS'] or \
                file_ext != validate_image(uploaded_file.stream):
            return "Invalid image", 400

我将要进行的最终更改并不是真正必要的，但它可以节省一些带宽。为了成功上传，服务器返回了一个redirect()返回到主路由的信息。这导致再次显示上传表单，并刷新页面底部的上传缩略图列表。现在这些都不是必需的，因为上传是由 dropzone 作为后台请求完成的，所以我们可以消除重定向并切换到代码为 204 的空响应。

完整代码

这是设计用于 dropzone.js的app.py的完整和更新版本：

import imghdr
import os
from flask import Flask, render_template, request, redirect, url_for, abort, \
    send_from_directory
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 2 * 1024 * 1024
app.config['UPLOAD_EXTENSIONS'] = ['.jpg', '.png', '.gif']
app.config['UPLOAD_PATH'] = 'uploads'

def validate_image(stream):
    header = stream.read(512)
    stream.seek(0)
    format = imghdr.what(None, header)
    if not format:
        return None
    return '.' + (format if format != 'jpeg' else 'jpg')

@app.errorhandler(413)
def too_large(e):
    return "File is too large", 413

@app.route('/')
def index():
    files = os.listdir(app.config['UPLOAD_PATH'])
    return render_template('index.html', files=files)

@app.route('/', methods=['POST'])
def upload_files():
    uploaded_file = request.files['file']
    filename = secure_filename(uploaded_file.filename)
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        if file_ext not in app.config['UPLOAD_EXTENSIONS'] or \
                file_ext != validate_image(uploaded_file.stream):
            return "Invalid image", 400
        uploaded_file.save(os.path.join(app.config['UPLOAD_PATH'], filename))
    return '', 204

@app.route('/uploads/<filename>')
def upload(filename):
    return send_from_directory(app.config['UPLOAD_PATH'], filename)

# if .... app.run ....

参考网站：https://blog.miguelgrinberg.com/post/handling-file-uploads-with-flask

Previous二、摸版Next四、 Flask 定时任务

Last updated