ci: tox pytest command

* packlist & songlist importer

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       - id: trailing-whitespace
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.8
+    rev: v0.4.4
     hooks:
       - id: ruff
         args: ["--fix"]

pyproject.toml

@@ -16,7 +16,7 @@ classifiers = [
 ]
 
 [project.optional-dependencies]
-dev = ["ruff~=0.6.8", "pre-commit~=3.3", "pytest~=7.4", "tox~=4.11"]
+dev = ["ruff~=0.4", "pre-commit~=3.3", "pytest~=7.4", "tox~=4.11"]
 
 [project.urls]
 "Homepage" = "https://github.com/283375/arcaea-offline"

requirements.dev.txt

@@ -1,4 +1,4 @@
-ruff~=0.6.8
+ruff~=0.4
 pre-commit~=3.3
 pytest~=7.4
 tox~=4.11

src/arcaea_offline/external/arcaea/__init__.py

@@ -0,0 +1,12 @@
+from .online import ArcaeaOnlineParser
+from .packlist import PacklistParser
+from .songlist import SonglistDifficultiesParser, SonglistParser
+from .st3 import St3ScoreParser
+
+__all__ = [
+    "ArcaeaOnlineParser",
+    "PacklistParser",
+    "SonglistDifficultiesParser",
+    "SonglistParser",
+    "St3ScoreParser",
+]

src/arcaea_offline/external/arcaea/common.py

@@ -0,0 +1,99 @@
+import contextlib
+import json
+import math
+import time
+from os import PathLike
+from typing import Any, List, Optional, Union
+
+from sqlalchemy.orm import DeclarativeBase, Session
+
+
+def fix_timestamp(timestamp: int) -> Union[int, None]:
+    """
+    Some of the `date` column in st3 are strangely truncated. For example,
+    a `1670283375` may be truncated to `167028`, even `1`. Yes, a single `1`.
+
+    To properly handle this situation, we check the timestamp's digits.
+    If `digits < 5`, we treat this timestamp as a `None`. Otherwise, we try to
+    fix the timestamp.
+
+    :param timestamp: a POSIX timestamp
+    :return: `None` if the timestamp's digits < 5, otherwise a fixed POSIX timestamp
+    """
+    # find digit length from https://stackoverflow.com/a/2189827/16484891
+    # CC BY-SA 2.5
+    # this might give incorrect result when timestamp > 999999999999997,
+    # see https://stackoverflow.com/a/28883802/16484891 (CC BY-SA 4.0).
+    # but that's way too later than 9999-12-31 23:59:59, 253402271999,
+    # I don't think Arcaea would still be an active updated game by then.
+    # so don't mind those small issues, just use this.
+    digits = int(math.log10(abs(timestamp))) + 1 if timestamp != 0 else 1
+    if digits < 5:
+        return None
+    timestamp_str = str(timestamp)
+    current_timestamp_digits = int(math.log10(int(time.time()))) + 1
+    timestamp_str = timestamp_str.ljust(current_timestamp_digits, "0")
+    return int(timestamp_str, 10)
+
+
+def to_db_value(val: Any) -> Any:
+    if not val:
+        return None
+    return json.dumps(val, ensure_ascii=False) if isinstance(val, list) else val
+
+
+def is_localized(item: dict, key: str, append_localized: bool = True):
+    item_key = f"{key}_localized" if append_localized else key
+    subitem: Optional[dict] = item.get(item_key)
+    return subitem and (
+        subitem.get("ja")
+        or subitem.get("ko")
+        or subitem.get("zh-Hant")
+        or subitem.get("zh-Hans")
+    )
+
+
+def set_model_localized_attrs(
+    model: DeclarativeBase, item: dict, model_key: str, item_key: Optional[str] = None
+):
+    if item_key is None:
+        item_key = f"{model_key}_localized"
+    subitem: dict = item.get(item_key, {})
+    if not subitem:
+        return
+    setattr(model, f"{model_key}_ja", to_db_value(subitem.get("ja")))
+    setattr(model, f"{model_key}_ko", to_db_value(subitem.get("ko")))
+    setattr(model, f"{model_key}_zh_hans", to_db_value(subitem.get("zh-Hans")))
+    setattr(model, f"{model_key}_zh_hant", to_db_value(subitem.get("zh-Hant")))
+
+
+class ArcaeaParser:
+    def __init__(self, filepath: Union[str, bytes, PathLike]):
+        self.filepath = filepath
+
+    def read_file_text(self):
+        file_handle = None
+
+        with contextlib.suppress(TypeError):
+            # original open
+            file_handle = open(self.filepath, "r", encoding="utf-8")
+
+        if file_handle is None:
+            try:
+                # or maybe a `pathlib.Path` subset
+                # or an `importlib.resources.abc.Traversable` like object
+                # e.g. `zipfile.Path`
+                file_handle = self.filepath.open(mode="r", encoding="utf-8")  # type: ignore
+            except Exception as e:
+                raise ValueError("Invalid `filepath`.") from e
+
+        with file_handle:
+            return file_handle.read()
+
+    def parse(self) -> List[DeclarativeBase]:
+        raise NotImplementedError()
+
+    def write_database(self, session: Session):
+        results = self.parse()
+        for result in results:
+            session.merge(result)

