第 9 章 · 文件上传与下载

<form action="/api/user/avatar" method="post" enctype="multipart/form-data">
  <input type="file" name="file" accept="image/png,image/jpeg,image/gif">
  <button type="submit">上传头像</button>
</form>

<!-- 关键点：
     method="post"          文件上传必须是 POST
     enctype="multipart/form-data"  缺少它就传不了文件
     name="file"            这个名字要和后端 request.files["file"] 对上
     accept="..."           只是给浏览器的「选择文件」对话框加过滤，不是安全校验 -->

import os
from flask import Blueprint, request, jsonify

user_bp = Blueprint("user", __name__, url_prefix="/api/user")


@user_bp.post("/upload")
def upload_demo():
    # ① 表单里没有 file 这个字段时，直接拒绝
    if "file" not in request.files:
        return jsonify(code=400, msg="请选择要上传的文件", data=None)

    file = request.files["file"]

    # ② 用户什么都没选就点了提交，filename 会是空字符串
    if file.filename == "":
        return jsonify(code=400, msg="文件名不能为空", data=None)

    # ③ 落盘（这里先直接存，安全问题下一节处理）
    save_path = os.path.join("static/upload", file.filename)
    file.save(save_path)

    return jsonify(code=200, msg="上传成功", data={"path": "/static/upload/" + file.filename})

import os
from werkzeug.utils import secure_filename

# 接口文档 2.5：头像只允许 jpg / png / gif —— 用白名单
ALLOWED_EXT = {"jpg", "jpeg", "png", "gif"}


def allowed_file(filename: str) -> bool:
    """借助 rsplit 取最后一个点后面的后缀，再判断是否在白名单里。"""
    return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXT


def safe_name(filename: str) -> str:
    """把文件名清洗成安全形式，例如 '../../../etc/passwd' -> 'etc_passwd'。"""
    return secure_filename(filename)

from flask import Flask, jsonify

app = Flask(__name__)

# 限制整个请求体最大 2MB（对齐接口文档 2.5：头像 ≤ 2MB）
app.config["MAX_CONTENT_LENGTH"] = 2 * 1024 * 1024


# 超限时 Flask 会抛 413 异常，捕获它并按统一格式返回
@app.errorhandler(413)
def too_large(e):
    return jsonify(code=400, msg="文件太大，头像不能超过 2MB", data=None)

import os

BASE_DIR = os.path.abspath(os.path.dirname(__file__))


class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-change-me")
    MAX_CONTENT_LENGTH = 2 * 1024 * 1024          # 单个请求最大 2MB

    # 本地上传根目录：放在 static 下，Flask 才能直接通过 /static/... 访问
    # 注意：生产环境要改成独立目录，见第 7 节
    UPLOAD_FOLDER = os.path.join(BASE_DIR, "static", "upload")

    ALLOWED_EXT = {"jpg", "jpeg", "png", "gif"}

import os
import uuid
from datetime import datetime
from flask import current_app


def build_save_path(ext: str, sub_dir: str = "avatar"):
    """返回 (绝对路径, 相对 URL)。按 年/月 分目录，文件名用 uuid 保证唯一。"""
    root = current_app.config["UPLOAD_FOLDER"]      # 从配置取，不写死

    # ① 日期分目录：upload/avatar/2026/10/
    now = datetime.now()
    rel_dir = os.path.join(sub_dir, now.strftime("%Y"), now.strftime("%m"))

    # ② 物理目录如果不存在就建出来（.save() 不会自动建目录）
    abs_dir = os.path.join(root, rel_dir)
    os.makedirs(abs_dir, exist_ok=True)

    # ③ 文件名用 uuid4，彻底避免同名覆盖
    filename = uuid.uuid4().hex + "." + ext

    abs_path = os.path.join(abs_dir, filename)
    # ④ 对外可访问的 URL（/static 前缀来自 Flask 默认静态目录）
    url_path = "/static/upload/" + rel_dir.replace(os.sep, "/") + "/" + filename
    return abs_path, url_path

from flask import current_app, g, jsonify, request
from ..utils.auth import login_required
from ..utils.upload import allowed_file, build_save_path
from ..models import db


@user_bp.post("/avatar")
@login_required                                    # ① 必须登录
def upload_avatar():
    if "file" not in request.files:
        return jsonify(code=400, msg="请选择要上传的图片", data=None)

    file = request.files["file"]
    if file.filename == "":
        return jsonify(code=400, msg="文件名不能为空", data=None)

    # ② 类型白名单（体积已由 MAX_CONTENT_LENGTH 在更前面拦掉）
    if not allowed_file(file.filename):
        return jsonify(code=400, msg="只允许上传 jpg/png/gif 图片", data=None)

    # ③ 用 uuid 生成存储路径，原始文件名只用来取后缀
    ext = file.filename.rsplit(".", 1)[1].lower()
    abs_path, url_path = build_save_path(ext, sub_dir="avatar")
    file.save(abs_path)

    # ④ 更新数据库里该用户的 avatar 字段
    g.user.avatar = url_path
    db.session.commit()

    return jsonify(code=200, msg="上传成功", data={"avatar": url_path})

import os
from flask import Blueprint, current_app, send_from_directory, jsonify

file_bp = Blueprint("file", __name__, url_prefix="/api/file")


@file_bp.get("/download/<path:filename>")
def download(filename):
    """从 UPLOAD_FOLDER 中读取文件并返回（可在此处加登录校验、下载计数）。"""
    folder = current_app.config["UPLOAD_FOLDER"]
    if not os.path.exists(os.path.join(folder, filename)):
        return jsonify(code=404, msg="文件不存在", data=None)
    # as_attachment=True 会让浏览器弹出「下载」而不是直接打开
    return send_from_directory(folder, filename, as_attachment=True)

import io
from flask import make_response, send_file


@file_bp.get("/export")
def export_csv():
    """动态生成 CSV 并作为附件下载（不用先落盘）。"""
    buf = io.BytesIO()
    buf.write("title,author\n静夜思,李白\n春望,杜甫\n".encode("utf-8-sig"))
    buf.seek(0)
    return send_file(
        buf,
        mimetype="text/csv",
        as_attachment=True,             # 强制下载
        download_name="article.csv",    # 下载时显示的文件名
    )


@file_bp.get("/preview/<filename>")
def preview(filename):
    """在线预览：不下载，浏览器内直接打开（图片 / PDF 常用）。"""
    resp = make_response(send_file(filename, mimetype="image/png"))
    # 显式声明为 inline，浏览器会尝试直接渲染而不是弹出保存框
    resp.headers["Content-Disposition"] = "inline; filename=" + filename
    return resp

import os


class Config:
    # 开发机默认放 static 下；生产用环境变量覆盖成独立目录
    UPLOAD_FOLDER = os.environ.get(
        "UPLOAD_FOLDER",
        os.path.join(os.path.dirname(__file__), "static", "upload"),
    )
    MAX_CONTENT_LENGTH = 2 * 1024 * 1024


# 生产环境启动前设置（示例）：
#   export UPLOAD_FOLDER=/data/hiwen/upload

server {
    listen 80;
    server_name example.com;

    # 用户上传的文件由 Nginx 直接返回，不经过 Flask
    location /static/upload/ {
        alias /data/hiwen/upload/;
        # 上传目录不允许执行任何脚本，防止上传 webshell 后被解析
        location ~ \.(php|jsp|py)$ {
            deny all;
        }
    }

    # 其余接口转发给 Flask（gunicorn 监听 127.0.0.1:8000）
    location /api/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}