move to package structure
Andronaco Marco marco.andronaco@olivetti.com
Mon, 17 Jul 2023 23:40:23 +0200
11 files changed,
249 insertions(+),
197 deletions(-)
A
myprecious/Auth.py
@@ -0,0 +1,55 @@
+import myprecious.Constants as c +from myprecious.Db import add_user_to_queue, get_user_from_username, get_user_from_id +from flask_login import UserMixin + +class User(UserMixin): + def __init__(self, user_id, username, password, email): + self.id = user_id + self.username = username + self.password = password + self.email = email + +def construct_user(id, username, password, email): + try: + return User(int(id), username, password, email) + except TypeError: + return None + +def handle_register(form): + username = form["username"].lower() + email = form["email"].lower() + password = form["password"] + + if len(password) < c.MIN_PW_LENGTH or len(username) < c.MIN_USERNAME_LENGTH: + return "Your username or password is too short." + + if len(password) > c.MAX_LENGTH or len(username) > c.MAX_LENGTH: + return "Your username or password is too long." + + res = add_user_to_queue(username, password, email) + if res == None: + return "This username is already registered." + return None + +def handle_login(form): + username = form["username"].lower() + password = form["password"] + + r = get_user_from_username(username) + if r is None: + return "That account does not exist.", { "username": username } + user = construct_user(r[0], r[1], r[2], r[3]) + + last_user = { "username": username } + if user is None: + return "Parsing error.", last_user + + if user.password == password: + return None, user + return "Wrong password.", last_user + +def get_logged_user(user_id): + lu = get_user_from_id(user_id) + if lu is None: + return None + return construct_user(lu[0], lu[1], lu[2], lu[3])
M
myprecious/Db.py
→
myprecious/Db.py
@@ -1,6 +1,6 @@
import uuid, hashlib, sqlite3 -import Constants as c +import myprecious.Constants as c def hash(password: str): salt = uuid.uuid4().hex
A
myprecious/Encoding.py
@@ -0,0 +1,13 @@
+import base64, json + +def obj_decode(text): + base64_bytes = text.encode('utf-8') + message_bytes = base64.b64decode(base64_bytes) + message = message_bytes.decode('utf-8') + return json.loads(message) + +def obj_encode(info): + temp_json = json.dumps(info) + temp_bytes = temp_json.encode("utf-8") + temp_base64 = base64.b64encode(temp_bytes) + return temp_base64.decode("utf-8")
A
myprecious/Utils.py
@@ -0,0 +1,30 @@
+from myprecious.Encoding import obj_encode +import myprecious.Constants as c + +def handle_platform(game, platform): + try: + game_cover = "https:" + game["cover"]["url"] + except KeyError: + game_cover = c.MISSING_COVER_URL + temp_obj = { + "game_id": game["id"], + "platform_id": platform["id"], + "cover": game_cover, + "title": game["name"], + "platform": platform["name"] + } + temp_obj["info"] = obj_encode(temp_obj) + return temp_obj + +def handle_response(response): + games = [ [ handle_platform(game, platform) for platform in game["platforms"] ] for game in response ] + return collapse_list_of_lists(games) + +def collapse_list_of_lists(l): + return [ item for sublist in l for item in sublist ] + +def parse_remember(form): + try: + return bool(form["remember"]) + except KeyError: + return False
M
myprecious/__init__.py
→
myprecious/__init__.py
@@ -0,0 +1,14 @@
+from flask import Flask +from flask_login import LoginManager +import myprecious.Constants as c +from myprecious.Db import init_db + +app = Flask(__name__) +login_manager = LoginManager(app) + +app.debug=c.DEBUG_SWITCH +app.secret_key = c.SECRET_KEY +app.config['SESSION_TYPE'] = 'filesystem' +init_db() + +import myprecious.views
D
myprecious/__main__.py
@@ -1,196 +0,0 @@
-from flask import Flask, request, render_template, url_for, redirect -from flask_login import LoginManager, UserMixin, login_user, logout_user, current_user -from werkzeug.utils import secure_filename -from contextlib import suppress -from Db import init_db, get_user_from_username, get_user_from_id, add_user_to_queue -import os, json, base64 -import Constants as c -if c.DEBUG_SWITCH: - from GamesApiTest import search_game -else: - from GamesApi import search_game - -app = Flask(__name__) -login_manager = LoginManager(app) - -class User(UserMixin): - def __init__(self, user_id, username, password, email): - self.id = user_id - self.username = username - self.password = password - self.email = email - -def construct_user(id, username, password, email): - try: - return User(int(id), username, password, email) - except TypeError: - return None - -def render(template): - return render_template(template, user=current_user) - -def handle_platform(game, platform): - try: - game_cover = "https:" + game["cover"]["url"] - except KeyError: - game_cover = c.MISSING_COVER_URL - temp_obj = { - "game_id": game["id"], - "platform_id": platform["id"], - "cover": game_cover, - "title": game["name"], - "platform": platform["name"] - } - temp_json = json.dumps(temp_obj) - temp_bytes = temp_json.encode("utf-8") - temp_base64 = base64.b64encode(temp_bytes) - temp_obj["info"] = temp_base64.decode("utf-8") - return temp_obj - -def handle_response(response): - return [ [ handle_platform(game, platform) for platform in game["platforms"] ] for game in response ] - -def collapse_list_of_lists(l): - return [ item for sublist in l for item in sublist ] - -@login_manager.user_loader -def load_user(user_id): - lu = get_user_from_id(user_id) - if lu is None: - return None - return construct_user(lu[0], lu[1], lu[2], lu[3]) - -@app.route('/') -def route_index(): - return render("index.html") - -@app.route('/login', methods=['GET', 'POST']) -def route_login(): - if current_user.is_authenticated: - return redirect('/') - - if request.method == "GET": - return render("login.html") - - form = request.form - username = form["username"].lower() - password = form["password"] - try: - remember = bool(form["remember"]) - except KeyError: - remember = False - - r = get_user_from_username(username) - if r is None: - return render_template("login.html", user=current_user, last_user=username) - user = construct_user(r[0], r[1], r[2], r[3]) - - if user is None: - return redirect(url_for("login")) - - if user.password == password: - login_user(user, remember=remember) - return redirect("/") - else: - return render_template("login.html", user=current_user, last_user=username) - -@app.route('/register', methods=['GET', 'POST']) -def route_register(): - if current_user.is_authenticated: - return redirect('/') - - if request.method == "GET": - return render("register.html") - - form = request.form - username = form["username"].lower() - email = form["email"].lower() - password = form["password"] - - if len(password) < c.MIN_PW_LENGTH or len(username) < c.MIN_USERNAME_LENGTH: - return render_template("register.html", user=current_user, error="Your username or password is too short.") - - if len(password) > c.MAX_LENGTH or len(username) > c.MAX_LENGTH: - return render_template("register.html", user=current_user, error="Your username or password is too long.") - - res = add_user_to_queue(username, password, email) - if res == None: - return render_template("register.html", user=current_user, error="This username is already registered.") - return render("register_done.html") - -@app.route('/logout') -def route_logout(): - logout_user() - return redirect("/") - -@app.route('/search', methods=['GET', 'POST']) -def route_search(): - if not current_user.is_authenticated: - return redirect('/login') - if request.method == 'GET': - return render("search.html") - - query = request.form["query"] - search_response = search_game(query) - - games = collapse_list_of_lists(handle_response(search_response)) - return render_template("search.html", user=current_user, games=games, query=query) - -@app.route('/upload', methods=['GET', 'POST']) -def route_upload(): - if not current_user.is_authenticated: - return redirect('/login') - if request.method == 'GET': - info = request.args.get("info") - if info is None: - return render_template("upload.html", user=current_user, game=c.NO_GAME) - # info = base64 - base64_bytes = info.encode('utf-8') - message_bytes = base64.b64decode(base64_bytes) - message = message_bytes.decode('utf-8') - game = json.loads(message) - return render_template("upload.html", user=current_user, game=game) - - f = request.files['file'] - try: - game_id = int(request.form['game_id']) - platform_id = int(request.form['platform_id']) - except ValueError: - return redirect("/upload") - - # TODO: use IGDB api to validate game_id, platform_id and title before adding - # TODO: save game in DB - - save_folder = os.path.join(c.BASE_DIRECTORY, c.CONTENT_DIRECTORY, str(current_user.id), str(game_id), str(platform_id)) - with suppress(FileExistsError): - os.makedirs(save_folder) - if f.filename is None: - return redirect("/upload") - save_file = os.path.join(save_folder, secure_filename(f.filename)) - f.save(save_file) - return render("index.html") - - -@app.route('/admin', methods=['GET', 'POST']) -def route_admin(): - if not current_user.is_authenticated: - return redirect('/') - - if current_user.id != 1: - return redirect('/') - - if request.method == "GET": - return render("admin.html") - - return render("admin.html") - -@app.route('/about') -def route_about(): - return render("about.html") - -if __name__ == "__main__": - app.debug=c.DEBUG_SWITCH - app.secret_key = c.SECRET_KEY - app.config['SESSION_TYPE'] = 'filesystem' - init_db() - app.run(port=1111)
M
myprecious/static/style.css
→
myprecious/static/style.css
@@ -45,3 +45,7 @@
.hidden { display: none !important; } + +.error { + color: lightcoral; +}
M
myprecious/templates/login.html
→
myprecious/templates/login.html
@@ -8,4 +8,7 @@ <input type="checkbox" name="remember" checked>
<label for="remember">remember me</label><br /> <input type="submit" value="login" /> </form> +{% if error %} +<p class="error">{{ error }}</p> +{% endif %} {% endblock %}
A
myprecious/views.py
@@ -0,0 +1,117 @@
+from myprecious import app, login_manager +from flask import request, redirect, render_template +from flask_login import login_user, logout_user, current_user +from werkzeug.utils import secure_filename +from contextlib import suppress +import myprecious.Constants as c +from myprecious.Utils import handle_response, parse_remember +from myprecious.Auth import handle_register, handle_login, get_logged_user +from myprecious.Encoding import obj_decode +import os +if c.DEBUG_SWITCH: + from myprecious.GamesApiTest import search_game +else: + from myprecious.GamesApi import search_game + +def render(template, **context): + return render_template(template, user=current_user, **context) + +@login_manager.user_loader +def load_user(user_id): + return get_logged_user(user_id) + +@app.route('/') +def route_index(): + return render("index.html") + +@app.route('/login', methods=['GET', 'POST']) +def route_login(): + if current_user.is_authenticated: + return redirect('/') + if request.method == "GET": + return render("login.html") + + form = request.form + remember = parse_remember(form) + + error, user = handle_login(request.form) + if error is None: + login_user(user, remember=remember) + return redirect("/") + last_user = user.username if user else None + return render("login.html", last_user=last_user, error=error) + + +@app.route('/register', methods=['GET', 'POST']) +def route_register(): + if current_user.is_authenticated: + return redirect('/') + if request.method == "GET": + return render("register.html") + error = handle_register(request.form) + if error is None: + return render("register_done.html") + return render("register.html", error=error) + +@app.route('/logout') +def route_logout(): + logout_user() + return redirect("/") + +@app.route('/search', methods=['GET', 'POST']) +def route_search(): + if not current_user.is_authenticated: + return redirect('/login') + if request.method == 'GET': + return render("search.html") + query = request.form["query"] + search_response = search_game(query) + return render("search.html", games=handle_response(search_response), query=query) + +@app.route('/upload', methods=['GET', 'POST']) +def route_upload(): + if not current_user.is_authenticated: + return redirect('/login') + if request.method == 'GET': + info = request.args.get("info") + if info is None: + return render("upload.html", game=c.NO_GAME) + game = obj_decode(info) + return render("upload.html", game=game) + + f = request.files['file'] + try: + game_id = int(request.form['game_id']) + platform_id = int(request.form['platform_id']) + except ValueError: + return redirect("/upload") + + # TODO: use IGDB api to validate game_id, platform_id and title before adding + # TODO: save game in DB + + save_folder = os.path.join(c.BASE_DIRECTORY, c.CONTENT_DIRECTORY, str(current_user.id), str(game_id), str(platform_id)) + with suppress(FileExistsError): + os.makedirs(save_folder) + if f.filename is None: + return redirect("/upload") + save_file = os.path.join(save_folder, secure_filename(f.filename)) + f.save(save_file) + return render("index.html") + + +@app.route('/admin', methods=['GET', 'POST']) +def route_admin(): + if not current_user.is_authenticated: + return redirect('/') + + if current_user.id != 1: + return redirect('/') + + if request.method == "GET": + return render("admin.html") + + return render("admin.html") + +@app.route('/about') +def route_about(): + return render("about.html")