1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
"""Module for generic files endpoints."""
import traceback
from pathlib import Path
from flask import request, jsonify, Blueprint, current_app as app
from .chunks import chunk_name, chunks_directory
files = Blueprint("files", __name__)
def target_file(fileid: str) -> Path:
"""Compute the full path for the target file."""
return Path(app.config["UPLOAD_FOLDER"], fileid)
@files.route("/upload/resumable", methods=["GET"])
def resumable_upload_get():
"""Used for checking whether **ALL** chunks have been uploaded."""
fileid = request.args.get("resumableIdentifier", type=str) or ""
filename = request.args.get("resumableFilename", type=str) or ""
chunk = request.args.get("resumableChunkNumber", type=int) or 0
if not(fileid or filename or chunk):
return jsonify({
"message": "At least one required query parameter is missing.",
"error": "BadRequest",
"statuscode": 400
}), 400
# If the complete target file exists, return 200 for all chunks.
_targetfile = target_file(fileid)
if _targetfile.exists():
return jsonify({
"uploaded-file": _targetfile.name,
"original-name": filename,
"chunk": chunk,
"message": "The complete file already exists.",
"statuscode": 200
}), 200
if Path(chunks_directory(fileid),
chunk_name(filename, chunk)).exists():
return jsonify({
"chunk": chunk,
"message": f"Chunk {chunk} exists.",
"statuscode": 200
}), 200
return jsonify({
"message": f"Chunk {chunk} was not found.",
"error": "NotFound",
"statuscode": 404
}), 404
def __merge_chunks__(targetfile: Path, chunkpaths: tuple[Path, ...]) -> Path:
"""Merge the chunks into a single file."""
with open(targetfile, "ab") as _target:
for chunkfile in chunkpaths:
with open(chunkfile, "rb") as _chunkdata:
_target.write(_chunkdata.read())
chunkfile.unlink()
return targetfile
@files.route("/upload/resumable", methods=["POST"])
def resumable_upload_post():
"""Do the actual chunks upload here."""
_totalchunks = request.form.get("resumableTotalChunks", type=int) or 0
_chunk = request.form.get("resumableChunkNumber", default=1, type=int)
_uploadfilename = request.form.get(
"resumableFilename", default="", type=str) or ""
_fileid = request.form.get(
"resumableIdentifier", default="", type=str) or ""
_targetfile = target_file(_fileid)
if _targetfile.exists():
return jsonify({
"uploaded-file": _targetfile.name,
"original-name": _uploadfilename,
"message": "File was uploaded successfully!",
"statuscode": 200
}), 200
try:
chunks_directory(_fileid).mkdir(exist_ok=True, parents=True)
request.files["file"].save(Path(chunks_directory(_fileid),
chunk_name(_uploadfilename, _chunk)))
# Check whether upload is complete
chunkpaths = tuple(
Path(chunks_directory(_fileid), chunk_name(_uploadfilename, _achunk))
for _achunk in range(1, _totalchunks+1))
if all(_file.exists() for _file in chunkpaths):
# merge_files and clean up chunks
__merge_chunks__(_targetfile, chunkpaths)
chunks_directory(_fileid).rmdir()
return jsonify({
"uploaded-file": _targetfile.name,
"original-name": _uploadfilename,
"message": "File was uploaded successfully!",
"statuscode": 200
}), 200
return jsonify({
"message": f"Chunk {int(_chunk)} uploaded successfully.",
"statuscode": 201
}), 201
except Exception as exc:# pylint: disable=[broad-except]
msg = "Error processing uploaded file chunks."
app.logger.error(msg, exc_info=True, stack_info=True)
return jsonify({
"message": msg,
"error": type(exc).__name__,
"error-description": " ".join(str(arg) for arg in exc.args),
"error-trace": traceback.format_exception(exc)
}), 500
|
