ED/SquadsActivityMonitor
1
0
Fork 0
mirror of https://github.com/norohind/SquadsActivityMonitor.git synced 2025-04-14 13:57:17 +03:00
Code Issues Projects Releases Wiki Activity

move insert_leaderboard_db from main.py to model.py

Browse Source
This commit is contained in:
norohind 2021-11-21 00:50:20 +03:00
parent 5b1216dd41
commit 56bd4f335b
Signed by: norohind
GPG Key ID: 01C3BECC26FB59E1
2 changed files with 40 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
main.py
View File

@ -23,15 +23,13 @@ import typing
import requests import requests
import sqlite3 import sqlite3
import model
import sql_requests import sql_requests
# from EDMCLogging import get_main_logger # from EDMCLogging import get_main_logger
import utils import utils
db: sqlite3.Connection = sqlite3.connect('squads_stat.sqlite3') db: sqlite3.Connection = sqlite3.connect('squads_stat.sqlite3')
db.executescript(sql_requests.schema_create)
def request_leaderboard(platform_enum: utils.Platform, leaderboard_type_enum: utils.LeaderboardTypes) -> dict: def request_leaderboard(platform_enum: utils.Platform, leaderboard_type_enum: utils.LeaderboardTypes) -> dict:
""" """
@ -55,57 +53,18 @@ def request_leaderboard(platform_enum: utils.Platform, leaderboard_type_enum: ut
'type': leaderboard_type # str 'type': leaderboard_type # str
} }
# return SAPIRequest.json()['leaderboards'][leaderboard_type]
def get_and_save_leaderboard(platform_enum: utils.Platform, leaderboard_type_enum: utils.LeaderboardTypes) -> None:
def insert_leaderboard_db(db_conn: sqlite3.Connection, leaderboard_list: dict) -> None:
"""
Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
:param db_conn: Connection to DB
:param leaderboard_list: list from request_leaderboard
:return:
"""
platform: str = leaderboard_list['platform']
LB_type: str = leaderboard_list['type']
leaderboard: list = leaderboard_list['leaderboard']
action_id: int # not last, current that we will use
sql_req_action_id: sqlite3.Cursor = db_conn.execute(sql_requests.select_last_action_id)
action_id_fetch_one: typing.Union[None, tuple[int]] = sql_req_action_id.fetchone()
if action_id_fetch_one is None:
# i.e. first launch
action_id = 1 # yep, not 0
else:
action_id = action_id_fetch_one[0] + 1
# Patch for additional values
for squad in leaderboard:
squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
with db_conn:
db_conn.executemany(
sql_requests.insert_leader_board,
leaderboard)
def get_and_save_leaderboard(platform_enum: utils.Platform,
leaderboard_type_enum: utils.LeaderboardTypes,
db_conn: sqlite3.Connection) -> None:
""" """
High logic function to get and save information about specified for type and platform leaderboard High logic function to get and save information about specified for type and platform leaderboard
:param platform_enum: :param platform_enum:
:param leaderboard_type_enum: :param leaderboard_type_enum:
:param db_conn:
:return: :return:
""" """
req = request_leaderboard(platform_enum, leaderboard_type_enum) req = request_leaderboard(platform_enum, leaderboard_type_enum)
insert_leaderboard_db(db_conn, req) model.insert_leaderboard_db(req)
def main(): def main():

39
model.py
View File

@ -1,15 +1,17 @@
import sqlite3 import sqlite3
import typing
import sql_requests import sql_requests
import utils import utils
db: sqlite3.Connection = sqlite3.connect('squads_stat.sqlite3', check_same_thread=False) db: sqlite3.Connection = sqlite3.connect('squads_stat.sqlite3', check_same_thread=False)
db.executescript(sql_requests.schema_create) # schema creation
# thx https://stackoverflow.com/a/48789604 # thx https://stackoverflow.com/a/48789604
db.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r)) db.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
cur = db.cursor()
def get_activity_changes(platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp) -> list: def get_activity_changes(platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp) -> list:
sql_req: sqlite3.Cursor = db.execute(sql_requests.select_activity, { sql_req: sqlite3.Cursor = db.execute(sql_requests.select_activity, {
@ -21,3 +23,36 @@ def get_activity_changes(platform: str, leaderboard_type: str, limit: int, low_t
}) })
return sql_req.fetchall() return sql_req.fetchall()
def insert_leaderboard_db(leaderboard_list: dict) -> None:
"""
Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
:param leaderboard_list: list from request_leaderboard
:return:
"""
platform: str = leaderboard_list['platform']
LB_type: str = leaderboard_list['type']
leaderboard: list = leaderboard_list['leaderboard']
action_id: int # not last, current that we will use
sql_req_action_id: sqlite3.Cursor = db.execute(sql_requests.select_last_action_id)
action_id_fetch_one: typing.Union[None, dict[str, int]] = sql_req_action_id.fetchone()
if action_id_fetch_one is None:
# i.e. first launch
action_id = 1 # yep, not 0
else:
action_id = action_id_fetch_one['action_id'] + 1
# Patch for additional values
for squad in leaderboard:
squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
with db:
db.executemany(
sql_requests.insert_leader_board,
leaderboard)