src/arcaea_offline/external/arcaea/online.py

@@ -0,0 +1,72 @@
+import json
+import logging
+from datetime import datetime
+from typing import Dict, List, Literal, Optional, TypedDict
+
+from ...models import Score
+from .common import ArcaeaParser, fix_timestamp
+
+logger = logging.getLogger(__name__)
+
+
+class TWebApiRatingMeScoreItem(TypedDict):
+    song_id: str
+    difficulty: int
+    modifier: int
+    rating: float
+    score: int
+    perfect_count: int
+    near_count: int
+    miss_count: int
+    clear_type: int
+    title: Dict[Literal["ja", "en"], str]
+    artist: str
+    time_played: int
+    bg: str
+
+
+class TWebApiRatingMeValue(TypedDict):
+    best_rated_scores: List[TWebApiRatingMeScoreItem]
+    recent_rated_scores: List[TWebApiRatingMeScoreItem]
+
+
+class TWebApiRatingMeResult(TypedDict):
+    success: bool
+    error_code: Optional[int]
+    value: Optional[TWebApiRatingMeValue]
+
+
+class ArcaeaOnlineParser(ArcaeaParser):
+    def parse(self) -> List[Score]:
+        api_result_root: TWebApiRatingMeResult = json.loads(self.read_file_text())
+
+        api_result_value = api_result_root.get("value")
+        if not api_result_value:
+            error_code = api_result_root.get("error_code")
+            raise ValueError(f"Cannot parse API result, error code {error_code}")
+
+        best30_score_items = api_result_value.get("best_rated_scores", [])
+        recent_score_items = api_result_value.get("recent_rated_scores", [])
+        score_items = best30_score_items + recent_score_items
+
+        date_text = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+        results: List[Score] = []
+        for score_item in score_items:
+            score = Score()
+            score.song_id = score_item["song_id"]
+            score.rating_class = score_item["difficulty"]
+            score.score = score_item["score"]
+            score.pure = score_item["perfect_count"]
+            score.far = score_item["near_count"]
+            score.lost = score_item["miss_count"]
+            score.date = fix_timestamp(int(score_item["time_played"] / 1000))
+            score.modifier = score_item["modifier"]
+            score.clear_type = score_item["clear_type"]
+
+            if score.lost == 0:
+                score.max_recall = score.pure + score.far
+
+            score.comment = f"Parsed from web API at {date_text}"
+            results.append(score)
+        return results

src/arcaea_offline/external/arcaea/packlist.py

@@ -0,0 +1,29 @@
+import json
+from typing import List, Union
+
+from ...models.songs import Pack, PackLocalized
+from .common import ArcaeaParser, is_localized, set_model_localized_attrs
+
+
+class PacklistParser(ArcaeaParser):
+    def parse(self) -> List[Union[Pack, PackLocalized]]:
+        packlist_json_root = json.loads(self.read_file_text())
+
+        packlist_json = packlist_json_root["packs"]
+        results: List[Union[Pack, PackLocalized]] = [
+            Pack(id="single", name="Memory Archive")
+        ]
+        for item in packlist_json:
+            pack = Pack()
+            pack.id = item["id"]
+            pack.name = item["name_localized"]["en"]
+            pack.description = item["description_localized"]["en"] or None
+            results.append(pack)
+
+            if is_localized(item, "name") or is_localized(item, "description"):
+                pack_localized = PackLocalized(id=pack.id)
+                set_model_localized_attrs(pack_localized, item, "name")
+                set_model_localized_attrs(pack_localized, item, "description")
+                results.append(pack_localized)
+
+        return results

