diff options
author | lonkaars <l.leblansch@gmail.com> | 2021-04-11 17:50:58 +0200 |
---|---|---|
committer | lonkaars <l.leblansch@gmail.com> | 2021-04-11 17:50:58 +0200 |
commit | 28f104de9ae9abe4b42abafbf3865ede5687996c (patch) | |
tree | 65e651f09d8fbf81380384692e45803cb4f9d61c | |
parent | 7b4859059b3bbabf4139ccdf3270a82c094f5d8e (diff) |
dprint yapf python formatting
39 files changed, 1385 insertions, 1027 deletions
diff --git a/api/auth/login.py b/api/auth/login.py index 78a9add..4ae1650 100644 --- a/api/auth/login.py +++ b/api/auth/login.py @@ -5,38 +5,50 @@ import passwords login = Blueprint('login', __name__) -@login.route('/login', methods = ['POST']) -def index(): - data = request.get_json() - - # get form data - email = data.get("email") or "" - password = data.get("password") or "" - - # return malformed request if email or password is missing - if not email or \ - not password: - return "", 400 - - # resolve user_id from username or email - user_id = None - user_id = user_id or cursor.execute("select user_id from users where email = ?", [email]).fetchone() - user_id = user_id or cursor.execute("select user_id from users where lower(username) = lower(?)", [email]).fetchone() - if user_id == None: return "", 401 - # check the password - passwd = cursor.execute("select password_hash from users where user_id = ?", [user_id[0]]).fetchone() - check = passwords.check_password(password, passwd[0]) - if not check: return "", 401 - - # generate a new authentication token and add it to the users valid token list - new_token = token.generate_token() - token.add_token(user_id[0], token.hash_token(new_token)) - - # make response with the set_cookie header - res = make_response("", 200) - res.set_cookie("token", new_token["token"], expires = int(new_token["expirationDate"] / 1000)) +@login.route('/login', methods=['POST']) +def index(): + data = request.get_json() + + # get form data + email = data.get("email") or "" + password = data.get("password") or "" + + # return malformed request if email or password is missing + if not email or \ + not password: + return "", 400 + + # resolve user_id from username or email + user_id = None + user_id = user_id or cursor.execute( + "select user_id from users where email = ?", [email] + ).fetchone() + user_id = user_id or cursor.execute( + "select user_id from users where lower(username) = lower(?)", [email] + ).fetchone() + if user_id == None: return "", 401 + + # check the password + passwd = cursor.execute( + "select password_hash from users where user_id = ?", [user_id[0]] + ).fetchone() + check = passwords.check_password(password, passwd[0]) + if not check: return "", 401 + + # generate a new authentication token and add it to the users valid token list + new_token = token.generate_token() + token.add_token(user_id[0], token.hash_token(new_token)) + + # make response with the set_cookie header + res = make_response("", 200) + res.set_cookie( + "token", + new_token["token"], + expires=int(new_token["expirationDate"] / 1000) + ) + + return res - return res dynamic_route = ["/auth", login] diff --git a/api/auth/login_token.py b/api/auth/login_token.py index d920eea..bb67c4f 100644 --- a/api/auth/login_token.py +++ b/api/auth/login_token.py @@ -2,22 +2,29 @@ from flask import Blueprint, request from db import cursor from auth.token import validate_token, hash_token + # get user_id from authentication token def token_login(token): - hashed = hash_token({ "token": token, "expirationDate": 0 }) - user_id = cursor.execute("select user_id from users where valid_tokens like ?", [f"%{hashed['token']}%"]).fetchone() - return None if not user_id else user_id[0] + hashed = hash_token({"token": token, "expirationDate": 0}) + user_id = cursor.execute( + "select user_id from users where valid_tokens like ?", + [f"%{hashed['token']}%"] + ).fetchone() + return None if not user_id else user_id[0] + token = Blueprint('token', __name__) + # this endpoint is currently unused, but verifies that a token is valid -@token.route('/token', methods = ['POST']) +@token.route('/token', methods=['POST']) def index(): - data = request.get_json() + data = request.get_json() + + auth_token = data.get("token") or "" + if not auth_token: return "", 400 - auth_token = data.get("token") or "" - if not auth_token: return "", 400 + return "", 200 if token_login(auth_token) else 401 - return "", 200 if token_login(auth_token) else 401 dynamic_route = ["/auth", token] diff --git a/api/auth/signup.py b/api/auth/signup.py index e758a4e..f9a1af5 100644 --- a/api/auth/signup.py +++ b/api/auth/signup.py @@ -6,71 +6,87 @@ import passwords import time import re + # checks if the usename is between 3 and 35 charachters def validate_username(username): - return len(username) in range(3, 35 + 1) + return len(username) in range(3, 35 + 1) + # garbage email validation (see todo) def validate_email(email): - #TODO: use node_modules/email-validator/index.js - return len(email) > 1 and \ - "@" in email + #TODO: use node_modules/email-validator/index.js + return len(email) > 1 and \ + "@" in email + # checks if the password is safe (regex explanation in pages/register.tsx) def validate_password(password): - passwordRegex = r"^(?=.*[A-Z])(?=.*[a-z])(?=.*[0-9]).{8,}$" # r"" = raw string - return re.match(passwordRegex, password) + passwordRegex = r"^(?=.*[A-Z])(?=.*[a-z])(?=.*[0-9]).{8,}$" # r"" = raw string + return re.match(passwordRegex, password) + signup = Blueprint('signup', __name__) -@signup.route('/signup', methods = ['POST']) + +@signup.route('/signup', methods=['POST']) def index(): - # parse request data if the content-type header is set to application/json - data = request.get_json() - - # force string if {}.get(...) returns None - username = data.get("username") or "" - email = data.get("email") or "" - password = data.get("password") or "" - - # return 400 (malformed request) if any of the required data is missing - if not username or \ - not email or \ - not password: - return "", 400 - - # return 403 (forbidden) if any of the required data is invalid - if not validate_username(username) or \ - not validate_email(email) or \ - not validate_password(password): - return {"error": "form_data_invalid"}, 403 - - # check if username is taken - if cursor.execute("select username from users where lower(username) = lower(?)", [username]).fetchone(): - return {"error": "username_taken"}, 403 - - # check if email is taken - if cursor.execute("select email from users where email = ?", [email]).fetchone(): - return {"error": "email_taken"}, 403 - - # create new user_id, hash password and note timestamp - user_id = new_uuid("users") - password_hash = passwords.password_hash(password) - registered = int( time.time() * 1000 ) - - # write new user to database and commit - cursor.execute("insert into users values (?, ?, ?, NULL, NULL, ?, ?, \"[]\", FALSE, \"user\", \"{}\", \"online\") ", - (user_id, username, email, password_hash, registered)) - connection.commit() - - # create a new token for the user to use for authentication - new_token = token.generate_token() - token.add_token(user_id, token.hash_token(new_token)) - - # create a flask response object to add the set-cookie header to - res = make_response("", 200) - res.set_cookie("token", new_token["token"], expires = int(new_token["expirationDate"] / 1000)) - - return res + # parse request data if the content-type header is set to application/json + data = request.get_json() + + # force string if {}.get(...) returns None + username = data.get("username") or "" + email = data.get("email") or "" + password = data.get("password") or "" + + # return 400 (malformed request) if any of the required data is missing + if not username or \ + not email or \ + not password: + return "", 400 + + # return 403 (forbidden) if any of the required data is invalid + if not validate_username(username) or \ + not validate_email(email) or \ + not validate_password(password): + return {"error": "form_data_invalid"}, 403 + + # check if username is taken + if cursor.execute( + "select username from users where lower(username) = lower(?)", + [username] + ).fetchone(): + return {"error": "username_taken"}, 403 + + # check if email is taken + if cursor.execute("select email from users where email = ?", + [email]).fetchone(): + return {"error": "email_taken"}, 403 + + # create new user_id, hash password and note timestamp + user_id = new_uuid("users") + password_hash = passwords.password_hash(password) + registered = int(time.time() * 1000) + + # write new user to database and commit + cursor.execute( + "insert into users values (?, ?, ?, NULL, NULL, ?, ?, \"[]\", FALSE, \"user\", \"{}\", \"online\") ", + (user_id, username, email, password_hash, registered) + ) + connection.commit() + + # create a new token for the user to use for authentication + new_token = token.generate_token() + token.add_token(user_id, token.hash_token(new_token)) + + # create a flask response object to add the set-cookie header to + res = make_response("", 200) + res.set_cookie( + "token", + new_token["token"], + expires=int(new_token["expirationDate"] / 1000) + ) + + return res + dynamic_route = ["/auth", signup] diff --git a/api/auth/token.py b/api/auth/token.py index 113c2c7..d75c91b 100644 --- a/api/auth/token.py +++ b/api/auth/token.py @@ -4,37 +4,57 @@ import secrets import json import time + # get valid token hashes for a given user_id def valid_tokens(user_id): - tokens = json.loads(cursor.execute("select valid_tokens from users where user_id = ?", [user_id]).fetchone()[0]) - # return only tokens that aren't expired - return [token for token in tokens if token["expirationDate"] > int( time.time() * 1000 )] + tokens = json.loads( + cursor.execute( + "select valid_tokens from users where user_id = ?", [user_id] + ).fetchone()[0] + ) + # return only tokens that aren't expired + return [ + token for token in tokens + if token["expirationDate"] > int(time.time() * 1000) + ] + def validate_token(user_id, token): - tokens = valid_tokens(user_id) - return hashlib.sha256(str(token).encode()).hexdigest() in [ t["token"] for t in tokens if t["expirationDate"] > int( time.time() * 1000 ) ] + tokens = valid_tokens(user_id) + return hashlib.sha256(str(token).encode()).hexdigest() in [ + t["token"] for t in tokens + if t["expirationDate"] > int(time.time() * 1000) + ] + def modify_tokens(user_id, formatted_token, remove): - temp_tokens = valid_tokens(user_id) - temp_tokens.remove(formatted_token) if remove else temp_tokens.append(formatted_token) - cursor.execute("update users set valid_tokens = ? where user_id = ?", [json.dumps(temp_tokens), user_id]) - connection.commit() + temp_tokens = valid_tokens(user_id) + temp_tokens.remove(formatted_token + ) if remove else temp_tokens.append(formatted_token) + cursor.execute( + "update users set valid_tokens = ? where user_id = ?", + [json.dumps(temp_tokens), user_id] + ) + connection.commit() + def add_token(user_id, formatted_token): - modify_tokens(user_id, formatted_token, False) + modify_tokens(user_id, formatted_token, False) + def revoke_token(user_id, formatted_token): - modify_tokens(user_id, formatted_token, True) + modify_tokens(user_id, formatted_token, True) + def hash_token(token): - return { - "token": hashlib.sha256(str(token["token"]).encode()).hexdigest(), - "expirationDate": token["expirationDate"] - } + return { + "token": hashlib.sha256(str(token["token"]).encode()).hexdigest(), + "expirationDate": token["expirationDate"] + } -def generate_token(): - return { - "token": secrets.token_hex(128), - "expirationDate": int( time.time() * 1000 ) + ( 24 * 60 * 60 * 1000 ) - } +def generate_token(): + return { + "token": secrets.token_hex(128), + "expirationDate": int(time.time() * 1000) + (24 * 60 * 60 * 1000) + } @@ -12,4 +12,3 @@ connection.load_extension("./database/levenshtein.sqlext") # sql cursor cursor = connection.cursor() - diff --git a/api/dynamic_import.py b/api/dynamic_import.py index 159950c..5ce9b6b 100644 --- a/api/dynamic_import.py +++ b/api/dynamic_import.py @@ -8,24 +8,25 @@ import glob # get all python files in api/ directory and convert them to python import names files = glob.glob(os.path.dirname(__file__) + "/**/*.py", recursive=True) files.remove(__file__) -files = [str(filename) - .replace(os.path.dirname(__file__) + "/", '') - .replace("/", ".") - .replace(".py", '') - for filename in files] +files = [ + str(filename).replace(os.path.dirname(__file__) + "/", + '').replace("/", ".").replace(".py", '') + for filename in files +] + def route(dynamic_route): - app.register_blueprint(dynamic_route[1], url_prefix=dynamic_route[0]) - path = (dynamic_route[0] + "/" + dynamic_route[1].name).replace('//', '/') - log.info(f"dynamically routing {path}") + app.register_blueprint(dynamic_route[1], url_prefix=dynamic_route[0]) + path = (dynamic_route[0] + "/" + dynamic_route[1].name).replace('//', '/') + log.info(f"dynamically routing {path}") -for file in files: - mod = importlib.import_module(file) - # check if module has `dynamic_route` defined (single route) - if hasattr(mod, "dynamic_route"): - route(mod.dynamic_route) - # check if module has `dynamic_routes` defined (multiple routes as list) - elif hasattr(mod, "dynamic_routes"): - for dynamic_route in mod.dynamic_routes: - route(dynamic_route) +for file in files: + mod = importlib.import_module(file) + # check if module has `dynamic_route` defined (single route) + if hasattr(mod, "dynamic_route"): + route(mod.dynamic_route) + # check if module has `dynamic_routes` defined (multiple routes as list) + elif hasattr(mod, "dynamic_routes"): + for dynamic_route in mod.dynamic_routes: + route(dynamic_route) diff --git a/api/events.py b/api/events.py index ef3ae12..150f403 100644 --- a/api/events.py +++ b/api/events.py @@ -7,27 +7,28 @@ from http import cookies from game.cleanup import set_interval import time + # get token from flask_socketio's request.environment def get_token(environ): - cookie = environ.get("HTTP_COOKIE") - if not cookie: return None + cookie = environ.get("HTTP_COOKIE") + if not cookie: return None + + parsed = cookies.SimpleCookie() + parsed.load(cookie) - parsed = cookies.SimpleCookie() - parsed.load(cookie) + token = parsed.get("token") + if not token: return None - token = parsed.get("token") - if not token: return None + return token.value - return token.value # global socket connection @io.on("connect") def connect(): - token = get_token(request.environ) - if not token: return - - user_id = token_login(token) - if not user_id: return + token = get_token(request.environ) + if not token: return - join_room("user-" + user_id) + user_id = token_login(token) + if not user_id: return + join_room("user-" + user_id) diff --git a/api/game/accept.py b/api/game/accept.py index 073f422..a231d3a 100644 --- a/api/game/accept.py +++ b/api/game/accept.py @@ -12,19 +12,18 @@ from game.new import start_game join_game = Blueprint('game_accept', __name__) + # join a game by game_id (public or private) -@join_game.route('/accept', methods = ['POST']) +@join_game.route('/accept', methods=['POST']) @auth_required("user") def index(game_id): - if cursor.execute("select status from games where game_id = ?", [game_id]).fetchone()[0] != "wait_for_opponent": - return "", 403 + if cursor.execute("select status from games where game_id = ?", + [game_id]).fetchone()[0] != "wait_for_opponent": + return "", 403 + + start_game(game_id, user_id) - start_game(game_id, user_id) + return {"id": game_id, "player_1": False, "game_started": True}, 200 - return { - "id": game_id, - "player_1": False, - "game_started": True - }, 200 dynamic_route = ["/game", join_game] diff --git a/api/game/cleanup.py b/api/game/cleanup.py index 0a9aa46..5b705d2 100644 --- a/api/game/cleanup.py +++ b/api/game/cleanup.py @@ -2,22 +2,30 @@ from db import cursor, connection import threading import time + # cleanup function that's ran every five minutes def cleanup(): - now = int( time.time() * 1000 ) - old_games = cursor.execute("select game_id from games where (status = \"wait_for_opponent\" or status = \"in_progress\") and last_activity < ?", [now - 5 * 60 * 1e3]).fetchall() - for game_id in old_games: - cursor.execute("delete from games where game_id = ?", [game_id[0]]) - connection.commit() + now = int(time.time() * 1000) + old_games = cursor.execute( + "select game_id from games where (status = \"wait_for_opponent\" or status = \"in_progress\") and last_activity < ?", + [now - 5 * 60 * 1e3] + ).fetchall() + for game_id in old_games: + cursor.execute("delete from games where game_id = ?", [game_id[0]]) + connection.commit() + + +def set_interval( + func, sec +): # https://stackoverflow.com/questions/2697039/python-equivalent-of-setinterval + def func_wrapper(): + set_interval(func, sec) + func() + + t = threading.Timer(sec, func_wrapper) + t.start() + return t -def set_interval(func, sec): # https://stackoverflow.com/questions/2697039/python-equivalent-of-setinterval - def func_wrapper(): - set_interval(func, sec) - func() - t = threading.Timer(sec, func_wrapper) - t.start() - return t # run every five minutes set_interval(cleanup, 5 * 60) - diff --git a/api/game/info.py b/api/game/info.py index 76a3ef8..869a19c 100644 --- a/api/game/info.py +++ b/api/game/info.py @@ -5,69 +5,83 @@ from user.info import format_user from rating import outcome from ruleset import resolve_ruleset -def format_game(game_id, user_id = None): - game = cursor.execute("select " + ", ".join([ - "game_id", # 0 - "parent_game", # 1 - "moves", # 2 - "player_1_id", # 3 - "player_2_id", # 4 - "outcome", # 5 - "created", # 6 - "started", # 7 - "duration", # 8 - "rating_delta_player_1", # 9 - "rating_delta_player_2", # 10 - "ruleset", # 11 - "status", # 12 - "private", # 13 - ]) + " from games where game_id = ?", [game_id]).fetchone() - - is_player_1 = game[4] != user_id - - # get opponent from perspective of `user_id` - opponent = game[4] if is_player_1 else game[3] - - # parse moves into list and return empty list if moves string is empty - moves = [] if len(game[2]) == 0 else [int(move) for move in str(game[2] + "0").split(",")] - - return { - "id": game[0], - "parent": game[1], - "moves": moves, - "opponent": None if not opponent else format_user(opponent), - "outcome": None if not game[5] else outcome(game[5], is_player_1), - "created": game[6], - "started": game[7], - "duration": game[8], - "rating": game[9] if is_player_1 else game[10], - "rating_opponent": game[10] if is_player_1 else game[9], - "ruleset": resolve_ruleset(game[11]), - "status": game[12], - "private": bool(game[13]), - } + +def format_game(game_id, user_id=None): + game = cursor.execute( + "select " + ", ".join( + [ + "game_id", # 0 + "parent_game", # 1 + "moves", # 2 + "player_1_id", # 3 + "player_2_id", # 4 + "outcome", # 5 + "created", # 6 + "started", # 7 + "duration", # 8 + "rating_delta_player_1", # 9 + "rating_delta_player_2", # 10 + "ruleset", # 11 + "status", # 12 + "private", # 13 + ] + ) + " from games where game_id = ?", + [game_id] + ).fetchone() + + is_player_1 = game[4] != user_id + + # get opponent from perspective of `user_id` + opponent = game[4] if is_player_1 else game[3] + + # parse moves into list and return empty list if moves string is empty + moves = [] if len(game[2]) == 0 else [ + int(move) for move in str(game[2] + "0").split(",") + ] + + return { + "id": game[0], + "parent": game[1], + "moves": moves, + "opponent": None if not opponent else format_user(opponent), + "outcome": None if not game[5] else outcome(game[5], is_player_1), + "created": game[6], + "started": game[7], + "duration": game[8], + "rating": game[9] if is_player_1 else game[10], + "rating_opponent": game[10] if is_player_1 else game[9], + "ruleset": resolve_ruleset(game[11]), + "status": game[12], + "private": bool(game[13]), + } + # check if game_id exists in database def valid_game_id(game_id): - query = cursor.execute("select game_id from games where game_id = ?", [game_id]).fetchone() - return bool(query) + query = cursor.execute( + "select game_id from games where game_id = ?", [game_id] + ).fetchone() + return bool(query) + game_info = Blueprint('game_info', __name__) -@game_info.route('/info', methods = ['POST']) + +@game_info.route('/info', methods=['POST']) def index(): - data = request.get_json() - if not data: return "", 400 + data = request.get_json() + if not data: return "", 400 + + game_id = data.get("id") or "" + if not game_id: return "", 400 - game_id = data.get("id") or "" - if not game_id: return "", 400 + user_id = None + token = request.cookies.get("token") or "" + if token: user_id = token_login(token) - user_id = None - token = request.cookies.get("token") or "" - if token: user_id = token_login(token) + if not valid_game_id(game_id): return "", 403 - if not valid_game_id(game_id): return "", 403 + return format_game(game_id, user_id), 200 - return format_game(game_id, user_id), 200 dynamic_route = ["/game", game_info] diff --git a/api/game/new.py b/api/game/new.py index 9868099..7f0862b 100644 --- a/api/game/new.py +++ b/api/game/new.py @@ -6,39 +6,60 @@ from randid import new_uuid from game.socket import games, game from hierarchy import auth_required -def create_game(user_1_id, private = False, user_2_id = None): - timestamp = int( time.time() * 1000 ) - game_id = new_uuid("games") +def create_game(user_1_id, private=False, user_2_id=None): + timestamp = int(time.time() * 1000) - cursor.execute("insert into games values (?, NULL, \"\", ?, ?, NULL, ?, NULL, ?, NULL, NULL, NULL, \"wait_for_opponent\", \"default\", ?, FALSE) ", (game_id, user_1_id, user_2_id, timestamp, timestamp, private)) - connection.commit() + game_id = new_uuid("games") + + cursor.execute( + "insert into games values (?, NULL, \"\", ?, ?, NULL, ?, NULL, ?, NULL, NULL, NULL, \"wait_for_opponent\", \"default\", ?, FALSE) ", + (game_id, user_1_id, user_2_id, timestamp, timestamp, private) + ) + connection.commit() + + return game_id - return game_id def start_game(game_id, user_2_id): - timestamp = int( time.time() * 1000 ) + timestamp = int(time.time() * 1000) + + db_game = cursor.execute( + "select player_2_id, status, private from games where game_id = ?", + [game_id] + ).fetchone() + if db_game[1] != "wait_for_opponent": return False - db_game = cursor.execute("select player_2_id, status, private from games where game_id = ?", [game_id]).fetchone() - if db_game[1] != "wait_for_opponent": return False + if db_game[0] == None: + cursor.execute( + "update games set player_2_id = ? where game_id = ?", + (user_2_id, game_id) + ) + cursor.execute( + "update games set status = \"in_progress\", started = ?, last_activity = ? where game_id = ?", + (timestamp, timestamp, game_id) + ) + connection.commit() - if db_game[0] == None: cursor.execute("update games set player_2_id = ? where game_id = ?", (user_2_id, game_id)) - cursor.execute("update games set status = \"in_progress\", started = ?, last_activity = ? where game_id = ?", (timestamp, timestamp, game_id)) - connection.commit() + players = cursor.execute( + "select player_1_id, player_2_id from games where game_id = ?", + [game_id] + ).fetchone() + games[game_id] = game(game_id, io, players[0], players[1]) - players = cursor.execute("select player_1_id, player_2_id from games where game_id = ?", [game_id]).fetchone() - games[game_id] = game(game_id, io, players[0], players[1]) + io.emit("gameStart", room=games[game_id].room) - io.emit("gameStart", room=games[game_id].room) new_game = Blueprint('new_game', __name__) -@new_game.route('/new', methods = ["GET", "POST"]) + +@new_game.route('/new', methods=["GET", "POST"]) @auth_required("user") def index(user_id): - # create a new private game (join by link) - #TODO: friend invites + notifications - game_id = create_game(user_id, True) - return { "id": game_id }, 200 + # create a new private game (join by link) + #TODO: friend invites + notifications + game_id = create_game(user_id, True) + return {"id": game_id}, 200 + dynamic_route = ["/game", new_game] diff --git a/api/game/random.py b/api/game/random.py index 4d70b56..2dfbe0b 100644 --- a/api/game/random.py +++ b/api/game/random.py @@ -11,27 +11,35 @@ from socket_io import io random_game = Blueprint('random', __name__) + @random_game.route('/random') @auth_required("user") def index(user_id): - # get public_games (random opponent queue) - public_games = cursor.execute("select game_id from games where private = FALSE and status = \"wait_for_opponent\"").fetchall() + # get public_games (random opponent queue) + public_games = cursor.execute( + "select game_id from games where private = FALSE and status = \"wait_for_opponent\"" + ).fetchall() + + game_started = False - game_started = False + # create a new public game if the queue is empty + if len(public_games) == 0: + game_id = create_game(user_id) + player_1 = True + # otherwise join a random public game + else: + game_id = random.choice(public_games)[0] - # create a new public game if the queue is empty - if len(public_games) == 0: - game_id = create_game(user_id) - player_1 = True - # otherwise join a random public game - else: - game_id = random.choice(public_games)[0] + start_game(game_id, user_id) - start_game(game_id, user_id) + player_1 = False + game_started = True - player_1 = False - game_started = True + return { + "id": game_id, + "player_1": player_1, + "game_started": game_started + }, 200 - return { "id": game_id, "player_1": player_1, "game_started": game_started }, 200 dynamic_route = ["/game", random_game] diff --git a/api/game/socket.py b/api/game/socket.py index cf45eb6..63f680c 100644 --- a/api/game/socket.py +++ b/api/game/socket.py @@ -9,97 +9,109 @@ from socket_io import io games = {} + class game: - def __init__(self, game_id, io, player_1_id, player_2_id): - self.game_id = game_id - self.room = "game-" + game_id - self.board = bord(7, 6) - self.io = io - self.player_1_id = player_1_id - self.player_2_id = player_2_id - - # drop a disc in `column` - def move(self, user_id, column): - if user_id != self.player_1_id and user_id != self.player_2_id: return - move = self.player_1_id if self.board.player_1 else self.player_2_id - if user_id != move: return - - self.board.drop_fisje(column) - - io.emit("fieldUpdate", { "field": self.board.board }, room=self.room) - - now = int( time.time() * 1000 ) - cursor.execute("update games set last_activity = ?, moves = moves || ? || ',' where game_id = ?", [now, column, self.game_id]) - connection.commit() - - if len(self.board.win_positions) > 0 or self.board.board_full: - outcome = "d" - if not self.board.board_full: - winner = self.board.board[int(self.board.win_positions[0][0])] - outcome = "w" if winner == "2" else "l" - io.emit("finish", { - "winPositions": self.board.win_positions, - "boardFull": self.board.board_full - }, room=self.room) - self.close("finished", outcome) - return - - io.emit("turnUpdate", { "player1": self.board.player_1 }, room=self.room) - - def resign(self): - self.board.kill_voerbak() - io.emit("resign", room=self.room) - self.close("resign", "d") - - def close(self, new_status, outcome): - cursor.execute(" ".join([ - "update games set", - "moves = moves || '0',", - "duration = ?,", - "status = ?,", - "outcome = ?", - "where game_id = ?" - ]), [ - int( time.time() * 1000 ) - cursor.execute("select started from games where game_id = ?", [self.game_id]).fetchone()[0], - new_status, - outcome, - self.game_id - ]) - connection.commit() - - games.pop(self.game_id) + def __init__(self, game_id, io, player_1_id, player_2_id): + self.game_id = game_id + self.room = "game-" + game_id + self.board = bord(7, 6) + self.io = io + self.player_1_id = player_1_id + self.player_2_id = player_2_id + + # drop a disc in `column` + def move(self, user_id, column): + if user_id != self.player_1_id and user_id != self.player_2_id: return + move = self.player_1_id if self.board.player_1 else self.player_2_id + if user_id != move: return + + self.board.drop_fisje(column) + + io.emit("fieldUpdate", {"field": self.board.board}, room=self.room) + + now = int(time.time() * 1000) + cursor.execute( + "update games set last_activity = ?, moves = moves || ? || ',' where game_id = ?", + [now, column, self.game_id] + ) + connection.commit() + + if len(self.board.win_positions) > 0 or self.board.board_full: + outcome = "d" + if not self.board.board_full: + winner = self.board.board[int(self.board.win_positions[0][0])] + outcome = "w" if winner == "2" else "l" + io.emit( + "finish", { + "winPositions": self.board.win_positions, + "boardFull": self.board.board_full + }, + room=self.room + ) + self.close("finished", outcome) + return + + io.emit("turnUpdate", {"player1": self.board.player_1}, room=self.room) + + def resign(self): + self.board.kill_voerbak() + io.emit("resign", room=self.room) + self.close("resign", "d") + + def close(self, new_status, outcome): + cursor.execute( + " ".join( + [ + "update games set", "moves = moves || '0',", + "duration = ?,", "status = ?,", "outcome = ?", + "where game_id = ?" + ] + ), [ + int(time.time() * 1000) - cursor.execute( + "select started from games where game_id = ?", + [self.game_id] + ).fetchone()[0], new_status, outcome, self.game_id + ] + ) + connection.commit() + + games.pop(self.game_id) + @io.on("newMove") def new_move(data): - if not data["game_id"] or \ - not data["move"] or \ - not data["token"]: return - if not data["game_id"] in games: return + if not data["game_id"] or \ + not data["move"] or \ + not data["token"]: + return + if not data["game_id"] in games: return + + game = games[data["game_id"]] + if (len(game.board.win_positions) > 0 or game.board.board_full): return + user_id = token_login(data["token"]) + game.move(user_id, data["move"]) - game = games[data["game_id"]] - if(len(game.board.win_positions) > 0 or game.board.board_full): return - user_id = token_login(data["token"]) - game.move(user_id, data["move"]) @io.on("resign") def resign(data): - if not data["game_id"] or \ - not request.cookies.get("token"): return - if not data["game_id"] in games: return + if not data["game_id"] or \ + not request.cookies.get("token"): + return + if not data["game_id"] in games: return - user_id = token_login(request.cookies.get("token")) - if not user_id: return + user_id = token_login(request.cookies.get("token")) + if not user_id: return - if games[data["game_id"]].player_1_id != user_id and \ - games[data["game_id"]].player_2_id != user_id: - return + if games[data["game_id"]].player_1_id != user_id and \ + games[data["game_id"]].player_2_id != user_id: + return + + games[data["game_id"]].resign() - games[data["game_id"]].resign() @io.on("registerGameListener") def register_game_listener(data): - game_id = data.get("game_id") - if not game_id: return - - join_room("game-" + game_id) + game_id = data.get("game_id") + if not game_id: return + join_room("game-" + game_id) diff --git a/api/game/voerbak_connector.py b/api/game/voerbak_connector.py index 048a5d1..412a512 100644 --- a/api/game/voerbak_connector.py +++ b/api/game/voerbak_connector.py @@ -11,82 +11,83 @@ EMPTY = Fore.LIGHTBLACK_EX + "_" + Fore.RESET VOERBAK_LOCATION = os.path.join(os.getcwd(), "voerbak/", "voerbak") if os.name == "nt": VOERBAK_LOCATION += ".exe" + class bord: - def __init__(self, w, h): - self.width = w - self.height = h - self.player_1 = True - self.board = "0" * (w * h) - self.board_full = False - self.win_positions = [] - self.process = subprocess.Popen([VOERBAK_LOCATION, f"-w {w}", f"-h {h}"], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=None) - self.process.stdin.flush() + def __init__(self, w, h): + self.width = w + self.height = h + self.player_1 = True + self.board = "0" * (w * h) + self.board_full = False + self.win_positions = [] + self.process = subprocess.Popen( + [VOERBAK_LOCATION, f"-w {w}", f"-h {h}"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=None + ) + self.process.stdin.flush() + + # get output from voerbak without trailing newline character (this might break on windows because crlf) + def get_output(self): + return self.process.stdout.readline().decode()[:-1] - # get output from voerbak without trailing newline character (this might break on windows because crlf) - def get_output(self): - return self.process.stdout.readline().decode()[:-1] + def kill_voerbak(self): + self.process.stdin.write(bytearray("0", "utf-8")) + self.process.stdin.flush() - def kill_voerbak(self): - self.process.stdin.write(bytearray("0", "utf-8")) - self.process.stdin.flush() + # read messages from voerbak + def update_board(self): + buffer = self.get_output() + while not buffer.isdigit(): + # win message + if buffer.startswith("w:"): + self.win_positions.append(buffer[2:].split("-")) + log.info(f"won: {buffer[2:].split('-')}") + self.kill_voerbak() + # error message + elif buffer.startswith("e:"): + log.warning(buffer[2:]) + # turn update message + elif buffer.startswith("m:"): + substr = buffer[2:] + self.player_1 = True if substr == "true" else False + # draw game message + elif buffer.startswith("d:"): + self.board_full = True + self.kill_voerbak() + buffer = self.get_output() + self.board = buffer - # read messages from voerbak - def update_board(self): - buffer = self.get_output() - while not buffer.isdigit(): - # win message - if buffer.startswith("w:"): - self.win_positions.append(buffer[2:].split("-")) - log.info(f"won: {buffer[2:].split('-')}") - self.kill_voerbak() - # error message - elif buffer.startswith("e:"): - log.warning(buffer[2:]) - # turn update message - elif buffer.startswith("m:"): - substr = buffer[2:] - self.player_1 = True if substr == "true" else False - # draw game message - elif buffer.startswith("d:"): - self.board_full = True - self.kill_voerbak() - buffer = self.get_output() - self.board = buffer + # debug board print function + def print(self): + for y in range(self.height - 1, -1, -1): + for x in range(self.width): + state = self.board[x + y * self.width] + char = [EMPTY, DISC_A, DISC_B] + print(char[int(state)], end=" ") + print("\n", end="") - # debug board print function - def print(self): - for y in range(self.height -1, -1, -1): - for x in range(self.width): - state = self.board[x + y * self.width] - char = [EMPTY, - DISC_A, - DISC_B - ] - print(char[int(state)], end=" ") - print("\n", end="") + def drop_fisje(self, column): + self.process.stdin.write(bytearray(f"{column}\n", "utf-8")) + self.process.stdin.flush() + self.update_board() - def drop_fisje(self, column): - self.process.stdin.write(bytearray(f"{column}\n", "utf-8")) - self.process.stdin.flush() - self.update_board() # debug game def main(): - gert = bord(7, 6) - while True: - print(gert.player_1) - if len(gert.win_positions) > 0: - print(f"won: {gert.win_positions}") - exit(0) - gert.print() - column = int(input("column?: ")) - 1 - if column not in range(gert.width): - continue - gert.drop_fisje(column + 1) + gert = bord(7, 6) + while True: + print(gert.player_1) + if len(gert.win_positions) > 0: + print(f"won: {gert.win_positions}") + exit(0) + gert.print() + column = int(input("column?: ")) - 1 + if column not in range(gert.width): + continue + gert.drop_fisje(column + 1) -if __name__ == "__main__": - main() +if __name__ == "__main__": + main() diff --git a/api/hierarchy.py b/api/hierarchy.py index 6844fe6..f080c45 100644 --- a/api/hierarchy.py +++ b/api/hierarchy.py @@ -4,24 +4,28 @@ from db import cursor ranks = ["none", "user", "moderator", "admin", "bot"] + # @auth_required function decorator (use after @flask.Blueprint.route() decorator) def auth_required(level): - def decorator(func): - def wrapper(): - token = request.cookies.get("token") or "" - if not token: return "", 403 + def decorator(func): + def wrapper(): + token = request.cookies.get("token") or "" + if not token: return "", 403 + + user_id = token_login(token) + if not user_id: return "", 403 - user_id = token_login(token) - if not user_id: return "", 403 + user_rank_text = cursor.execute( + "select type from users where user_id = ?", [user_id] + ).fetchone()[0] - user_rank_text = cursor.execute("select type from users where user_id = ?", [user_id]).fetchone()[0] + required_rank = ranks.index(level) + user_rank = ranks.index(user_rank_text) + if required_rank > user_rank: return "", 403 - required_rank = ranks.index(level) - user_rank = ranks.index(user_rank_text) - if required_rank > user_rank: return "", 403 + return func(user_id) - return func(user_id) - wrapper.__name__ = func.__name__ - return wrapper - return decorator + wrapper.__name__ = func.__name__ + return wrapper + return decorator @@ -8,4 +8,3 @@ logging.basicConfig(format="[ %(levelname)s ]: %(message)s", level=VERBOSE) error = logging.error info = logging.info warning = logging.warning - diff --git a/api/main.py b/api/main.py index a35c856..8f4e087 100644 --- a/api/main.py +++ b/api/main.py @@ -6,5 +6,4 @@ import game.socket # start the flask/socket.io server if __name__ == "__main__": - io.run(app, host="127.0.0.1", port=5000, debug=True) - + io.run(app, host="127.0.0.1", port=5000, debug=True) diff --git a/api/passwords.py b/api/passwords.py index 8fa15c3..9cd56da 100644 --- a/api/passwords.py +++ b/api/passwords.py @@ -1,14 +1,16 @@ import bcrypt + # encode string as utf-8 def enc(string): - return string.encode('utf-8') + return string.encode('utf-8') + # check if password matches against hash in database def check_password(password, password_hash): - return bcrypt.checkpw(enc(password), password_hash) + return bcrypt.checkpw(enc(password), password_hash) + # hash a password for storing in the database def password_hash(password): - return bcrypt.hashpw(enc(password), bcrypt.gensalt()) - + return bcrypt.hashpw(enc(password), bcrypt.gensalt()) diff --git a/api/randid.py b/api/randid.py index 0683c75..c837c2f 100644 --- a/api/randid.py +++ b/api/randid.py @@ -1,18 +1,18 @@ from db import cursor import uuid -tables = { - "users": "user_id", - "games": "game_id" - } +tables = {"users": "user_id", "games": "game_id"} + # generate a new uuid and check for collisions (unlikely but still) def new_uuid(table_name): - temp_uuid = str(uuid.uuid4()) - column_name = tables[table_name] - # check if id is already taken - if cursor.execute(f"select {column_name} from {table_name} where {column_name} = ?", [temp_uuid]).fetchone(): - return new_uuid(table_name) - else: - return temp_uuid - + temp_uuid = str(uuid.uuid4()) + column_name = tables[table_name] + # check if id is already taken + if cursor.execute( + f"select {column_name} from {table_name} where {column_name} = ?", + [temp_uuid] + ).fetchone(): + return new_uuid(table_name) + else: + return temp_uuid diff --git a/api/rating.py b/api/rating.py index 9cf11f5..d2beaa3 100644 --- a/api/rating.py +++ b/api/rating.py @@ -1,30 +1,41 @@ from db import cursor + def outcome(outcome_str, player_1): - outcome_int = { "w": 1, "l": -1, "d": 0 }[outcome_str] - if not player_1: outcome_int *= -1 - return { 1: "w", -1: "l", 0: "d" }[outcome_int] + outcome_int = {"w": 1, "l": -1, "d": 0}[outcome_str] + if not player_1: outcome_int *= -1 + return {1: "w", -1: "l", 0: "d"}[outcome_int] + + +def rating_v1(won_games): # python is a garbage language + return abs(won_games)**(1 / 2.2) * 23 * (1, -1)[won_games < 0] -def rating_v1(won_games): # python is a garbage language - return abs(won_games) ** (1 / 2.2) * 23 * (1, -1)[won_games < 0] def get_all_games(user_id): - return cursor.execute("select player_1_id, player_2_id, outcome " + \ - "from games " + \ - "where (player_1_id = ? or player_2_id = ?) " + \ - "and status = \"finished\" or status = \"resign\"", [user_id, user_id]).fetchall() + return cursor.execute("select player_1_id, player_2_id, outcome " + \ + "from games " + \ + "where (player_1_id = ? or player_2_id = ?) " + \ + "and status = \"finished\" or status = \"resign\"", [user_id, user_id]).fetchall() + # simple rating function that doesn't use game analysis def get_rating(user_id): - score = 400 - games = get_all_games(user_id) - # get all games for user_id and switch perspective in which user_id is player_2_id - mapped_games = [game if game[0] == user_id else (game[1], game[0], outcome(game[2], False)) for game in games] - counted_opponents = {} - for game in mapped_games: - # calculate sum score against user (+1 for win, -1 for lose, 0 for draw game) - counted_opponents[game[1]] = (counted_opponents.get(game[1]) or 0) + { "w": 1, "l": -1, "d": 0 }[game[2]] - for opponent in counted_opponents: - # apply the cool curve to the sum score and add to the base score of 400 - score += rating_v1(counted_opponents.get(opponent)) - return int(score) + score = 400 + games = get_all_games(user_id) + # get all games for user_id and switch perspective in which user_id is player_2_id + mapped_games = [ + game if game[0] == user_id else + (game[1], game[0], outcome(game[2], False)) for game in games + ] + counted_opponents = {} + for game in mapped_games: + # calculate sum score against user (+1 for win, -1 for lose, 0 for draw game) + counted_opponents[game[1]] = (counted_opponents.get(game[1]) or 0) + { + "w": 1, + "l": -1, + "d": 0 + }[game[2]] + for opponent in counted_opponents: + # apply the cool curve to the sum score and add to the base score of 400 + score += rating_v1(counted_opponents.get(opponent)) + return int(score) diff --git a/api/ruleset.py b/api/ruleset.py index b42b4e9..92c7077 100644 --- a/api/ruleset.py +++ b/api/ruleset.py @@ -15,23 +15,23 @@ rulesets = { } } + # resolve ruleset from ruleset name or dict def resolve_ruleset(ruleset): - # create return variable - export = {} - try: - # try to parse the ruleset as json - export = json.loads(ruleset) - merged = dict(rulesets["default"]) - - # fill missing keys in dict - merge(merged, export) - export = merged - except ValueError as e: - # if the ruleset is a name like 'default' or 'columns+2', read it from the predefined rulesets - if ruleset in rulesets: - export = rulesets[ruleset] - if not export: - export = rulesets["default"] - return export + # create return variable + export = {} + try: + # try to parse the ruleset as json + export = json.loads(ruleset) + merged = dict(rulesets["default"]) + # fill missing keys in dict + merge(merged, export) + export = merged + except ValueError as e: + # if the ruleset is a name like 'default' or 'columns+2', read it from the predefined rulesets + if ruleset in rulesets: + export = rulesets[ruleset] + if not export: + export = rulesets["default"] + return export diff --git a/api/social/create_relation.py b/api/social/create_relation.py index 7e7c466..f58e105 100644 --- a/api/social/create_relation.py +++ b/api/social/create_relation.py @@ -4,56 +4,66 @@ from hierarchy import auth_required from socket_io import io import time + # @two_person_endpoint decorator # defines (user_1_id, user_2_id) in endpoint handler function arguments def two_person_endpoint(func): - @auth_required("user") - def wrapper(user_1_id): - data = request.get_json() - user_2_id = data.get("id") or "" + @auth_required("user") + def wrapper(user_1_id): + data = request.get_json() + user_2_id = data.get("id") or "" + + if not user_1_id or \ + not user_2_id: + return "", 403 + + return func(user_1_id, user_2_id) - if not user_1_id or \ - not user_2_id: - return "", 403 + wrapper.__name__ = func.__name__ + return wrapper - return func(user_1_id, user_2_id) - wrapper.__name__ = func.__name__ - return wrapper def create_relation(user_1_id, user_2_id, relation_type): - remove_relation(user_1_id, user_2_id) - remove_relation(user_2_id, user_1_id) - timestamp = int( time.time() * 1000 ) - cursor.execute("insert into social values (?, ?, ?, ?)", - [user_1_id, user_2_id, relation_type, timestamp]) - connection.commit() + remove_relation(user_1_id, user_2_id) + remove_relation(user_2_id, user_1_id) + timestamp = int(time.time() * 1000) + cursor.execute( + "insert into social values (?, ?, ?, ?)", + [user_1_id, user_2_id, relation_type, timestamp] + ) + connection.commit() + # remove relation between user_1_id and user_2_id (one-way) def remove_relation(user_1_id, user_2_id): - cursor.execute("delete from social where user_1_id = ? and user_2_id = ?", - [user_1_id, user_2_id]) - connection.commit() + cursor.execute( + "delete from social where user_1_id = ? and user_2_id = ?", + [user_1_id, user_2_id] + ) + connection.commit() + def create_relation_route(relation_type): - @two_person_endpoint - def route(user_1_id, user_2_id): - create_relation(user_1_id, user_2_id, relation_type) + @two_person_endpoint + def route(user_1_id, user_2_id): + create_relation(user_1_id, user_2_id, relation_type) - if relation_type == "outgoing": - io.emit("incomingFriendRequest", room="user-"+user_2_id) + if relation_type == "outgoing": + io.emit("incomingFriendRequest", room="user-" + user_2_id) - return "", 200 - return route + return "", 200 -friend_request = Blueprint('friend_request', __name__) -friend_request.add_url_rule('/request', 'route', create_relation_route("outgoing"), methods = ["POST"]) + return route -block = Blueprint('block', __name__) -block.add_url_rule('/block', 'route', create_relation_route("block"), methods = ["POST"]) -dynamic_routes = [ - ["/social", friend_request], - ["/social", block] - ] +friend_request = Blueprint('friend_request', __name__) +friend_request.add_url_rule( + '/request', 'route', create_relation_route("outgoing"), methods=["POST"] +) +block = Blueprint('block', __name__) +block.add_url_rule( + '/block', 'route', create_relation_route("block"), methods=["POST"] +) +dynamic_routes = [["/social", friend_request], ["/social", block]] diff --git a/api/social/destroy_relation.py b/api/social/destroy_relation.py index 4b61ecd..ab72c48 100644 --- a/api/social/destroy_relation.py +++ b/api/social/destroy_relation.py @@ -7,32 +7,32 @@ import time remove = Blueprint('remove', __name__) -@remove.route('/remove', methods = ['POST']) + +@remove.route('/remove', methods=['POST']) @two_person_endpoint def index(user_1_id, user_2_id): - relation = get_relation_to(user_1_id, user_2_id) - if relation == "none": return "", 403 + relation = get_relation_to(user_1_id, user_2_id) + if relation == "none": return "", 403 + + remove_relation(user_1_id, user_2_id) + remove_relation(user_2_id, user_1_id) - remove_relation(user_1_id, user_2_id) - remove_relation(user_2_id, user_1_id) + io.emit("changedRelation", {"id": user_2_id}, room="user-" + user_1_id) + io.emit("changedRelation", {"id": user_1_id}, room="user-" + user_2_id) - io.emit("changedRelation", { "id": user_2_id }, room="user-"+user_1_id) - io.emit("changedRelation", { "id": user_1_id }, room="user-"+user_2_id) + return "", 200 - return "", 200 unblock = Blueprint('unblock', __name__) -@unblock.route('/unblock', methods = ['POST']) + +@unblock.route('/unblock', methods=['POST']) @two_person_endpoint def index(user_1_id, user_2_id): - if get_relation_to(user_1_id, user_2_id) != "blocked": return "", 403 + if get_relation_to(user_1_id, user_2_id) != "blocked": return "", 403 - remove_relation(user_1_id, user_2_id) - return "", 200 + remove_relation(user_1_id, user_2_id) + return "", 200 -dynamic_routes = [ - ["/social", remove], - ["/social", unblock] - ] +dynamic_routes = [["/social", remove], ["/social", unblock]] diff --git a/api/social/friend_accept.py b/api/social/friend_accept.py index 4f1f847..75dd3b9 100644 --- a/api/social/friend_accept.py +++ b/api/social/friend_accept.py @@ -6,17 +6,20 @@ import time accept = Blueprint('accept', __name__) -@accept.route("/accept", methods = ['POST']) + +@accept.route("/accept", methods=['POST']) @two_person_endpoint def route(user_1_id, user_2_id): - cursor.execute("update social set type = \"friendship\" where user_1_id = ? and user_2_id = ?", - [user_2_id, user_1_id]) - connection.commit() + cursor.execute( + "update social set type = \"friendship\" where user_1_id = ? and user_2_id = ?", + [user_2_id, user_1_id] + ) + connection.commit() - io.emit("changedRelation", { "id": user_2_id }, room="user-"+user_1_id) - io.emit("changedRelation", { "id": user_1_id }, room="user-"+user_2_id) + io.emit("changedRelation", {"id": user_2_id}, room="user-" + user_1_id) + io.emit("changedRelation", {"id": user_1_id}, room="user-" + user_2_id) - return "", 200 + return "", 200 -dynamic_route = ["/social", accept] +dynamic_route = ["/social", accept] diff --git a/api/social/request_list.py b/api/social/request_list.py index 624b3b4..8d1acd6 100644 --- a/api/social/request_list.py +++ b/api/social/request_list.py @@ -6,19 +6,22 @@ import time requests = Blueprint('requests', __name__) + @requests.route("/requests") @auth_required("user") def route(user_2_id): - # get a list of friend requests - request_list = cursor.execute("select user_1_id from social where user_2_id = ? and type = \"outgoing\"", - [user_2_id]).fetchall() + # get a list of friend requests + request_list = cursor.execute( + "select user_1_id from social where user_2_id = ? and type = \"outgoing\"", + [user_2_id] + ).fetchall() - # get user_id for each result to prevent repeat user/info requests - formatted_request_list = [] - for user_1_id in [q[0] for q in request_list]: - formatted_request_list.append(format_user(user_1_id)) + # get user_id for each result to prevent repeat user/info requests + formatted_request_list = [] + for user_1_id in [q[0] for q in request_list]: + formatted_request_list.append(format_user(user_1_id)) - return { "requests": formatted_request_list }, 200 + return {"requests": formatted_request_list}, 200 -dynamic_route = ["/social/list", requests] +dynamic_route = ["/social/list", requests] diff --git a/api/social/search.py b/api/social/search.py index 1159c67..a22ea73 100644 --- a/api/social/search.py +++ b/api/social/search.py @@ -5,25 +5,29 @@ import json search = Blueprint('search', __name__) -@search.route('/search', methods = ['POST']) + +@search.route('/search', methods=['POST']) def index(): - data_string = request.data or "{}" - data = json.loads(data_string) - query = data.get("query") or "" - if not query: return "", 400 - if len(query) < 3: return "", 403 + data_string = request.data or "{}" + data = json.loads(data_string) + query = data.get("query") or "" + if not query: return "", 400 + if len(query) < 3: return "", 403 - # use levenshtein with max distance 3 to search for users - #TODO: use mysql and sort by best match - results = cursor.execute("select user_id from users where levenshtein(lower(username), lower(?), 3)", [query]).fetchmany(20); + # use levenshtein with max distance 3 to search for users + #TODO: use mysql and sort by best match + results = cursor.execute( + "select user_id from users where levenshtein(lower(username), lower(?), 3)", + [query] + ).fetchmany(20) - formatted = { "results": [] } + formatted = {"results": []} - # get user_id for each result to prevent repeat user/info requests - for user in results: - formatted["results"].append(format_user(user[0])) + # get user_id for each result to prevent repeat user/info requests + for user in results: + formatted["results"].append(format_user(user[0])) - return formatted, 200 + return formatted, 200 -dynamic_route = ["/social", search] +dynamic_route = ["/social", search] diff --git a/api/socket_io.py b/api/socket_io.py index de3dd9e..dbe17ab 100644 --- a/api/socket_io.py +++ b/api/socket_io.py @@ -3,4 +3,3 @@ from app import app # socket.io wrapper to avoid circular imports (same as db.py) io = SocketIO(app, cors_allowed_origins="*") - diff --git a/api/status.py b/api/status.py index 19a01a1..45410a4 100644 --- a/api/status.py +++ b/api/status.py @@ -3,12 +3,18 @@ from db import cursor status = Blueprint('server_status', __name__) + @status.route('/status') def index(): - return { - # "users": int, - "games": len(cursor.execute ("select game_id from games where status = \"in_progress\"").fetchall()) - } + return { + # "users": int, + "games": + len( + cursor.execute( + "select game_id from games where status = \"in_progress\"" + ).fetchall() + ) + } -dynamic_route = ["/", status] +dynamic_route = ["/", status] diff --git a/api/user/avatar.py b/api/user/avatar.py index b4edeed..1b5500a 100644 --- a/api/user/avatar.py +++ b/api/user/avatar.py @@ -10,29 +10,34 @@ default_avatar = open("database/avatars/default.png", "rb").read() avatar = Blueprint('avatar', __name__) -@avatar.route('/avatar', methods = ["GET"]) + +@avatar.route('/avatar', methods=["GET"]) def get_avatar(): - token = request.cookies.get("token") or "" - login = token_login(token) or "" + token = request.cookies.get("token") or "" + login = token_login(token) or "" + + user_id = request.args.get("id") or login + if not user_id: return "", 400 + if not valid_user_id(user_id): return "", 403 - user_id = request.args.get("id") or login - if not user_id: return "", 400 - if not valid_user_id(user_id): return "", 403 + avatar_path = f"database/avatars/{user_id}.png" + avatar = "" + if exists(avatar_path): + avatar = open(avatar_path, "rb").read() + return Response(avatar or default_avatar, 200, mimetype="image/png") - avatar_path = f"database/avatars/{user_id}.png" - avatar = "" - if exists(avatar_path): - avatar = open(avatar_path, "rb").read() - return Response(avatar or default_avatar, 200, mimetype="image/png") -@avatar.route('/avatar', methods = ["POST"]) #TODO: pillow image size validation (client side resize) +@avatar.route( + '/avatar', methods=["POST"] +) #TODO: pillow image size validation (client side resize) @auth_required("user") def update_avatar(login): - if not request.data: return "", 400 + if not request.data: return "", 400 - open(f"database/avatars/{login}.png", "wb").write(decode(request.data, "base64")) + open(f"database/avatars/{login}.png", + "wb").write(decode(request.data, "base64")) - return "", 200 + return "", 200 -dynamic_route = ["/user", avatar] +dynamic_route = ["/user", avatar] diff --git a/api/user/games.py b/api/user/games.py index 3936566..83d721a 100644 --- a/api/user/games.py +++ b/api/user/games.py @@ -8,69 +8,89 @@ from ruleset import resolve_ruleset from game.info import format_game import json + # get total game outcome amount for user -def sum_games(user_id): #! SANITIZE USER_ID FIRST - wld_querys = [' '.join([ - "select count(game_id)", - "from games", - "where", - f"player_{x[0]}_id = \"{user_id}\" and", - f"outcome = \"{x[1]}\"", - ]) for x in [(1, "w"), (1, "l"), (2, "w"), (2, "l")]] - wld_querys.insert(0, ' '.join([ - "select count(game_id)", - "from games", - "where", - f"(player_1_id = \"{user_id}\" or player_2_id = \"{user_id}\") and", - "outcome = \"d\"", - ])) - - big_query = "select " + ", ".join([f"({query})" for query in wld_querys]) - - results = cursor.execute(big_query).fetchone() - - # win and lose are calculated from user_id's perspective (player_1_id, player_2_id in db) - return { - "draw": results[0], - "win": results[1] + results[4], - "lose": results[2] + results[3], - "games": reduce(lambda a, b: a + b, results) - } +def sum_games(user_id): #! SANITIZE USER_ID FIRST + wld_querys = [ + ' '.join( + [ + "select count(game_id)", + "from games", + "where", + f"player_{x[0]}_id = \"{user_id}\" and", + f"outcome = \"{x[1]}\"", + ] + ) for x in [(1, "w"), (1, "l"), (2, "w"), (2, "l")] + ] + wld_querys.insert( + 0, ' '.join( + [ + "select count(game_id)", + "from games", + "where", + f"(player_1_id = \"{user_id}\" or player_2_id = \"{user_id}\") and", + "outcome = \"d\"", + ] + ) + ) + + big_query = "select " + ", ".join([f"({query})" for query in wld_querys]) + + results = cursor.execute(big_query).fetchone() + + # win and lose are calculated from user_id's perspective (player_1_id, player_2_id in db) + return { + "draw": results[0], + "win": results[1] + results[4], + "lose": results[2] + results[3], + "games": reduce(lambda a, b: a + b, results) + } + # get `count` games that `user_id` participated in, sorted by newest game def fetch_games(user_id, count): - game_ids = cursor.execute("select game_id from games where player_1_id = ? or player_2_id = ? order by created desc", [user_id, user_id]).fetchmany(count) - export = [] + game_ids = cursor.execute( + "select game_id from games where player_1_id = ? or player_2_id = ? order by created desc", + [user_id, user_id] + ).fetchmany(count) + export = [] + + for game_id in game_ids: + export.append(format_game(game_id[0], user_id)) - for game_id in game_ids: - export.append(format_game(game_id[0], user_id)) + return export - return export games = Blueprint('games', __name__) -@games.route('/games', methods = ['GET', 'POST']) + +@games.route('/games', methods=['GET', 'POST']) def index(): - data_string = request.data or "{}" - data = json.loads(data_string) + data_string = request.data or "{}" + data = json.loads(data_string) + + user_id = data.get("id") or "" + token = request.cookies.get("token") or "" - user_id = data.get("id") or "" - token = request.cookies.get("token") or "" + if not user_id and \ + not token: + return "", 400 - if not user_id and \ - not token: - return "", 400 + if token and not user_id: + user_id = token_login(token) - if token and not user_id: - user_id = token_login(token) + if not cursor.execute( + "select user_id from users where user_id = ?", [user_id] + ).fetchone(): + return "", 403 - if not cursor.execute("select user_id from users where user_id = ?", [user_id]).fetchone(): return "", 403 + export = {} + merge( + export, {"totals": sum_games(user_id)}, + {"games": fetch_games(user_id, 20)} + ) - export = {} - merge(export, - {"totals": sum_games(user_id)}, - {"games": fetch_games(user_id, 20)}) + return export, 200 - return export, 200 dynamic_route = ["/user", games] diff --git a/api/user/info.py b/api/user/info.py index be48ef1..de0d2e8 100644 --- a/api/user/info.py +++ b/api/user/info.py @@ -4,84 +4,103 @@ from auth.login_token import token_login from rating import get_rating import json + # check if user_id exists in database def valid_user_id(user_id): - query = cursor.execute("select user_id from users where user_id = ?", [user_id]).fetchone() - return bool(query) + query = cursor.execute( + "select user_id from users where user_id = ?", [user_id] + ).fetchone() + return bool(query) + # get relation to user_2_id from user_1_id's perspective def get_relation_to(user_1_id, user_2_id): - relation = cursor.execute("select * from social where " + \ - "(user_1_id = ? and user_2_id = ?) or " + \ - "(user_1_id = ? and user_2_id = ?)", [user_1_id, user_2_id, user_2_id, user_1_id]).fetchone() - if not relation: return "none" - if relation[2] == "friendship": return "friends" - if relation[2] == "outgoing" and relation[0] == user_1_id: return "outgoing" - if relation[2] == "outgoing" and relation[1] == user_1_id: return "incoming" - if relation[2] == "block" and relation[0] == user_1_id: return "blocked" - return "none" + relation = cursor.execute("select * from social where " + \ + "(user_1_id = ? and user_2_id = ?) or " + \ + "(user_1_id = ? and user_2_id = ?)", [user_1_id, user_2_id, user_2_id, user_1_id]).fetchone() + if not relation: return "none" + if relation[2] == "friendship": return "friends" + if relation[2] == "outgoing" and relation[0] == user_1_id: + return "outgoing" + if relation[2] == "outgoing" and relation[1] == user_1_id: + return "incoming" + if relation[2] == "block" and relation[0] == user_1_id: return "blocked" + return "none" + # get users friend count def count_friends(user_id): - query = cursor.execute("select type from social where (user_1_id = ? or user_2_id = ?) and type = \"friendship\"", [user_id, user_id]).fetchall() - return len(query) #FIXME: use SQL count() instead of python's len() + query = cursor.execute( + "select type from social where (user_1_id = ? or user_2_id = ?) and type = \"friendship\"", + [user_id, user_id] + ).fetchall() + return len(query) #FIXME: use SQL count() instead of python's len() + # get user/info of `user_id` as `viewer` (id) -def format_user(user_id, viewer = ''): - user = cursor.execute("select " + ", ".join([ - "username", - "user_id", - "country", - "registered", - "status", - ]) + " from users where user_id = ?", [user_id]).fetchone() - formatted_user = { - "username": user[0], - "id": user[1], - "country": user[2], - "registered": user[3], - "status": user[4], - "friends": count_friends(user_id), - "rating": get_rating(user_id), #TODO: calculate rating based on game analysis - } - if viewer: - #FIXME: validate viewer id? - formatted_user["relation"] = get_relation_to(viewer, user_id) - return formatted_user +def format_user(user_id, viewer=''): + user = cursor.execute( + "select " + ", ".join( + [ + "username", + "user_id", + "country", + "registered", + "status", + ] + ) + " from users where user_id = ?", [user_id] + ).fetchone() + formatted_user = { + "username": user[0], + "id": user[1], + "country": user[2], + "registered": user[3], + "status": user[4], + "friends": count_friends(user_id), + "rating": + get_rating(user_id), #TODO: calculate rating based on game analysis + } + if viewer: + #FIXME: validate viewer id? + formatted_user["relation"] = get_relation_to(viewer, user_id) + return formatted_user + info = Blueprint('info', __name__) + # view own user/info if no user_id or username is provided and is logged in, # else view user/info of user with user_id = `user_id` or username = `username` -@info.route('/info', methods = ['GET', 'POST']) +@info.route('/info', methods=['GET', 'POST']) def index(): - data_string = request.data or "{}" - data = json.loads(data_string) - - username = data.get("username") or "" - user_id = data.get("id") or "" - token = request.cookies.get("token") or "" - viewer = "" - - if not username and \ - not user_id and \ - not token: - return "", 400 - - if username: - temp_user_id = cursor.execute("select user_id from users where username = ?", [username]).fetchone() - if len(temp_user_id) > 0: user_id = temp_user_id - - if token: - self_id = token_login(token) - if not (username or user_id): - user_id = self_id - if user_id: - viewer = self_id - - if user_id and not valid_user_id(user_id): return "", 403 - user = format_user(user_id, viewer) - - return user, 200 + data_string = request.data or "{}" + data = json.loads(data_string) + + username = data.get("username") or "" + user_id = data.get("id") or "" + token = request.cookies.get("token") or "" + viewer = "" + + if all(not v for v in [username, user_id, token]): + return "", 400 + + if username: + temp_user_id = cursor.execute( + "select user_id from users where username = ?", [username] + ).fetchone() + if len(temp_user_id) > 0: user_id = temp_user_id + + if token: + self_id = token_login(token) + if not (username or user_id): + user_id = self_id + if user_id: + viewer = self_id + + if user_id and not valid_user_id(user_id): return "", 403 + user = format_user(user_id, viewer) + + return user, 200 + dynamic_route = ["/user", info] diff --git a/api/user/password.py b/api/user/password.py index 0c1cb70..ff52ba4 100644 --- a/api/user/password.py +++ b/api/user/password.py @@ -3,16 +3,17 @@ from db import cursor password = Blueprint('password', __name__) + # this endpoint is unfinished @password.route('/password') def index(): - data = request.get_json() + data = request.get_json() - if not data["password"] or \ - not data["newPassword"]: - return "", 400 + if not data["password"] or \ + not data["newPassword"]: + return "", 400 - return {}, 200 + return {}, 200 -dynamic_route = ["/user", password] +dynamic_route = ["/user", password] diff --git a/api/user/preferences.py b/api/user/preferences.py index d4e27c9..2feaade 100644 --- a/api/user/preferences.py +++ b/api/user/preferences.py @@ -4,38 +4,49 @@ from ruleset import resolve_ruleset from hierarchy import auth_required import json + # fill missing dict keys in preferences object def format_preferences(prefs): - return { - "darkMode": prefs.get("darkMode") or False, - "ruleset": resolve_ruleset(json.dumps(prefs.get("ruleset") or {}) or "default"), - "userColors": { - "diskA": prefs.get("userColors", {}).get("diskA") or "", - "diskB": prefs.get("userColors", {}).get("diskB") or "", - "background": prefs.get("userColors", {}).get("background") or "" - } - } + return { + "darkMode": + prefs.get("darkMode") or False, + "ruleset": + resolve_ruleset(json.dumps(prefs.get("ruleset") or {}) or "default"), + "userColors": { + "diskA": prefs.get("userColors", {}).get("diskA") or "", + "diskB": prefs.get("userColors", {}).get("diskB") or "", + "background": prefs.get("userColors", {}).get("background") or "" + } + } + preferences = Blueprint('preferences', __name__) -@preferences.route('/preferences', methods = ["GET"]) + +@preferences.route('/preferences', methods=["GET"]) @auth_required("user") def get_preferences(login): - user_prefs = cursor.execute("select preferences from users where user_id = ?", [login]).fetchone() - return { "preferences": format_preferences(json.loads(user_prefs[0])) }, 200 + user_prefs = cursor.execute( + "select preferences from users where user_id = ?", [login] + ).fetchone() + return {"preferences": format_preferences(json.loads(user_prefs[0]))}, 200 -@preferences.route('/preferences', methods = ["POST"]) + +@preferences.route('/preferences', methods=["POST"]) @auth_required("user") def index(login): - data = request.get_json() - new_preferences = data.get("newPreferences") or "" + data = request.get_json() + new_preferences = data.get("newPreferences") or "" - formatted_json = format_preferences(new_preferences) + formatted_json = format_preferences(new_preferences) - cursor.execute("update users set preferences = ? where user_id = ?", [json.dumps(formatted_json), login]) - connection.commit() + cursor.execute( + "update users set preferences = ? where user_id = ?", + [json.dumps(formatted_json), login] + ) + connection.commit() - return "", 200 + return "", 200 -dynamic_route = ["/user", preferences] +dynamic_route = ["/user", preferences] diff --git a/api/user/status.py b/api/user/status.py index 3adb013..e762d12 100644 --- a/api/user/status.py +++ b/api/user/status.py @@ -5,17 +5,21 @@ import json status = Blueprint('user_status', __name__) -@status.route('/status', methods = ['POST']) + +@status.route('/status', methods=['POST']) @auth_required("user") def index(user_id): - data = request.get_json() - status = data.get("status") or "" - if not status: return "", 400 + data = request.get_json() + status = data.get("status") or "" + if not status: return "", 400 - cursor.execute("update users set status = ? where user_id = ?", [status[0:200], user_id]) - connection.commit() + cursor.execute( + "update users set status = ? where user_id = ?", + [status[0:200], user_id] + ) + connection.commit() - return "", 200 + return "", 200 -dynamic_route = ["/user", status] +dynamic_route = ["/user", status] diff --git a/database/database.py b/database/database.py index 30d23bb..14f6cb0 100644 --- a/database/database.py +++ b/database/database.py @@ -8,203 +8,331 @@ from uuid import uuid4 connection = sqlite3.connect("./database.db") cursor = connection.cursor() + def random_string(): - return str(floor(random() * 1e16)) # 1e16 is wetenschappelijke notatie (1 * 10^16) + return str( + floor(random() * 1e16) + ) # 1e16 is wetenschappelijke notatie (1 * 10^16) + def repeat_action(action, name): - def return_function(): - count = int(input(f"Hoeveel {name} wil je aanmaken? ")) - for i in range(count): - action() - print(f"{i+1}/{count} {name} toegevoegd!") - print(f"{count} {name} toegevoegd!") - return return_function + def return_function(): + count = int(input(f"Hoeveel {name} wil je aanmaken? ")) + for i in range(count): + action() + print(f"{i+1}/{count} {name} toegevoegd!") + print(f"{count} {name} toegevoegd!") + + return return_function + # dit is om dit script te organiseren programma_acties = [] + + class programma_actie: - def __init__(self, name, execute): - self.name = name - self.exec = execute + def __init__(self, name, execute): + self.name = name + self.exec = execute + # initialiseer een nieuwe (lege) database def init_db(): - commands_text = open("./init_db.sql").read() - commands = commands_text.strip().split("\n"*2) # python sqlite3 kan maar 1 "statement" per cursor.execute doen, en de statements zijn gescheden door witregels in ./init_db.sql - for i, command in enumerate(commands): - cursor.execute(command) - print(f"SQL commando {i+1} van de {len(commands)} uitgevoerd...") - print("Klaar!") + commands_text = open("./init_db.sql").read() + commands = commands_text.strip().split( + "\n" * 2 + ) # python sqlite3 kan maar 1 "statement" per cursor.execute doen, en de statements zijn gescheden door witregels in ./init_db.sql + for i, command in enumerate(commands): + cursor.execute(command) + print(f"SQL commando {i+1} van de {len(commands)} uitgevoerd...") + print("Klaar!") + + programma_acties.append(programma_actie("Maak tabellen aan", init_db)) + def destroy_tables(): - for table_name in ["games", "users", "social"]: - # format strings + sql is gevaarlijk vanwege sql injectie, maar omdat de invoer hier niet door de gebruiker aangepast kan worden is het semi-ok - cursor.execute(f"drop table if exists {table_name}") - print(f"Tabel \"{table_name}\" verwijderd!") -programma_acties.append(programma_actie("Verwijder alle tabellen", destroy_tables)) + for table_name in ["games", "users", "social"]: + # format strings + sql is gevaarlijk vanwege sql injectie, maar omdat de invoer hier niet door de gebruiker aangepast kan worden is het semi-ok + cursor.execute(f"drop table if exists {table_name}") + print(f"Tabel \"{table_name}\" verwijderd!") + + +programma_acties.append( + programma_actie("Verwijder alle tabellen", destroy_tables) +) + def user_amount(): - return len(cursor.execute("select user_id from users").fetchall()) + return len(cursor.execute("select user_id from users").fetchall()) + # pak een willekeurige user_id uit de users tabel (gebruikt voor insert_random::games) def random_player(): - return choice(cursor.execute("select user_id from users").fetchall())[0] + return choice(cursor.execute("select user_id from users").fetchall())[0] + # class om alle functies die willekeurige data invoegen te organiseren (python heeft geen namespaces :( ) class insert_random: - # omdat veel kolommen in onze database gebruikersinteracties opslaan laat ik ze hier leeg omdat ik geen tijd hem om gebruikersgedrag te simuleren. - - def __init__(self): - pass + # omdat veel kolommen in onze database gebruikersinteracties opslaan laat ik ze hier leeg omdat ik geen tijd hem om gebruikersgedrag te simuleren. + + def __init__(self): + pass + + def users(self): + user_id = str(uuid4()) + username = random_string() + email = f"{username}@example.com" + password_hash = bcrypt.hashpw( + random_string().encode("utf-8"), bcrypt.gensalt() + ) + registered = int(time.time() * 1000) + + cursor.execute( + "insert into users values (?, ?, ?, NULL, ?, ?, \"[]\", FALSE, \"user\", \"{}\", NULL, \"online\") ", + (user_id, username, email, password_hash, registered) + ) + + def games(self): + if user_amount() < 2: + print("Er zijn niet genoeg gebruikers om spellen aan te maken!") + return + + game_id = str(uuid4()) + timestamp = int(time.time() * 1000) + + cursor.execute( + "insert into games values (?, NULL, NULL, ?, ?, NULL, ?, NULL, NULL, NULL, FALSE, \"in_progress\", \"default\") ", + (game_id, random_player(), random_player(), timestamp) + ) + + def friends(self): + users = list( + rij[0] + for rij in cursor.execute("select user_id from users").fetchall() + ) # cursor.execute.fetchall() stuurt rijen altijd terug als tuples ookal vraag je 1 kolom op, dus dit zet het om naar een list van strings + # maak een nieuwe rij aan voor elke gebruiker in tabel users als deze nog niet bestaat + for user in users: + if cursor.execute( + "select user_id from social where user_id = ?", [user] + ).fetchone(): + continue + cursor.execute( + "insert into social values (?, NULL, NULL, NULL)", [user] + ) + + +programma_acties.append( + programma_actie( + "Maak nepgebruikers aan", + repeat_action(insert_random().users, "nepgebruikers") + ) +) +programma_acties.append( + programma_actie( + "Maak voorbeeld spellen aan", + repeat_action(insert_random().games, "spellen") + ) +) +programma_acties.append( + programma_actie( + "Vul social tabel met gebruikers", + insert_random().friends + ) +) - def users(self): - user_id = str(uuid4()) - username = random_string() - email = f"{username}@example.com" - password_hash = bcrypt.hashpw(random_string().encode("utf-8"), bcrypt.gensalt()) - registered = int( time.time() * 1000 ) - cursor.execute("insert into users values (?, ?, ?, NULL, ?, ?, \"[]\", FALSE, \"user\", \"{}\", NULL, \"online\") ", - (user_id, username, email, password_hash, registered)) - - def games(self): - if user_amount() < 2: - print("Er zijn niet genoeg gebruikers om spellen aan te maken!") - return +# Verifiëer een user_id of verkrijg een user_id uit een username +def resolve(resolveable): + user_id = None + user_id = user_id or cursor.execute( + "select user_id from users where user_id = ?", [resolveable] + ).fetchone() + user_id = user_id or cursor.execute( + "select user_id from users where username = ?", [resolveable] + ).fetchone() + # stuur de eerste kolom van de rij terug omdat het weer een tuple is >:( + return user_id[0] if user_id else None - game_id = str(uuid4()) - timestamp = int( time.time() * 1000 ) - cursor.execute("insert into games values (?, NULL, NULL, ?, ?, NULL, ?, NULL, NULL, NULL, FALSE, \"in_progress\", \"default\") ", - (game_id, random_player(), random_player(), timestamp)) +def edit_username(): + resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") + user_id = resolve(resolveable) + if not user_id: + print( + "Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"" + ) + return + new_username = input("Voer een nieuwe gebruikersnaam in: ") + cursor.execute( + "update users set username = ? where user_id = ?", + [new_username, user_id] + ) + print("Nieuwe gebruikersnaam succesvol toegepast!") + + +programma_acties.append( + programma_actie( + "Pas de gebruikersnaam van een gebruiker aan", edit_username + ) +) - def friends(self): - users = list(rij[0] for rij in cursor.execute("select user_id from users").fetchall()) # cursor.execute.fetchall() stuurt rijen altijd terug als tuples ookal vraag je 1 kolom op, dus dit zet het om naar een list van strings - # maak een nieuwe rij aan voor elke gebruiker in tabel users als deze nog niet bestaat - for user in users: - if cursor.execute("select user_id from social where user_id = ?", [user]).fetchone(): continue - cursor.execute("insert into social values (?, NULL, NULL, NULL)", [user]) -programma_acties.append(programma_actie("Maak nepgebruikers aan", repeat_action(insert_random().users, "nepgebruikers"))) -programma_acties.append(programma_actie("Maak voorbeeld spellen aan", repeat_action(insert_random().games, "spellen"))) -programma_acties.append(programma_actie("Vul social tabel met gebruikers", insert_random().friends)) +def delete_user(): + resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") + user_id = resolve(resolveable) + if not user_id: + print( + "Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"" + ) + return + cursor.execute("delete from users where user_id = ?", [user_id]) + print("Gebruiker verwijderd!") -# Verifiëer een user_id of verkrijg een user_id uit een username -def resolve(resolveable): - user_id = None - user_id = user_id or cursor.execute("select user_id from users where user_id = ?", [resolveable]).fetchone() - user_id = user_id or cursor.execute("select user_id from users where username = ?", [resolveable]).fetchone() - # stuur de eerste kolom van de rij terug omdat het weer een tuple is >:( - return user_id[0] if user_id else None -def edit_username(): - resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") - user_id = resolve(resolveable) - if not user_id: - print("Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"") - return - new_username = input("Voer een nieuwe gebruikersnaam in: ") - cursor.execute("update users set username = ? where user_id = ?", [new_username, user_id]) - print("Nieuwe gebruikersnaam succesvol toegepast!") -programma_acties.append(programma_actie("Pas de gebruikersnaam van een gebruiker aan", edit_username)) +programma_acties.append( + programma_actie("Verwijder een gebruiker", delete_user) +) -def delete_user(): - resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") - user_id = resolve(resolveable) - if not user_id: - print("Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"") - return - cursor.execute("delete from users where user_id = ?", [user_id]) - print("Gebruiker verwijderd!") -programma_acties.append(programma_actie("Verwijder een gebruiker", delete_user)) def mark_game_done(): - game_id = input("Voer een game-id in om als klaar te markeren: ") - if not cursor.execute("select game_id from games where game_id = ?", [game_id]).fetchone(): - print("Kon geen spel vinden met die game_id!") - return - - outcome = "w" - starttime = cursor.execute("select timestamp from games where game_id = ?", [game_id]).fetchone()[0] - duration = int( time.time() * 1000 ) - starttime - cursor.execute("update games set outcome = ?, duration = ?, status = \"finished\" where game_id = ?", [outcome, duration, game_id]) - print("Spel gemarkeerd als afgelopen!") -programma_acties.append(programma_actie("Spel markeren als afgelopen", mark_game_done)) + game_id = input("Voer een game-id in om als klaar te markeren: ") + if not cursor.execute( + "select game_id from games where game_id = ?", [game_id] + ).fetchone(): + print("Kon geen spel vinden met die game_id!") + return + + outcome = "w" + starttime = cursor.execute( + "select timestamp from games where game_id = ?", [game_id] + ).fetchone()[0] + duration = int(time.time() * 1000) - starttime + cursor.execute( + "update games set outcome = ?, duration = ?, status = \"finished\" where game_id = ?", + [outcome, duration, game_id] + ) + print("Spel gemarkeerd als afgelopen!") + + +programma_acties.append( + programma_actie("Spel markeren als afgelopen", mark_game_done) +) + def delete_game(): - game_id = input("Voer een game-id in om te verwijderen: ") - if not cursor.execute("select game_id from games where game_id = ?", [game_id]).fetchone(): - print("Kon geen spel vinden met die game_id!") - return + game_id = input("Voer een game-id in om te verwijderen: ") + if not cursor.execute( + "select game_id from games where game_id = ?", [game_id] + ).fetchone(): + print("Kon geen spel vinden met die game_id!") + return + + cursor.execute("delete from games where game_id = ?", [game_id]) + print("Spel verwijderd!") + - cursor.execute("delete from games where game_id = ?", [game_id]) - print("Spel verwijderd!") programma_acties.append(programma_actie("Spel verwijderen", delete_game)) + # maakt een vriendschap tussen 2 willekeurige user_id's def make_friendship(): - # Dit is misschien de lelijkste functie die ik ooit heb geschreven, wees gewaarschuwd - user_1 = random_player() - user_2 = random_player() - while user_1 == user_2: # garandeer dat user_1 != user_2 - user_2 = random_player() - - # [v for v in ? if v] zorgt er voor dat lege list elementen worden verwijderd - # .fetchone()[0] or "" is omdat deze kolommen ook NULL kunnen zijn, maar ik wil altijd een string - # .split(",") is omdat het user_id's zijn die gescheden zijn door komma's - user_1_friend_list = [v for v in (cursor.execute("select friends from social where user_id = ?", [user_1]).fetchone()[0] or "").split(",") if v] - user_2_friend_list = [v for v in (cursor.execute("select friends from social where user_id = ?", [user_2]).fetchone()[0] or "").split(",") if v] - - user_1_friend_list.append(user_2) - user_2_friend_list.append(user_1) + # Dit is misschien de lelijkste functie die ik ooit heb geschreven, wees gewaarschuwd + user_1 = random_player() + user_2 = random_player() + while user_1 == user_2: # garandeer dat user_1 != user_2 + user_2 = random_player() + + # [v for v in ? if v] zorgt er voor dat lege list elementen worden verwijderd + # .fetchone()[0] or "" is omdat deze kolommen ook NULL kunnen zijn, maar ik wil altijd een string + # .split(",") is omdat het user_id's zijn die gescheden zijn door komma's + user_1_friend_list = [ + v for v in ( + cursor.execute( + "select friends from social where user_id = ?", [user_1] + ).fetchone()[0] or "" + ).split(",") if v + ] + user_2_friend_list = [ + v for v in ( + cursor.execute( + "select friends from social where user_id = ?", [user_2] + ).fetchone()[0] or "" + ).split(",") if v + ] + + user_1_friend_list.append(user_2) + user_2_friend_list.append(user_1) + + cursor.execute( + "update social set friends = ? where user_id = ?", + [",".join(user_1_friend_list), user_1] + ) + cursor.execute( + "update social set friends = ? where user_id = ?", + [",".join(user_2_friend_list), user_2] + ) + + print(f"Nieuwe vriendschap gemaakt tussen {user_1} en {user_2}") - cursor.execute("update social set friends = ? where user_id = ?", [",".join(user_1_friend_list), user_1]) - cursor.execute("update social set friends = ? where user_id = ?", [",".join(user_2_friend_list), user_2]) - print(f"Nieuwe vriendschap gemaakt tussen {user_1} en {user_2}") programma_acties.append(programma_actie("Vriendschap maken", make_friendship)) + def reset_friendlist(): - resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") - user_id = resolve(resolveable) - if not user_id: - print("Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"") - return - cursor.execute("delete from social where user_id = ?", [user_id]) - cursor.execute("insert into social values (?, NULL, NULL, NULL)", [user_id]) - print(f"Vriendenlijst van {user_id} geleegd!") -programma_acties.append(programma_actie("Vriendenlijst legen", reset_friendlist)) + resolveable = input("Voer een gebruikers-id of gebruikersnaam in: ") + user_id = resolve(resolveable) + if not user_id: + print( + "Er is geen gebruiker met gebruikers-id of gebruikersnaam \"{resolveable}\"" + ) + return + cursor.execute("delete from social where user_id = ?", [user_id]) + cursor.execute( + "insert into social values (?, NULL, NULL, NULL)", [user_id] + ) + print(f"Vriendenlijst van {user_id} geleegd!") + + +programma_acties.append( + programma_actie("Vriendenlijst legen", reset_friendlist) +) for table in ["users", "games", "social"]: - programma_acties.append(programma_actie( - f"Tabel '{table}' weergeven", - lambda: print(cursor.execute(f"select * from {table}").fetchall()) - )) - -programma_acties.append(programma_actie("Sla alle wijzigingen op", connection.commit)) + programma_acties.append( + programma_actie( + f"Tabel '{table}' weergeven", + lambda: print(cursor.execute(f"select * from {table}").fetchall()) + ) + ) + +programma_acties.append( + programma_actie("Sla alle wijzigingen op", connection.commit) +) programma_acties.append(programma_actie("Stop", exit)) + def main(): - while True: - # print witregels - print("\n"*2, end="") + while True: + # print witregels + print("\n" * 2, end="") + + # print alle acties in programma_acties + for i, actie in enumerate(programma_acties): + print(f"#{i}: {actie.name}") - # print alle acties in programma_acties - for i, actie in enumerate(programma_acties): - print(f"#{i}: {actie.name}") + # vraag om keuze + keuze = int(input("Geef je keuze: ")) + print("\n", end="") - # vraag om keuze - keuze = int(input("Geef je keuze: ")) - print("\n", end="") + # kijk of keuze bestaat + if keuze > len(programma_acties) - 1: + print("Die actie bestaat niet!") + continue # ga door naar de volgende iteratie van de while loop (voer keuze niet uit) - # kijk of keuze bestaat - if keuze > len(programma_acties) - 1: - print("Die actie bestaat niet!") - continue # ga door naar de volgende iteratie van de while loop (voer keuze niet uit) + # voer keuze uit + programma_acties[keuze].exec() - # voer keuze uit - programma_acties[keuze].exec() if __name__ == "__main__": - main() + main() diff --git a/dprint.json b/dprint.json index bd359a8..1eeb09e 100644 --- a/dprint.json +++ b/dprint.json @@ -17,7 +17,8 @@ "yapf": { "based_on_style": "pep8", "dedent_closing_brackets": true, - "use_tabs": true + "use_tabs": true, + "indent_width": 4 }, "includes": [ "**/*.{ts,tsx}", diff --git a/voerbak/main.py b/voerbak/main.py index 433f130..d802603 100644 --- a/voerbak/main.py +++ b/voerbak/main.py @@ -6,85 +6,92 @@ DISC_A = Fore.RED + DISC_SHAPE + Fore.RESET DISC_B = Fore.BLUE + DISC_SHAPE + Fore.RESET EMPTY = Fore.LIGHTBLACK_EX + "_" + Fore.RESET + class bord: - def __init__(self, w, h): - self.width = w - self.height = h - self.board = [[EMPTY for x in range(self.width)] for u in range(self.height)] + def __init__(self, w, h): + self.width = w + self.height = h + self.board = [ + [EMPTY for x in range(self.width)] for u in range(self.height) + ] + + def print(self): + print( + "\n".join( + [ + " ".join(self.board[y]) + for y in range(len(self.board) - 1, -1, -1) + ] + ) + ) - def print(self): - print("\n".join([" ".join(self.board[y]) for y in range(len(self.board) -1, -1, -1)])) + def outside_board(self, coords): + return coords[0] < 0 or \ + coords[1] < 0 or \ + coords[0] > self.height - 1 or \ + coords[1] > self.width - 1 - def outside_board(self, coords): - return coords[0] < 0 or \ - coords[1] < 0 or \ - coords[0] > self.height - 1 or \ - coords[1] > self.width - 1 + def recursive_solve(self, coords, check_for, direction, current_length): + new_position = (coords[0] + direction[0], coords[1] + direction[1]) + if self.outside_board(new_position) or self.board[new_position[0]][ + new_position[1]] != check_for: + return current_length + else: + return self.recursive_solve( + new_position, check_for, direction, current_length + 1 + ) - def recursive_solve(self, coords, check_for, direction, current_length): - new_position = ( - coords[0] + direction[0], - coords[1] + direction[1] - ) - if self.outside_board(new_position) or self.board[new_position[0]][new_position[1]] != check_for: - return current_length - else: - return self.recursive_solve(new_position, check_for, direction, current_length + 1) + def check_win(self, coords): + directions = [ + (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1), (0, -1), + (1, -1) + ] + values = list() + for direction in directions: + values.append( + self.recursive_solve( + coords, self.board[coords[0]][coords[1]], direction, 0 + ) + ) + joined_directions = [ + values[0] + values[4], values[1] + values[5], + values[2] + values[6], values[3] + values[7] + ] + won = any(i >= 3 for i in joined_directions) + if won: + for i, value in enumerate(joined_directions): + if value >= 3: + start_pos = ( + coords[0] + directions[i][0] * values[i], + coords[1] + directions[i][1] * values[i], + ) + end_pos = ( + coords[0] + directions[i + 4][0] * values[i + 4], + coords[1] + directions[i + 4][1] * values[i + 4], + ) + print(start_pos, end_pos) - def check_win(self, coords): - directions = [ - ( 1, 0), - ( 1, 1), - ( 0, 1), - (-1, 1), - (-1, 0), - (-1, -1), - ( 0, -1), - ( 1, -1) - ] - values = list() - for direction in directions: - values.append(self.recursive_solve(coords, self.board[coords[0]][coords[1]], direction, 0)) - joined_directions = [ - values[0] + values[4], - values[1] + values[5], - values[2] + values[6], - values[3] + values[7] - ] - won = any(i >= 3 for i in joined_directions) - if won: - for i, value in enumerate(joined_directions): - if value >= 3: - start_pos = ( - coords[0] + directions[i][0] * values[i], - coords[1] + directions[i][1] * values[i], - ) - end_pos = ( - coords[0] + directions[i+4][0] * values[i+4], - coords[1] + directions[i+4][1] * values[i+4], - ) - print(start_pos, end_pos) + return won - return won + def drop_fisje(self, column, disc): + for row, value in enumerate(self.board): + if self.board[row][column] == EMPTY: + self.board[row][column] = disc + won = self.check_win((row, column)) + print(won) + return - def drop_fisje(self, column, disc): - for row, value in enumerate(self.board): - if self.board[row][column] == EMPTY: - self.board[row][column] = disc - won = self.check_win((row, column)) - print(won) - return def main(): - disc_a = True - gert = bord(11, 8) - while True: - gert.print() - column = int(input("column?: ")) - 1 - os.system("clear") - gert.drop_fisje(column, DISC_A if disc_a else DISC_B) - disc_a = not disc_a + disc_a = True + gert = bord(11, 8) + while True: + gert.print() + column = int(input("column?: ")) - 1 + os.system("clear") + gert.drop_fisje(column, DISC_A if disc_a else DISC_B) + disc_a = not disc_a -if __name__ == "__main__": - main() +if __name__ == "__main__": + main() diff --git a/voerbak/test.py b/voerbak/test.py index 8cc1db9..6481817 100644 --- a/voerbak/test.py +++ b/voerbak/test.py @@ -4,10 +4,9 @@ w = 7 h = 6 column = 3 -process = subprocess.Popen(["./voerbak"], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=None) +process = subprocess.Popen( + ["./voerbak"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=None +) process.stdin.write(bytearray(f"{w} {h}\n", "utf-8")) process.stdin.flush() @@ -21,4 +20,3 @@ process.stdin.flush() # sys.stdout.write(c) print(process.stdout.readlines(5)) - diff --git a/voerbak/v2.py b/voerbak/v2.py index 69e61aa..26030f5 100644 --- a/voerbak/v2.py +++ b/voerbak/v2.py @@ -7,99 +7,104 @@ DISC_A = Fore.RED + DISC_SHAPE + Fore.RESET DISC_B = Fore.BLUE + DISC_SHAPE + Fore.RESET EMPTY = Fore.LIGHTBLACK_EX + "_" + Fore.RESET + class bord: - def __init__(self, w, h): - self.width = w - self.height = h - self.board = [EMPTY] * (w * h) + def __init__(self, w, h): + self.width = w + self.height = h + self.board = [EMPTY] * (w * h) + + def print(self): + for y in range(self.height - 1, -1, -1): + for x in range(self.width): + print(self.board[x + y * self.width], end=" ") + print("\n", end="") - def print(self): - for y in range(self.height -1, -1, -1): - for x in range(self.width): - print(self.board[x + y * self.width], end=" ") - print("\n", end="") + def recursive_solve(self, pos, check_for, direction, current_length): + overflow = (pos % self.width) + direction + if overflow == self.width or overflow == -1: # horizontal overflow + return current_length + new_position = pos + direction + if new_position < 0 or new_position > self.width * self.height - 1: # vertical overflow + return current_length + if self.board[new_position] != check_for: + return current_length + return self.recursive_solve( + new_position, check_for, direction, current_length + 1 + ) - def recursive_solve(self, pos, check_for, direction, current_length): - overflow = (pos % self.width) + direction - if overflow == self.width or overflow == -1: # horizontal overflow - return current_length - new_position = pos + direction - if new_position < 0 or new_position > self.width * self.height - 1: # vertical overflow - return current_length - if self.board[new_position] != check_for: - return current_length - return self.recursive_solve(new_position, check_for, direction, current_length + 1) + def check_win(self, pos): + directions = [ + self.width, # north + self.width + 1, # northeast + 1, # east + -self.width + 1, # southeast + -self.width, # south + -self.width - 1, # southwest + -1, # west + self.width - 1, # northwest + ] + values = list() + for direction in directions: + values.append( + self.recursive_solve(pos, self.board[pos], direction, 0) + ) + joined_directions = [ + values[0] + values[4], values[1] + values[5], + values[2] + values[6], values[3] + values[7] + ] + won = any(i >= 3 for i in joined_directions) + if won: + for i, value in enumerate(joined_directions): + if value >= 3: + start_pos = pos + directions[i] * values[i] + end_pos = pos + directions[i + 4] * values[i + 4] + print(start_pos, end_pos) + self.board[start_pos] = "x" + self.board[end_pos] = "x" + return won - def check_win(self, pos): - directions = [ - self.width, # north - self.width + 1, # northeast - 1, # east - -self.width + 1, # southeast - -self.width, # south - -self.width - 1, # southwest - -1, # west - self.width - 1, # northwest - ] - values = list() - for direction in directions: - values.append(self.recursive_solve(pos, self.board[pos], direction, 0)) - joined_directions = [ - values[0] + values[4], - values[1] + values[5], - values[2] + values[6], - values[3] + values[7] - ] - won = any(i >= 3 for i in joined_directions) - if won: - for i, value in enumerate(joined_directions): - if value >= 3: - start_pos = pos + directions[i] * values[i] - end_pos = pos + directions[i+4] * values[i+4] - print(start_pos, end_pos) - self.board[start_pos] = "x" - self.board[end_pos] = "x" - return won + def debug(self, pos): + self.board[pos] = "x" + directions = [ + self.width, # 0: north + self.width + 1, # 1: northeast + 1, # 2: east + -self.width + 1, # 3: southeast + -self.width, # 4: south + -self.width - 1, # 5: southwest + -1, # 6: west + self.width - 1, # 7: northwest + ] - def debug(self, pos): - self.board[pos] = "x" - directions = [ - self.width, # 0: north - self.width + 1, # 1: northeast - 1, # 2: east - -self.width + 1, # 3: southeast - -self.width, # 4: south - -self.width - 1, # 5: southwest - -1, # 6: west - self.width - 1, # 7: northwest - ] + for index, direction in enumerate(directions): + new_position = pos + direction + if new_position > len(self.board) - 1 or new_position < 0: continue + if index in range(1, 4) and pos % self.width == self.width - 1: + continue + if index in range(5, 8) and pos % self.width == 0: continue + self.board[new_position] = "o" - for index, direction in enumerate(directions): - new_position = pos + direction - if new_position > len(self.board) - 1 or new_position < 0: continue - if index in range(1, 4) and pos % self.width == self.width -1: continue - if index in range(5, 8) and pos % self.width == 0: continue - self.board[new_position] = "o" + def drop_fisje(self, column, disc): + for row in range(self.height): + pos = column + row * self.width + if self.board[pos] == EMPTY: + self.board[pos] = disc + won = self.check_win(pos) + print(won) + return - def drop_fisje(self, column, disc): - for row in range(self.height): - pos = column + row * self.width - if self.board[pos] == EMPTY: - self.board[pos] = disc - won = self.check_win(pos) - print(won) - return def main(): - disc_a = True - gert = bord(7, 6) - for x in range(len(gert.board)): - gert = bord(7, 6) - gert.debug(x) - gert.print() - print("\n\n", end='') - time.sleep(0.1) + disc_a = True + gert = bord(7, 6) + for x in range(len(gert.board)): + gert = bord(7, 6) + gert.debug(x) + gert.print() + print("\n\n", end='') + time.sleep(0.1) -if __name__ == "__main__": - main() +if __name__ == "__main__": + main() |