280 lines
7.2 KiB
Python
280 lines
7.2 KiB
Python
import sqlite3
|
|
from sqlite3 import Error
|
|
import os
|
|
import urllib.parse
|
|
|
|
dir_name = os.path.dirname(__file__)
|
|
db_name = r"boggle.sqlite3"
|
|
db_file = os.path.join(dir_name, db_name)
|
|
|
|
def create_connection():
|
|
""" create a database connection to the SQLite database
|
|
specified by db_file
|
|
:return: Connection object or None
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = sqlite3.connect(db_file)
|
|
except Error as e:
|
|
print(e)
|
|
return conn
|
|
|
|
def add_user(conn):
|
|
"""
|
|
Create a new user record
|
|
:param conn: database connection object
|
|
:return: user id
|
|
"""
|
|
sql = ''' INSERT INTO users DEFAULT VALUES '''
|
|
cur = conn.cursor()
|
|
try:
|
|
cur.execute(sql)
|
|
conn.commit()
|
|
except:
|
|
return 'error in insert'
|
|
return cur.lastrowid
|
|
|
|
def add_hash(conn, tls_client_hash, user_id):
|
|
"""
|
|
Create a new record for this cert
|
|
:param conn: database connection object
|
|
:param tls_client_hash:
|
|
:param user_id:
|
|
:return: certificate id
|
|
"""
|
|
sql = ''' INSERT OR REPLACE INTO certs(hash,user_id) VALUES(?,?) '''
|
|
cur = conn.cursor()
|
|
cur.execute(sql, (tls_client_hash, user_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def get_user(conn, tls_client_hash):
|
|
"""
|
|
Get user id matching tls_client_hash if it exists
|
|
:param conn: database connection object
|
|
:param tls_client_hash:
|
|
:return: user id or None
|
|
"""
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT user_id FROM certs WHERE hash=?", (tls_client_hash,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # user_id
|
|
|
|
def get_user_by_keycode(conn, key_code):
|
|
"""
|
|
Get user id of tls_client_hash if it exists
|
|
:param conn: database connection object
|
|
:param key_code: authorization key code
|
|
:return: user id or None
|
|
"""
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT id FROM users WHERE add_key_code=?", (key_code,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # user_id
|
|
|
|
def check_hash(tls_client_hash):
|
|
"""
|
|
Check for existing user with hash or add a new one
|
|
:param conn:
|
|
:param tls_client_hash:
|
|
:return: user id
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
user_id = get_user(conn, tls_client_hash)
|
|
if (user_id is None):
|
|
user_id = add_user(conn)
|
|
add_hash(conn, tls_client_hash, user_id)
|
|
return user_id
|
|
|
|
def get_name(user_id):
|
|
"""
|
|
Get user name if it exists
|
|
:param user_id:
|
|
:return: user name
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT name FROM users WHERE id=?", (user_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # user_name
|
|
|
|
def set_name(user_id, user_name):
|
|
"""
|
|
Update or set name on user
|
|
:param user_id:
|
|
:param user_name:
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
sql = ''' UPDATE users SET name=(?) WHERE id=? '''
|
|
cur = conn.cursor()
|
|
cur.execute(sql, (urllib.parse.unquote(user_name), user_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
|
|
def set_add_key_code(user_id, key_code):
|
|
"""
|
|
Update or set key code on user
|
|
:param user_id:
|
|
:param key_code:
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
""" There should probably be a check here to ensure no user
|
|
has this key_code already
|
|
"""
|
|
sql = ''' UPDATE users SET add_key_code=(?) WHERE id=? '''
|
|
cur = conn.cursor()
|
|
cur.execute(sql, (key_code, user_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def add_cert_to_user(key_code, tls_client_hash):
|
|
"""
|
|
Check for existing user with hash or add a new one
|
|
:param conn:
|
|
:param key_code:
|
|
:param tls_client_hash:
|
|
:return: true if added, false if not found
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
user_id = get_user_by_keycode(conn, key_code)
|
|
if (user_id is None):
|
|
return False
|
|
add_hash(conn, tls_client_hash, user_id)
|
|
return True
|
|
|
|
|
|
|
|
def create_new_game(conn,board,words,graph):
|
|
"""
|
|
Create a new game record
|
|
:param conn: database connection object
|
|
:param board: game board
|
|
:param words: found words
|
|
:param graph: word guessed list
|
|
:return: game id
|
|
"""
|
|
sql = ''' INSERT INTO games(board,words,graph,stats) VALUES(?,?,?,?) '''
|
|
cur = conn.cursor()
|
|
cur.execute(sql,
|
|
(board,words,graph,"PLAY")
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def set_current_game(user_id,game_id):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("UPDATE users SET game_id=? WHERE id=?", (game_id,user_id))
|
|
|
|
|
|
def get_game_id(user_id):
|
|
"""
|
|
Get game id if it exists
|
|
:param user_id:.
|
|
:return: game id
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT game_id FROM users WHERE id=?", (user_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # game_id
|
|
|
|
def get_words(game_id,user_id):
|
|
"""
|
|
Get game id if it exists
|
|
:param user_id:.
|
|
:return: game id
|
|
"""
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT words FROM plays WHERE game_id=? and user_id=?", (game_id,user_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # game_id
|
|
|
|
def add_plays(game_id,user_id,words):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("INSERT OR REPLACE INTO plays(game_id,user_id,words) values(?,?,?)", (game_id,user_id,words))
|
|
|
|
def set_game_graph(game_id,graph):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("UPDATE games SET graph=? WHERE id=?", (graph,game_id))
|
|
|
|
def get_game_graph(game_id):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT graph FROM games WHERE id=?", (game_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # game_id
|
|
|
|
|
|
def set_game_stats(game_id,stats):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("UPDATE games SET stats=? WHERE id=?", (stats,game_id))
|
|
|
|
def get_game_stats(game_id):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT stats FROM games WHERE id=?", (game_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # game_id
|
|
|
|
def add_score(user_id,points):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("UPDATE users SET score=score+? WHERE id=?", (points,user_id))
|
|
|
|
def get_score(user_id):
|
|
conn = create_connection()
|
|
with conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT score FROM users WHERE id=?", (user_id,))
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
return None
|
|
else:
|
|
return row[0] # game_id
|
|
|
|
|
|
|
|
#vim:fenc=utf-8:ts=4:sw=4:sta:noet:sts=4:fdm=marker:ai
|