src/arcaea_offline/external/arcaea/songlist.py

@@ -0,0 +1,101 @@
+import json
+from typing import List, Union
+
+from ...models.songs import Difficulty, DifficultyLocalized, Song, SongLocalized
+from .common import ArcaeaParser, is_localized, set_model_localized_attrs, to_db_value
+
+
+class SonglistParser(ArcaeaParser):
+    def parse(
+        self,
+    ) -> List[Union[Song, SongLocalized, Difficulty, DifficultyLocalized]]:
+        songlist_json_root = json.loads(self.read_file_text())
+
+        songlist_json = songlist_json_root["songs"]
+        results = []
+        for item in songlist_json:
+            song = Song()
+            song.idx = item["idx"]
+            song.id = item["id"]
+            song.title = item["title_localized"]["en"]
+            song.artist = item["artist"]
+            song.bpm = item["bpm"]
+            song.bpm_base = item["bpm_base"]
+            song.set = item["set"]
+            song.audio_preview = item["audioPreview"]
+            song.audio_preview_end = item["audioPreviewEnd"]
+            song.side = item["side"]
+            song.version = item["version"]
+            song.date = item["date"]
+            song.bg = to_db_value(item.get("bg"))
+            song.bg_inverse = to_db_value(item.get("bg_inverse"))
+            if item.get("bg_daynight"):
+                song.bg_day = to_db_value(item["bg_daynight"].get("day"))
+                song.bg_night = to_db_value(item["bg_daynight"].get("night"))
+            if item.get("source_localized"):
+                song.source = item["source_localized"]["en"]
+            song.source_copyright = to_db_value(item.get("source_copyright"))
+            results.append(song)
+
+            if (
+                is_localized(item, "title")
+                or is_localized(item, "search_title", append_localized=False)
+                or is_localized(item, "search_artist", append_localized=False)
+                or is_localized(item, "source")
+            ):
+                song_localized = SongLocalized(id=song.id)
+                set_model_localized_attrs(song_localized, item, "title")
+                set_model_localized_attrs(
+                    song_localized, item, "search_title", "search_title"
+                )
+                set_model_localized_attrs(
+                    song_localized, item, "search_artist", "search_artist"
+                )
+                set_model_localized_attrs(song_localized, item, "source")
+                results.append(song_localized)
+
+        return results
+
+
+class SonglistDifficultiesParser(ArcaeaParser):
+    def parse(self) -> List[Union[Difficulty, DifficultyLocalized]]:
+        songlist_json_root = json.loads(self.read_file_text())
+
+        songlist_json = songlist_json_root["songs"]
+        results = []
+        for song_item in songlist_json:
+            if not song_item.get("difficulties"):
+                continue
+
+            for item in song_item["difficulties"]:
+                if item["rating"] == 0:
+                    continue
+
+                chart = Difficulty(song_id=song_item["id"])
+                chart.rating_class = item["ratingClass"]
+                chart.rating = item["rating"]
+                chart.rating_plus = item.get("ratingPlus") or False
+                chart.chart_designer = item["chartDesigner"]
+                chart.jacket_desginer = item.get("jacketDesigner") or None
+                chart.audio_override = item.get("audioOverride") or False
+                chart.jacket_override = item.get("jacketOverride") or False
+                chart.jacket_night = item.get("jacketNight") or None
+                chart.title = item.get("title_localized", {}).get("en") or None
+                chart.artist = item.get("artist") or None
+                chart.bg = item.get("bg") or None
+                chart.bg_inverse = item.get("bg_inverse")
+                chart.bpm = item.get("bpm") or None
+                chart.bpm_base = item.get("bpm_base") or None
+                chart.version = item.get("version") or None
+                chart.date = item.get("date") or None
+                results.append(chart)
+
+                if is_localized(item, "title") or is_localized(item, "artist"):
+                    chart_localized = DifficultyLocalized(
+                        song_id=chart.song_id, rating_class=chart.rating_class
+                    )
+                    set_model_localized_attrs(chart_localized, item, "title")
+                    set_model_localized_attrs(chart_localized, item, "artist")
+                    results.append(chart_localized)
+
+        return results

