134 lines
3.3 KiB
Python
134 lines
3.3 KiB
Python
import zipfile
|
|
import io
|
|
import os
|
|
import time
|
|
from flask_executor import Executor
|
|
import uuid
|
|
|
|
from flask import Flask, render_template, jsonify, redirect, request, send_file, flash
|
|
from flask_login import login_user, logout_user, LoginManager, UserMixin
|
|
from flask_sessionstore import Session
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
from werkzeug.utils import secure_filename
|
|
|
|
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = b'4554bvcj34k34 kl43 n4 v43nlk34nkv5nm34v535'
|
|
app.config['SESSION_TYPE'] = 'filesystem'
|
|
app.config['SESSION_FILE_DIR'] = 'tmp/sessions'
|
|
|
|
admins = {
|
|
# {"username" : "bobby", "password": "40bd001563085fc35165329ea1ff5c5ecbdbbeef"} #sha1
|
|
# {"username" : "bobby", "password": "$2y$12$11VQXE/ouspmtf/9U02BAuRNHwTi0Qzd.HUn4Oh.uEgDKi8vP.Lw2"} #bcrypt
|
|
"bobby": {
|
|
"password_hash": "pbkdf2:sha256:150000$GkUZvh2C$5cf29b24cb7ec99e44984b6f6cf739fd7fd01ff27e84bc08e72291d0532410e1"
|
|
}
|
|
}
|
|
|
|
|
|
|
|
session_store = Session()
|
|
session_store.init_app(app)
|
|
|
|
login_manager = LoginManager()
|
|
login_manager.init_app(app)
|
|
|
|
pool = Executor(app)
|
|
|
|
class AdminUser(UserMixin):
|
|
def __init__(self, user_id):
|
|
self.id = user_id
|
|
|
|
|
|
def worker_done(future):
|
|
print("done")
|
|
|
|
pool.add_default_done_callback(worker_done)
|
|
|
|
def worker():
|
|
for i in range(1, 5):
|
|
print("working")
|
|
time.sleep(1)
|
|
# print("finished")
|
|
|
|
@app.route("/work")
|
|
def work():
|
|
id = uuid.uuid1()
|
|
print("new job: %s" % id)
|
|
pool.submit_stored(id, worker)
|
|
return redirect('/')
|
|
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
if user_id in admins:
|
|
return AdminUser(user_id)
|
|
return None
|
|
|
|
@app.route("/", methods=['GET', 'POST'])
|
|
def index():
|
|
if request.method == 'POST':
|
|
files = request.files.getlist("up[]")
|
|
for f in files:
|
|
if f.filename:
|
|
filename = secure_filename(f.filename)
|
|
f.save(os.path.join(app.root_path, "tmp", filename))
|
|
|
|
done = False
|
|
|
|
if pool.futures.done('worker1'):
|
|
pool.futures.pop('worker1')
|
|
done = True
|
|
|
|
return render_template("index.jinja", thread_done=done)
|
|
|
|
|
|
@app.route("/zip")
|
|
def zipf():
|
|
# membytes = io.BytesIO()
|
|
membytes = io.BytesIO()
|
|
zip_obj = zipfile.ZipFile(membytes, "w")
|
|
zip_obj.writestr("a.txt", "hello, world")
|
|
zip_obj.close()
|
|
|
|
membytes.seek(0)
|
|
return send_file(membytes, mimetype='application/zip', as_attachment=True, attachment_filename="zip.zip")
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if request.method == 'POST':
|
|
username = request.form.get('username')
|
|
password = request.form.get('password')
|
|
if username in admins:
|
|
userdata = admins[username]
|
|
if check_password_hash(userdata["password_hash"], password):
|
|
u = AdminUser(username)
|
|
login_user(u)
|
|
return redirect('/')
|
|
else:
|
|
print("wrong password")
|
|
else:
|
|
print("user not found")
|
|
|
|
return redirect('/login')
|
|
else:
|
|
return render_template("login.jinja")
|
|
# u = AdminUser("bobby")
|
|
# login_user(u)
|
|
return redirect('/')
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
logout_user()
|
|
return redirect('/')
|
|
|
|
@app.route("/api", methods=['GET'])
|
|
def api():
|
|
j = {
|
|
"username": "Bobby"
|
|
}
|
|
return jsonify(j)
|
|
|