1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
import sqlite3, typing
from app_types import *
from datetime import date
class UserRepository:
def __init__(self):
self.__initialize()
def __initialize(self):
cur = self.__get_connection().cursor()
cur.execute("CREATE TABLE IF NOT EXISTS user(chat_id INT, alias TEXT)")
cur.execute("CREATE UNIQUE INDEX IF NOT EXISTS chat_id_idx ON user(chat_id)")
def __get_connection(self):
return sqlite3.connect('db.sqlite')
def get_user(self, chat_id):
connection = self.__get_connection()
cur = connection.cursor()
query = "SELECT * FROM user WHERE chat_id = {chat_id}".format(chat_id = chat_id)
cur.execute(query)
return cur.fetchone()
def get_all_users(self):
connection = self.__get_connection()
cur = connection.cursor()
query = "SELECT * FROM user"
cur.execute(query)
return cur.fetchall()
def remove_chat(self, chat_id):
connection = self.__get_connection()
cur = connection.cursor()
query = "DELETE FROM user WHERE chat_id = {chat_id}".format(
chat_id = chat_id)
cur.execute(query)
connection.commit()
def put_user(self, chat_id, alias):
if not self.get_user(chat_id):
connection = self.__get_connection()
cur = connection.cursor()
query = "INSERT INTO user(chat_id, alias) VALUES({chat_id}, '{text}')".format(
chat_id = chat_id,
text = alias)
cur.execute(query)
connection.commit()
class StatsRepository:
def __init__(self):
self.__initialize()
def __get_connection(self):
return sqlite3.connect('db.sqlite')
def __initialize(self):
cur = self.__get_connection().cursor()
cur.execute("CREATE TABLE IF NOT EXISTS stats(user_id TEXT, stats_date DATE, display_name TEXT, level INT, matches_played INT, kills INT, wins INT)")
cur.execute("CREATE UNIQUE INDEX IF NOT EXISTS user_id_date_idx ON stats(user_id, stats_date)")
async def put_stats(self, user: User, stats_date: date = date.today()):
stats: UserStats = await user.fetch_stats()
connection = self.__get_connection()
cur = connection.cursor()
query = "INSERT OR REPLACE INTO stats(user_id, stats_date, display_name, level, matches_played, kills, wins) VALUES('{user_id}', '{stats_date}', '{display_name}', {level}, {matches_played}, {kills}, {wins})".format(
user_id = user.id,
stats_date = stats_date,
display_name = user.display_name,
level = stats.level,
matches_played = stats.matches_played,
kills = stats.kills,
wins = stats.wins)
cur.execute(query)
connection.commit()
def get_stats(self, stats_date: date = date.today()) -> typing.List[UserStats]:
connection = self.__get_connection()
cur = connection.cursor()
query = "SELECT * FROM stats WHERE stats_date IN (SELECT MIN(stats_date) FROM stats WHERE stats_date >= '{stats_date}')".format(stats_date = stats_date)
cur.execute(query)
result = cur.fetchall()
return [self.__map_from_db(stats) for stats in result]
def __map_from_db(self, record):
user_stats = UserStats()
user_stats.user_id = str(record[0])
user_stats.user_display_name = str(record[2])
user_stats.level = int(record[3])
user_stats.matches_played = int(record[4])
user_stats.kills = int(record[5])
user_stats.wins = int(record[6])
return user_stats
class PresenceRepository:
def __init__(self):
self.__initialize()
def __get_connection(self):
return sqlite3.connect('db.sqlite')
def __initialize(self):
cur = self.__get_connection().cursor()
cur.execute("CREATE TABLE IF NOT EXISTS user_presence(display_name TEXT, last_online INT)")
cur.execute("CREATE UNIQUE INDEX IF NOT EXISTS display_name_idx ON user_presence(display_name)")
def get_last_user_presence(self, display_name):
connection = self.__get_connection()
cur = connection.cursor()
query = "select last_online from user_presence where display_name = '{display_name}'".format(display_name = display_name)
cur.execute(query)
user_presence = cur.fetchone()
if user_presence:
return user_presence[0]
else:
return 0
def set_last_user_presence(self, display_name: str, last_online: int):
connection = self.__get_connection()
cur = connection.cursor()
query = "INSERT OR REPLACE INTO user_presence(display_name, last_online) VALUES('{display_name}', {last_online})".format(
display_name = display_name,
last_online = last_online)
cur.execute(query)
connection.commit()
|