src/arcaea_offline/external/arcaea/st3.py

@@ -0,0 +1,73 @@
+import logging
+import sqlite3
+from typing import List
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ...models.scores import Score
+from .common import ArcaeaParser, fix_timestamp
+
+logger = logging.getLogger(__name__)
+
+
+class St3ScoreParser(ArcaeaParser):
+    def parse(self) -> List[Score]:
+        items = []
+        with sqlite3.connect(self.filepath) as st3_conn:
+            cursor = st3_conn.cursor()
+            db_scores = cursor.execute(
+                "SELECT songId, songDifficulty, score, perfectCount, nearCount, missCount, "
+                "date, modifier FROM scores"
+            ).fetchall()
+            for (
+                song_id,
+                rating_class,
+                score,
+                pure,
+                far,
+                lost,
+                date,
+                modifier,
+            ) in db_scores:
+                clear_type = cursor.execute(
+                    "SELECT clearType FROM cleartypes WHERE songId = ? AND songDifficulty = ?",
+                    (song_id, rating_class),
+                ).fetchone()[0]
+
+                items.append(
+                    Score(
+                        song_id=song_id,
+                        rating_class=rating_class,
+                        score=score,
+                        pure=pure,
+                        far=far,
+                        lost=lost,
+                        date=fix_timestamp(date),
+                        modifier=modifier,
+                        clear_type=clear_type,
+                        comment="Parsed from st3",
+                    )
+                )
+
+        return items
+
+    def write_database(self, session: Session, *, skip_duplicate=True):
+        parsed_scores = self.parse()
+        for parsed_score in parsed_scores:
+            query_score = session.scalar(
+                select(Score).where(
+                    (Score.song_id == parsed_score.song_id)
+                    & (Score.rating_class == parsed_score.rating_class)
+                    & (Score.score == parsed_score.score)
+                )
+            )
+
+            if query_score and skip_duplicate:
+                logger.info(
+                    "%r skipped because potential duplicate item %r found.",
+                    parsed_score,
+                    query_score,
+                )
+                continue
+            session.add(parsed_score)

src/arcaea_offline/external/chart_info_db/__init__.py

@@ -0,0 +1,3 @@
+from .parser import ChartInfoDbParser
+
+__all__ = ["ChartInfoDbParser"]

src/arcaea_offline/external/chart_info_db/parser.py

@@ -0,0 +1,35 @@
+import contextlib
+import sqlite3
+from typing import List
+
+from sqlalchemy.orm import Session
+
+from ...models.songs import ChartInfo
+
+
+class ChartInfoDbParser:
+    def __init__(self, filepath):
+        self.filepath = filepath
+
+    def parse(self) -> List[ChartInfo]:
+        results = []
+        with sqlite3.connect(self.filepath) as conn:
+            with contextlib.closing(conn.cursor()) as cursor:
+                db_results = cursor.execute(
+                    "SELECT song_id, rating_class, constant, notes FROM charts_info"
+                ).fetchall()
+                for result in db_results:
+                    chart = ChartInfo(
+                        song_id=result[0],
+                        rating_class=result[1],
+                        constant=result[2],
+                        notes=result[3] or None,
+                    )
+                    results.append(chart)
+
+        return results
+
+    def write_database(self, session: Session):
+        results = self.parse()
+        for result in results:
+            session.merge(result)

src/arcaea_offline/external/importers/arcaea/__init__.py

@@ -1,10 +0,0 @@
-from .lists import ArcaeaPacklistParser, ArcaeaSonglistParser
-from .online import ArcaeaOnlineApiParser
-from .st3 import ArcaeaSt3Parser
-
-__all__ = [
-    "ArcaeaPacklistParser",
-    "ArcaeaSonglistParser",
-    "ArcaeaOnlineApiParser",
-    "ArcaeaSt3Parser",
-]

src/arcaea_offline/external/importers/arcaea/lists.py

@@ -26,7 +26,7 @@ class ArcaeaListParser:
         self.list_text = list_text
 
 
-class ArcaeaPacklistParser(ArcaeaListParser):
+class PacklistParser(ArcaeaListParser):
     def parse(self) -> List[Union[Pack, PackLocalized]]:
         root = json.loads(self.list_text)
 
