-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathflask_uploader.py
More file actions
80 lines (69 loc) · 2.79 KB
/
Copy pathflask_uploader.py
File metadata and controls
80 lines (69 loc) · 2.79 KB
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
import flask
import json, os
import requests
from datetime import datetime
import hashlib
# docker build -f dockerfile.uploader -t flask-uploader:latest .
app = flask.Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 * 10
ROOT_PATH = "/app/pages"
current_path = os.path.dirname(os.path.abspath(__file__))
with open(os.path.join(current_path, 'UploaderConfig.json'), 'r') as f:
config = json.load(f)
def auth_headers_from_request():
headers = {}
token = flask.request.headers.get('Bearer', None)
if token:
headers["Bearer"] = token
session_id = flask.request.cookies.get('AuthSession', None)
if session_id:
headers["Cookie"] = f"AuthSession={session_id}"
return headers
def api_check_admin(auth_headers: dict):
if not config["check_admin"]:
return True
if not auth_headers:
return False
r = requests.get(
"https://letovocorp.ru/letovo-api/auth/amiuploader",
headers=auth_headers,
verify=False,
timeout=10,
)
return json.loads(r.text)["status"] == 't'
@app.route('/', methods=['POST'])
def upload_file():
if 'file' not in flask.request.files:
return "No file part", 400
file = flask.request.files['file']
if file.filename == '':
return "No selected file", 400
file.filename.replace(" ", "_")
extention = file.filename.split('.')[-1].lower()
file.filename = hashlib.md5(file.filename.encode() + str(datetime.now()).encode()).hexdigest() + "." + extention
if extention in config["supported"]:
file_path = os.path.join(ROOT_PATH, config["paths"][config["supported"][extention]])
else:
file_path = os.path.join(ROOT_PATH, config["paths"]["other"])
auth_headers = auth_headers_from_request()
if not api_check_admin(auth_headers):
return "Not authorized", 403
file.save(os.path.join(file_path, file.filename))
return '{"file": "/' + str(os.path.join(config["paths"][config["supported"][extention]], file.filename)) + '"}'
@app.route('/avatar', methods=['POST'])
def upload_avatar():
if 'file' not in flask.request.files:
return "No file part", 400
file = flask.request.files['file']
if file.filename == '':
return "No selected file", 400
file_path = os.path.join(ROOT_PATH, config["ava_path"])
auth_headers = auth_headers_from_request()
if not api_check_admin(auth_headers):
return "Not authorized", 403
file.filename.replace(" ", "_")
file.filename = str(datetime.now().timestamp()).replace('.', '_') + "_" + file.filename
file.save(os.path.join(file_path, file.filename))
return '{"file": "/' + str(os.path.join(config["ava_path"], file.filename)) + '"}'
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8880, debug=False, threaded=True, use_reloader=False)