mmote.ru/content/posts/nokia5110-lcd-connection/main.py

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)