@@ -57,7 +57,7 @@ class ArcaeaPacklistParser(ArcaeaListParser):
         return results
 
 
-class ArcaeaSonglistParser(ArcaeaListParser):
+class SonglistParser(ArcaeaListParser):
     def parse_songs(self) -> List[Union[Song, SongLocalized, SongSearchWord]]:
         root = json.loads(self.list_text)
 

src/arcaea_offline/external/importers/arcaea/online.py

@@ -1,97 +0,0 @@
-import json
-import logging
-from datetime import datetime, timezone
-from typing import Dict, List, Literal, Optional, TypedDict
-
-from arcaea_offline.constants.enums import (
-    ArcaeaPlayResultClearType,
-    ArcaeaPlayResultModifier,
-    ArcaeaRatingClass,
-)
-from arcaea_offline.database.models.v5 import PlayResult
-
-from .common import fix_timestamp
-
-logger = logging.getLogger(__name__)
-
-
-class _RatingMePlayResultItem(TypedDict):
-    song_id: str
-    difficulty: int
-    modifier: int
-    rating: float
-    score: int
-    perfect_count: int
-    near_count: int
-    miss_count: int
-    clear_type: int
-    title: Dict[Literal["ja", "en"], str]
-    artist: str
-    time_played: int
-    bg: str
-
-
-class _RatingMeValue(TypedDict):
-    best_rated_scores: List[_RatingMePlayResultItem]
-    recent_rated_scores: List[_RatingMePlayResultItem]
-
-
-class _RatingMeResponse(TypedDict):
-    success: bool
-    error_code: Optional[int]
-    value: Optional[_RatingMeValue]
-
-
-class ArcaeaOnlineApiParser:
-    def __init__(self, api_result_text: str):
-        self.api_result_text = api_result_text
-        self.api_result: _RatingMeResponse = json.loads(api_result_text)
-
-    def parse(self) -> List[PlayResult]:
-        api_result_value = self.api_result.get("value")
-        if not api_result_value:
-            error_code = self.api_result.get("error_code")
-            raise ValueError(
-                f"Cannot parse Arcaea Online API result, error code {error_code}"
-            )
-
-        best30_items = api_result_value.get("best_rated_scores", [])
-        recent_items = api_result_value.get("recent_rated_scores", [])
-        items = best30_items + recent_items
-
-        date_text = (
-            datetime.now(tz=timezone.utc).astimezone().isoformat(timespec="seconds")
-        )
-
-        results: List[PlayResult] = []
-        results_time_played = []
-        for item in items:
-            date_millis = fix_timestamp(item["time_played"])
-
-            if date_millis in results_time_played:
-                # filter out duplicate play results
-                continue
-
-            if date_millis:
-                date = datetime.fromtimestamp(date_millis / 1000).astimezone()
-                results_time_played.append(date_millis)
-            else:
-                date = None
-
-            play_result = PlayResult()
-            play_result.song_id = item["song_id"]
-            play_result.rating_class = ArcaeaRatingClass(item["difficulty"])
-            play_result.score = item["score"]
-            play_result.pure = item["perfect_count"]
-            play_result.far = item["near_count"]
-            play_result.lost = item["miss_count"]
-            play_result.date = date
-            play_result.modifier = ArcaeaPlayResultModifier(item["modifier"])
-            play_result.clear_type = ArcaeaPlayResultClearType(item["clear_type"])
-
-            if play_result.lost == 0:
-                play_result.max_recall = play_result.pure + play_result.far
-
-            play_result.comment = f"Parsed from web API at {date_text}"
-            results.append(play_result)
-        return results

src/arcaea_offline/external/importers/arcaea/st3.py

@@ -19,7 +19,7 @@ from .common import fix_timestamp
 logger = logging.getLogger(__name__)
 
 
-class ArcaeaSt3Parser:
+class St3Parser:
     @classmethod
     @overload
     def parse(cls, db: sqlite3.Connection) -> List[PlayResult]: ...

src/arcaea_offline/external/importers/chart_info_database.py

