283375/arcaea-offline
1
0
Fork 0
mirror of https://github.com/283375/arcaea-offline.git synced 2025-07-01 12:16:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(db): use cursor to run queries

Browse Source
This commit is contained in:
283375
2023-06-06 15:17:14 +08:00
parent eee196d0ba
commit e874e38d3b
1 changed files with 219 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
src/arcaea_offline/database.py
View File

@ -24,8 +24,9 @@ class Database(metaclass=Singleton):
return self.__conn return self.__conn
def validate_song_id(self, song_id): def validate_song_id(self, song_id):
with self.conn: with self.conn as conn:
result = self.conn.execute( cursor = conn.cursor()
result = cursor.execute(
"SELECT song_id FROM charts WHERE song_id = ?", (song_id,) "SELECT song_id FROM charts WHERE song_id = ?", (song_id,)
).fetchall() ).fetchall()
return len(result) > 0 return len(result) > 0
@ -34,19 +35,20 @@ class Database(metaclass=Singleton):
arcsong_conn = sqlite3.connect(path) arcsong_conn = sqlite3.connect(path)
data = {} data = {}
with arcsong_conn: with arcsong_conn:
data["charts"] = arcsong_conn.execute( arcsong_cursor = arcsong_conn.cursor()
data["charts"] = arcsong_cursor.execute(
""" """
SELECT song_id, rating_class, name_en, name_jp, artist, bpm, bpm_base, "set", time, side, world_unlock, remote_download, bg, date, version, difficulty, rating, note, chart_designer, jacket_designer, jacket_override, audio_override SELECT song_id, rating_class, name_en, name_jp, artist, bpm, bpm_base, "set", time, side, world_unlock, remote_download, bg, date, version, difficulty, rating, note, chart_designer, jacket_designer, jacket_override, audio_override
FROM charts FROM charts
""" """
).fetchall() ).fetchall()
data["aliases"] = arcsong_conn.execute( data["aliases"] = arcsong_cursor.execute(
""" """
SELECT sid, alias SELECT sid, alias
FROM alias FROM alias
""" """
).fetchall() ).fetchall()
data["packages"] = arcsong_conn.execute( data["packages"] = arcsong_cursor.execute(
""" """
SELECT id, name SELECT id, name
FROM packages FROM packages
@ -54,10 +56,11 @@ class Database(metaclass=Singleton):
).fetchall() ).fetchall()
with self.conn as conn: with self.conn as conn:
cursor = conn.cursor()
for table in data: for table in data:
columns = [ columns = [
row[0] row[0]
for row in conn.execute( for row in cursor.execute(
f"SELECT * FROM {table} LIMIT 1" f"SELECT * FROM {table} LIMIT 1"
).description ).description
] ]
@ -68,13 +71,13 @@ class Database(metaclass=Singleton):
update_clauses = ", ".join( update_clauses = ", ".join(
[f"{column} = excluded.{column}" for column in columns] [f"{column} = excluded.{column}" for column in columns]
) )
conn.executemany( cursor.executemany(
f"INSERT INTO {table} ({columns_insert_str}) VALUES ({values_insert_str}) ON CONFLICT DO UPDATE SET {update_clauses}", f"INSERT INTO {table} ({columns_insert_str}) VALUES ({values_insert_str}) ON CONFLICT DO UPDATE SET {update_clauses}",
data[table], data[table],
) )
conn.commit()
def init(self): def init(self):
with self.conn as conn:
create_sqls = [ create_sqls = [
""" """
CREATE TABLE IF NOT EXISTS charts ( CREATE TABLE IF NOT EXISTS charts (
@ -225,8 +228,10 @@ class Database(metaclass=Singleton):
""" """
] ]
with self.conn as conn:
cursor = conn.cursor()
for sql in create_sqls: for sql in create_sqls:
conn.execute(sql) cursor.execute(sql)
conn.commit() conn.commit()
def __get_columns_from_dataclass(self, dataclass) -> List[str]: def __get_columns_from_dataclass(self, dataclass) -> List[str]:
@ -243,9 +248,10 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbPackageRow) self.__get_columns_from_dataclass(DbPackageRow)
) )
cursor = conn.cursor()
return [ return [
DbPackageRow(*row) DbPackageRow(*row)
for row in conn.execute( for row in cursor.execute(
f"SELECT {columns_clause} FROM packages" f"SELECT {columns_clause} FROM packages"
).fetchall() ).fetchall()
] ]
@ -255,9 +261,10 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbAliasRow) self.__get_columns_from_dataclass(DbAliasRow)
) )
cursor = conn.cursor()
return [ return [
DbAliasRow(*row) DbAliasRow(*row)
for row in conn.execute( for row in cursor.execute(
f"SELECT {columns_clause} FROM aliases" f"SELECT {columns_clause} FROM aliases"
).fetchall() ).fetchall()
] ]
@ -267,10 +274,11 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbAliasRow) self.__get_columns_from_dataclass(DbAliasRow)
) )
cursor = conn.cursor()
return [ return [
DbAliasRow(*row) DbAliasRow(*row)
for row in ( for row in (
conn.execute( cursor.execute(
f"SELECT {columns_clause} FROM aliases WHERE song_id = ?", f"SELECT {columns_clause} FROM aliases WHERE song_id = ?",
(song_id,), (song_id,),
).fetchall() ).fetchall()
@ -282,9 +290,10 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbChartRow) self.__get_columns_from_dataclass(DbChartRow)
) )
cursor = conn.cursor()
return [ return [
DbChartRow(*row) DbChartRow(*row)
for row in conn.execute( for row in cursor.execute(
f"SELECT {columns_clause} FROM charts" f"SELECT {columns_clause} FROM charts"
).fetchall() ).fetchall()
] ]
@ -294,10 +303,11 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbChartRow) self.__get_columns_from_dataclass(DbChartRow)
) )
cursor = conn.cursor()
return [ return [
DbChartRow(*row) DbChartRow(*row)
for row in ( for row in (
conn.execute( cursor.execute(
f"SELECT {columns_clause} FROM charts WHERE song_id = ?", f"SELECT {columns_clause} FROM charts WHERE song_id = ?",
(song_id,), (song_id,),
).fetchall() ).fetchall()
@ -309,9 +319,10 @@ class Database(metaclass=Singleton):
columns_clause = self.__get_columns_clause( columns_clause = self.__get_columns_clause(
self.__get_columns_from_dataclass(DbChartRow) self.__get_columns_from_dataclass(DbChartRow)
) )
cursor = conn.cursor()
return [ return [
DbChartRow(*row) DbChartRow(*row)
for row in conn.execute( for row in cursor.execute(
f"SELECT {columns_clause} FROM charts WHERE package_id = ?", f"SELECT {columns_clause} FROM charts WHERE package_id = ?",
(package_id,), (package_id,),
).fetchall() ).fetchall()
@ -323,7 +334,8 @@ class Database(metaclass=Singleton):
def fuzzy_search_song_id(self, input_str: str, limit: int= 5) -> List[FuzzySearchSongIdResult]: def fuzzy_search_song_id(self, input_str: str, limit: int= 5) -> List[FuzzySearchSongIdResult]:
with self.conn as conn: with self.conn as conn:
db_results = conn.execute("SELECT song_id, name FROM song_id_names").fetchall() cursor = conn.cursor()
db_results = cursor.execute("SELECT song_id, name FROM song_id_names").fetchall()
name_song_id_map = {r[1]: r[0] for r in db_results} name_song_id_map = {r[1]: r[0] for r in db_results}
names = name_song_id_map.keys() names = name_song_id_map.keys()
fuzzy_results = fuzz_process.extractBests(input_str, names, scorer=fuzz.partial_ratio, limit=limit) # type: ignore fuzzy_results = fuzz_process.extractBests(input_str, names, scorer=fuzz.partial_ratio, limit=limit) # type: ignore
@ -343,7 +355,6 @@ class Database(metaclass=Singleton):
song_id: Optional[List[str]] = None, song_id: Optional[List[str]] = None,
rating_class: Optional[List[int]] = None, rating_class: Optional[List[int]] = None,
): ):
with self.conn as conn:
columns = ",".join([f"[{field.name}]" for field in fields(DbScoreRow)]) columns = ",".join([f"[{field.name}]" for field in fields(DbScoreRow)])
where_clauses = [] where_clauses = []
params = [] params = []
@ -359,8 +370,10 @@ class Database(metaclass=Singleton):
if where_clauses: if where_clauses:
final_sql += " WHERE " final_sql += " WHERE "
final_sql += " AND ".join(where_clauses) final_sql += " AND ".join(where_clauses)
with self.conn as conn:
cursor = conn.cursor()
return [ return [
DbScoreRow(*row) for row in conn.execute(final_sql, params).fetchall() DbScoreRow(*row) for row in cursor.execute(final_sql, params).fetchall()
] ]
def get_calculated( def get_calculated(
@ -369,7 +382,6 @@ class Database(metaclass=Singleton):
song_id: Optional[List[str]] = None, song_id: Optional[List[str]] = None,
rating_class: Optional[List[int]] = None, rating_class: Optional[List[int]] = None,
): ):
with self.conn as conn:
columns = ",".join([f"[{field.name}]" for field in fields(DbCalculatedRow)]) columns = ",".join([f"[{field.name}]" for field in fields(DbCalculatedRow)])
where_clauses = [] where_clauses = []
params = [] params = []
@ -385,29 +397,35 @@ class Database(metaclass=Singleton):
if where_clauses: if where_clauses:
final_sql += " WHERE " final_sql += " WHERE "
final_sql += " AND ".join(where_clauses) final_sql += " AND ".join(where_clauses)
with self.conn as conn:
cursor = conn.cursor()
return [ return [
DbCalculatedRow(*row) DbCalculatedRow(*row)
for row in conn.execute(final_sql, params).fetchall() for row in cursor.execute(final_sql, params).fetchall()
] ]
def get_b30(self) -> float: def get_b30(self) -> float:
with self.conn as conn: with self.conn as conn:
return conn.execute("SELECT b30 FROM calculated_potential").fetchone()[0] cursor = conn.cursor()
return cursor.execute("SELECT b30 FROM calculated_potential").fetchone()[0]
def get_r10(self) -> float: def get_r10(self) -> float:
with self.conn as conn: with self.conn as conn:
return conn.execute("SELECT r10 FROM calculated_potential").fetchone()[0] cursor = conn.cursor()
return cursor.execute("SELECT r10 FROM calculated_potential").fetchone()[0]
def get_potential(self) -> float: def get_potential(self) -> float:
with self.conn as conn: with self.conn as conn:
return conn.execute("SELECT potential FROM calculated_potential").fetchone()[0] cursor = conn.cursor()
return cursor.execute("SELECT potential FROM calculated_potential").fetchone()[0]
def insert_score(self, score: DbScoreRow): def insert_score(self, score: DbScoreRow):
columns = self.__get_columns_from_dataclass(DbScoreRow) columns = self.__get_columns_from_dataclass(DbScoreRow)
columns_clause = self.__get_columns_clause(columns) columns_clause = self.__get_columns_clause(columns)
params = [score.__getattribute__(column) for column in columns] params = [score.__getattribute__(column) for column in columns]
with self.conn as conn: with self.conn as conn:
conn.execute( cursor = conn.cursor()
cursor.execute(
f"INSERT INTO scores({columns_clause}) VALUES ({', '.join('?' * len(params))})", f"INSERT INTO scores({columns_clause}) VALUES ({', '.join('?' * len(params))})",
params, params,
) )