aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorlonkaars <l.leblansch@gmail.com>2021-04-11 17:50:58 +0200
committerlonkaars <l.leblansch@gmail.com>2021-04-11 17:50:58 +0200
commit28f104de9ae9abe4b42abafbf3865ede5687996c (patch)
tree65e651f09d8fbf81380384692e45803cb4f9d61c
parent7b4859059b3bbabf4139ccdf3270a82c094f5d8e (diff)
dprint yapf python formatting
-rw-r--r--api/auth/login.py74
-rw-r--r--api/auth/login_token.py23
-rw-r--r--api/auth/signup.py124
-rw-r--r--api/auth/token.py60
-rw-r--r--api/db.py1
-rw-r--r--api/dynamic_import.py35
-rw-r--r--api/events.py27
-rw-r--r--api/game/accept.py17
-rw-r--r--api/game/cleanup.py34
-rw-r--r--api/game/info.py120
-rw-r--r--api/game/new.py61
-rw-r--r--api/game/random.py36
-rw-r--r--api/game/socket.py170
-rw-r--r--api/game/voerbak_connector.py137
-rw-r--r--api/hierarchy.py32
-rw-r--r--api/log.py1
-rw-r--r--api/main.py3
-rw-r--r--api/passwords.py10
-rw-r--r--api/randid.py24
-rw-r--r--api/rating.py53
-rw-r--r--api/ruleset.py34
-rw-r--r--api/social/create_relation.py78
-rw-r--r--api/social/destroy_relation.py32
-rw-r--r--api/social/friend_accept.py19
-rw-r--r--api/social/request_list.py21
-rw-r--r--api/social/search.py34
-rw-r--r--api/socket_io.py1
-rw-r--r--api/status.py16
-rw-r--r--api/user/avatar.py37
-rw-r--r--api/user/games.py116
-rw-r--r--api/user/info.py145
-rw-r--r--api/user/password.py13
-rw-r--r--api/user/preferences.py51
-rw-r--r--api/user/status.py20
-rw-r--r--database/database.py418
-rw-r--r--dprint.json3
-rw-r--r--voerbak/main.py149
-rw-r--r--voerbak/test.py8
-rw-r--r--voerbak/v2.py175
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)
+ }
diff --git a/api/db.py b/api/db.py
index b850a89..a7d7d86 100644
--- a/api/db.py
+++ b/api/db.py
@@ -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
diff --git a/api/log.py b/api/log.py
index 5b573b0..1b60670 100644
--- a/api/log.py
+++ b/api/log.py
@@ -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()