@@ -1,42 +0,0 @@
-import sqlite3
-from contextlib import closing
-from typing import List, overload
-
-from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
-from arcaea_offline.database.models.v5 import ChartInfo
-
-
-class ChartInfoDatabaseParser:
-    @classmethod
-    @overload
-    def parse(cls, conn: sqlite3.Connection) -> List[ChartInfo]: ...
-
-    @classmethod
-    @overload
-    def parse(cls, conn: sqlite3.Cursor) -> List[ChartInfo]: ...
-
-    @classmethod
-    def parse(cls, conn) -> List[ChartInfo]:
-        if isinstance(conn, sqlite3.Connection):
-            with closing(conn.cursor()) as cur:
-                return cls.parse(cur)
-
-        if not isinstance(conn, sqlite3.Cursor):
-            raise ValueError("conn must be sqlite3.Connection or sqlite3.Cursor!")
-
-        db_items = conn.execute(
-            "SELECT song_id, rating_class, constant, notes FROM charts_info"
-        ).fetchall()
-
-        results: List[ChartInfo] = []
-        for item in db_items:
-            (song_id, rating_class, constant, notes) = item
-
-            chart_info = ChartInfo()
-            chart_info.song_id = song_id
-            chart_info.rating_class = ArcaeaRatingClass(rating_class)
-            chart_info.constant = constant
-            chart_info.notes = notes
-
-            results.append(chart_info)
-        return results

tests/external/importers/arcaea/test_online.py

@@ -1,88 +0,0 @@
-import json
-from datetime import datetime, timezone
-
-from arcaea_offline.constants.enums.arcaea import (
-    ArcaeaPlayResultClearType,
-    ArcaeaPlayResultModifier,
-    ArcaeaRatingClass,
-)
-from arcaea_offline.database.models.v5.play_results import PlayResult
-from arcaea_offline.external.importers.arcaea.online import ArcaeaOnlineApiParser
-
-API_RESULT = {
-    "success": True,
-    "value": {
-        "best_rated_scores": [
-            {
-                "song_id": "test1",
-                "difficulty": 2,
-                "modifier": 0,
-                "rating": 12.5,
-                "score": 9908123,
-                "perfect_count": 1234,
-                "near_count": 12,
-                "miss_count": 4,
-                "clear_type": 1,
-                "title": {"ja": "テスト1", "en": "Test 1"},
-                "artist": "pytest",
-                "time_played": 1704067200000,  # 2024-01-01 00:00:00 UTC
-                "bg": "abcdefg123456hijklmn7890123opqrs",
-            },
-            {
-                "song_id": "test2",
-                "difficulty": 2,
-                "modifier": 0,
-                "rating": 12.0,
-                "score": 9998123,
-                "perfect_count": 1234,
-                "near_count": 1,
-                "miss_count": 0,
-                "clear_type": 1,
-                "title": {"ja": "テスト2", "en": "Test 2"},
-                "artist": "pytest",
-                "time_played": 1704067200000,
-                "bg": "abcdefg123456hijklmn7890123opqrs",
-            },
-        ],
-        "recent_rated_scores": [
-            {
-                "song_id": "test2",
-                "difficulty": 2,
-                "modifier": 0,
-                "rating": 12.0,
-                "score": 9998123,
-                "perfect_count": 1234,
-                "near_count": 1,
-                "miss_count": 0,
-                "clear_type": 1,
-                "title": {"ja": "テスト2", "en": "Test 2"},
-                "artist": "pytest",
-                "time_played": 1704153600000,  # 2024-01-02 00:00:00 UTC
-                "bg": "abcdefg123456hijklmn7890123opqrs",
-            }
-        ],
-    },
-}
-
-
-class TestArcaeaOnlineApiParser:
-    API_RESULT_CONTENT = json.dumps(API_RESULT, ensure_ascii=False)
-
-    def test_parse(self):
-        play_results = ArcaeaOnlineApiParser(self.API_RESULT_CONTENT).parse()
-        assert all(isinstance(item, PlayResult) for item in play_results)
-
-        assert len(play_results) == 2
-
-        test1 = next(filter(lambda x: x.song_id == "test1", play_results))
-        assert test1.rating_class is ArcaeaRatingClass.FUTURE
-        assert test1.score == 9908123
-        assert test1.pure == 1234
-        assert test1.far == 12
-        assert test1.lost == 4
-        assert test1.date == datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
-        assert test1.clear_type is ArcaeaPlayResultClearType.NORMAL_CLEAR
-        assert test1.modifier is ArcaeaPlayResultModifier.NORMAL
-
-        test2 = next(filter(lambda x: x.song_id == "test2", play_results))
-        assert test2.date == datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc)

