diff --git a/src/arcaea_offline/database.py b/src/arcaea_offline/database.py index 95dd992..20adeef 100644 --- a/src/arcaea_offline/database.py +++ b/src/arcaea_offline/database.py @@ -8,6 +8,7 @@ from typing import Callable, List, NamedTuple, Optional, TypeVar, Union from thefuzz import fuzz from thefuzz import process as fuzz_process +from .external import ExternalScoreItem from .init_sqls import INIT_SQLS from .models import ( DbAliasRow, @@ -116,6 +117,30 @@ class Database(metaclass=Singleton): conn.commit() self.__trigger_update_hooks() + @check_conn + def import_external(self, external_scores: List[ExternalScoreItem]): + for external in external_scores: + # 2017 Jan. 22, 00:00, UTC+8 + time = external.time if external.time > -1 else 1485014400 + pure = external.pure if external.pure > -1 else None + far = external.far if external.far > -1 else None + lost = external.lost if external.lost > -1 else None + max_recall = external.max_recall if external.max_recall > -1 else None + clear_type = external.clear_type if external.clear_type > -1 else None + + score = ScoreInsert( + song_id=external.song_id, + rating_class=external.rating_class, + score=external.score, + time=time, + pure=pure, + far=far, + lost=lost, + max_recall=max_recall, + clear_type=clear_type, + ) + self.insert_score(score) + @check_conn def init(self): create_sqls = INIT_SQLS[1]["init"] diff --git a/src/arcaea_offline/external.py b/src/arcaea_offline/external.py new file mode 100644 index 0000000..942afd3 --- /dev/null +++ b/src/arcaea_offline/external.py @@ -0,0 +1,60 @@ +import dataclasses +import sqlite3 +from typing import List, Union + + +@dataclasses.dataclass +class ExternalScoreItem: + song_id: str + rating_class: int + score: int + pure: int = -1 + far: int = -1 + lost: int = -1 + max_recall: int = -1 + clear_type: int = -1 + time: int = -1 + + +class ExternalScoreSource: + def get_score_items(self) -> List[ExternalScoreItem]: + ... + + +class St3ScoreSource(ExternalScoreSource): + db_path: Union[str, bytes] + CLEAR_TYPES_MAP = {0: -1, 1: 1, 2: 1, 3: 1, 4: 1, 5: 1} + + def __init__(self, db_path: Union[str, bytes]): + self.db_path = db_path + + def get_score_items(self): + items = [] + with sqlite3.connect(self.db_path) as st3_conn: + cursor = st3_conn.cursor() + db_scores = cursor.execute( + "SELECT songId, songDifficulty, score, perfectCount, nearCount, missCount, date FROM scores" + ).fetchall() + for song_id, rating_class, score, pure, far, lost, date in db_scores: + db_clear_type = cursor.execute( + "SELECT clearType FROM cleartypes WHERE songId = ? AND songDifficulty = ?", + (song_id, rating_class), + ).fetchone()[0] + clear_type = self.CLEAR_TYPES_MAP[db_clear_type] + + date_str = str(date) + date = None if len(date_str) < 7 else int(date_str.ljust(11, "0")) + + kwargs = { + "song_id": song_id, + "rating_class": rating_class, + "score": score, + "pure": pure, + "far": far, + "lost": lost, + "clear_type": clear_type, + } + if date: + kwargs["time"] = date + items.append(ExternalScoreItem(**kwargs)) + return items