tests/external/importers/arcaea/test_st3.py

@@ -2,23 +2,23 @@ import sqlite3
 from datetime import datetime
 
 import pytest
-
-import tests.resources
 from arcaea_offline.constants.enums.arcaea import (
     ArcaeaPlayResultClearType,
     ArcaeaPlayResultModifier,
     ArcaeaRatingClass,
 )
-from arcaea_offline.external.importers.arcaea.st3 import ArcaeaSt3Parser
+from arcaea_offline.external.importers.arcaea.st3 import St3Parser
 
-db = sqlite3.connect(":memory:")
+import tests.resources
-db.executescript(tests.resources.get_resource("st3.sql").read_text(encoding="utf-8"))
 
 
-class TestArcaeaSt3Parser:
+class TestSt3Parser:
+    DB_PATH = tests.resources.get_resource("st3-test.db")
+
     @property
     def play_results(self):
-        return ArcaeaSt3Parser.parse(db)
+        conn = sqlite3.connect(str(self.DB_PATH))
+        return St3Parser.parse(conn)
 
     def test_basic(self):
         play_results = self.play_results
@@ -51,5 +51,5 @@ class TestArcaeaSt3Parser:
         assert date1.date is None
 
     def test_invalid_input(self):
-        pytest.raises(TypeError, ArcaeaSt3Parser.parse, "abcdefghijklmn")
+        pytest.raises(TypeError, St3Parser.parse, "abcdefghijklmn")
-        pytest.raises(TypeError, ArcaeaSt3Parser.parse, 123456)
+        pytest.raises(TypeError, St3Parser.parse, 123456)

tests/external/importers/test_chart_info_database.py

@@ -1,34 +0,0 @@
-import sqlite3
-
-import tests.resources
-from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
-from arcaea_offline.database.models.v5 import ChartInfo
-from arcaea_offline.external.importers.chart_info_database import (
-    ChartInfoDatabaseParser,
-)
-
-db = sqlite3.connect(":memory:")
-db.executescript(tests.resources.get_resource("cidb.sql").read_text(encoding="utf-8"))
-
-
-class TestChartInfoDatabaseParser:
-    def test_parse(self):
-        items = ChartInfoDatabaseParser.parse(db)
-        assert all(isinstance(item, ChartInfo) for item in items)
-
-        assert len(items) == 3
-
-        test1 = next(filter(lambda x: x.song_id == "test1", items))
-        assert test1.rating_class is ArcaeaRatingClass.PRESENT
-        assert test1.constant == 90
-        assert test1.notes == 900
-
-        test2 = next(filter(lambda x: x.song_id == "test2", items))
-        assert test2.rating_class is ArcaeaRatingClass.FUTURE
-        assert test2.constant == 95
-        assert test2.notes == 950
-
-        test3 = next(filter(lambda x: x.song_id == "test3", items))
-        assert test3.rating_class is ArcaeaRatingClass.BEYOND
-        assert test3.constant == 100
-        assert test3.notes is None

tests/resources/cidb.sql

@@ -1,11 +0,0 @@
-CREATE TABLE charts_info (
-    song_id TEXT NOT NULL,
-    rating_class INTEGER NOT NULL,
-    constant INTEGER NOT NULL,
-    notes INTEGER
-);
-
-INSERT INTO charts_info (song_id, rating_class, constant, notes) VALUES
-    ("test1", 1, 90, 900),
-    ("test2", 2, 95, 950),
-    ("test3", 3, 100, NULL);

tests/resources/st3-test.json

@@ -0,0 +1,117 @@
+{
+  "_comment": "A quick preview of the data in st3-test.db",
+  "scores": [
+    {
+      "id": 1,
+      "version": 1,
+      "score": 9441167,
+      "shinyPerfectCount": 753,
+      "perfectCount": 895,
+      "nearCount": 32,
+      "missCount": 22,
+      "date": 1722100000,
+      "songId": "test1",
+      "songDifficulty": 2,
+      "modifier": 2,
+      "health": 0,
+      "ct": 0
+    },
+    {
+      "id": 2,
+      "version": 1,
+      "score": 9752087,
+      "shinyPerfectCount": 914,
+      "perfectCount": 1024,
+      "nearCount": 29,
+      "missCount": 12,
+      "date": 1722200000,
+      "songId": "test2",
+      "songDifficulty": 2,
+      "modifier": 0,
+      "health": 100,
+      "ct": 0
+    },
+    {
+      "id": 3,
+      "version": 1,
+      "score": 9750000,
+      "shinyPerfectCount": 900,
+      "perfectCount": 1000,
+      "nearCount": 20,
+      "missCount": 10,
+      "date": 1722200000,
+      "songId": "corrupt1",
+      "songDifficulty": 5,
+      "modifier": 0,
+      "health": 0,
+      "ct": 0
+    },
+    {
+      "id": 4,
+      "version": 1,
+      "score": 9750000,
+      "shinyPerfectCount": 900,
+      "perfectCount": 1000,
+      "nearCount": 20,
+      "missCount": 10,
+      "date": 1722200000,
+      "songId": "corrupt2",
+      "songDifficulty": 2,
+      "modifier": 9,
+      "health": 0,
+      "ct": 0
+    },
+    {
+      "id": 5,
+      "version": 1,
+      "score": 9750000,
+      "shinyPerfectCount": 900,
+      "perfectCount": 1000,
+      "nearCount": 20,
+      "missCount": 10,
+      "date": 1,
+      "songId": "date1",
+      "songDifficulty": 2,
+      "modifier": 0,
+      "health": 0,
+      "ct": 0
+    }
+  ],
+  "cleartypes": [
+    {
+      "id": 1,
+      "songId": "test1",
+      "songDifficulty": 2,
+      "clearType": 0,
+      "ct": 0
+    },
+    {
+      "id": 2,
+      "songId": "test2",
+      "songDifficulty": 2,
+      "clearType": 1,
+      "ct": 0
+    },
+    {
+      "id": 3,
+      "songId": "corrupt1",
+      "songDifficulty": 5,
+      "clearType": 0,
+      "ct": 0
+    },
+    {
+      "id": 4,
+      "songId": "corrupt2",
+      "songDifficulty": 2,
+      "clearType": 7,
+      "ct": 0
+    },
+    {
+      "id": 5,
+      "songId": "date1",
+      "songDifficulty": 2,
+      "clearType": 1,
+      "ct": 0
+    }
+  ]
+}

tests/resources/st3.sql

@@ -1,37 +0,0 @@
-CREATE TABLE scores (
-    id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
-    version INTEGER,
-    score INTEGER,
-    shinyPerfectCount INTEGER,
-    perfectCount INTEGER,
-    nearCount INTEGER,
-    missCount INTEGER,
-    date INTEGER,
-    songId TEXT,
-    songDifficulty INTEGER,
-    modifier INTEGER,
-    health INTEGER,
-    ct INTEGER DEFAULT 0
-);
-
-CREATE TABLE cleartypes (
-    id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
-    songId TEXT,
-    songDifficulty INTEGER,
-    clearType INTEGER,
-    ct INTEGER DEFAULT 0
-);
-
-INSERT INTO scores ("id", "version", "score", "shinyPerfectCount", "perfectCount", "nearCount", "missCount", "date", "songId", "songDifficulty", "modifier", "health", "ct") VALUES
-    ('1', '1', '9441167', '753', '895', '32', '22', '1722100000', 'test1', '2', '2', '0', '0'),
-    ('2', '1', '9752087', '914', '1024', '29', '12', '1722200000', 'test2', '2', '0', '100', '0'),
-    ('3', '1', '9750000', '900', '1000', '20', '10', '1722200000', 'corrupt1', '5', '0', '0', '0'),
-    ('4', '1', '9750000', '900', '1000', '20', '10', '1722200000', 'corrupt2', '2', '9', '0', '0'),
-    ('5', '1', '9750000', '900', '1000', '20', '10', '1', 'date1', '2', '0', '0', '0');
-
-INSERT INTO cleartypes ("id", "songId", "songDifficulty", "clearType", "ct") VALUES
-    ('1', 'test1', '2', '0', '0'),
-    ('2', 'test2', '2', '1', '0'),
-    ('3', 'corrupt1', '5', '0', '0'),
-    ('4', 'corrupt2', '2', '7', '0'),
-    ('5', 'date1', '2', '1', '0');