refactor: st3 parser (importer)

pyproject.toml

@@ -9,10 +9,7 @@ authors = [{ name = "283375", email = "log_283375@163.com" }]
 description = "Manage your local Arcaea score database."
 readme = "README.md"
 requires-python = ">=3.8"
-dependencies = [
-  "SQLAlchemy==2.0.20",
-  "SQLAlchemy-Utils==0.41.1",
-]
+dependencies = ["SQLAlchemy==2.0.20", "SQLAlchemy-Utils==0.41.1"]
 classifiers = [
   "Development Status :: 3 - Alpha",
   "Programming Language :: Python :: 3",
@@ -49,3 +46,8 @@ select = [
 ignore = [
   "E501", # line-too-long
 ]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = [
+  "PLR2004", # magic-value-comparison
+]

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

@@ -0,0 +1,118 @@
+"""
+Game database play results importer
+"""
+
+import logging
+import sqlite3
+from datetime import datetime, timezone
+from typing import List, overload
+
+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 St3Parser:
+    @classmethod
+    @overload
+    def parse(cls, db: sqlite3.Connection) -> List[PlayResult]: ...
+
+    @classmethod
+    @overload
+    def parse(cls, db: sqlite3.Cursor) -> List[PlayResult]: ...
+
+    @classmethod
+    def parse(cls, db) -> List[PlayResult]:
+        if isinstance(db, sqlite3.Connection):
+            return cls.parse(db.cursor())
+
+        if not isinstance(db, sqlite3.Cursor):
+            raise TypeError(
+                "Unknown overload of `db`. Expected `sqlite3.Connection` or `sqlite3.Cursor`."
+            )
+
+        entities = []
+        query_results = db.execute("""
+            SELECT s.id AS _id, s.songId, s.songDifficulty AS ratingClass, s.score,
+                s.perfectCount AS pure, s.nearCount AS far, s.missCount AS lost,
+                s.`date`, s.modifier, ct.clearType
+            FROM scores s JOIN cleartypes ct
+            ON s.songId = ct.songId AND s.songDifficulty = ct.songDifficulty""")
+        # maybe `s.id = ct.id`?
+
+        now = datetime.now(tz=timezone.utc)
+        import_comment = (
+            f"Imported from st3 at {now.astimezone().isoformat(timespec='seconds')}"
+        )
+        for result in query_results:
+            (
+                _id,
+                song_id,
+                rating_class,
+                score,
+                pure,
+                far,
+                lost,
+                date,
+                modifier,
+                clear_type,
+            ) = result
+
+            try:
+                rating_class_enum = ArcaeaRatingClass(rating_class)
+            except ValueError:
+                logger.warning(
+                    "Unknown rating class [%r] at entry id %d, skipping!",
+                    rating_class,
+                    _id,
+                )
+                continue
+
+            try:
+                clear_type_enum = ArcaeaPlayResultClearType(clear_type)
+            except ValueError:
+                logger.warning(
+                    "Unknown clear type [%r] at entry id %d, falling back to `None`!",
+                    clear_type,
+                    _id,
+                )
+                clear_type_enum = None
+
+            try:
+                modifier_enum = ArcaeaPlayResultModifier(modifier)
+            except ValueError:
+                logger.warning(
+                    "Unknown modifier [%r] at entry id %d, falling back to `None`!",
+                    modifier,
+                    _id,
+                )
+                modifier_enum = None
+
+            if date := fix_timestamp(date):
+                date = datetime.fromtimestamp(date).astimezone()
+            else:
+                date = None
+
+            entities.append(
+                PlayResult(
+                    song_id=song_id,
+                    rating_class=rating_class_enum,
+                    score=score,
+                    pure=pure,
+                    far=far,
+                    lost=lost,
+                    date=date,
+                    modifier=modifier_enum,
+                    clear_type=clear_type_enum,
+                    comment=import_comment,
+                )
+            )
+
+        return entities

tests/external/importers/arcaea/test_st3.py

@@ -0,0 +1,56 @@
+import sqlite3
+from datetime import datetime
+from importlib.resources import files
+
+import pytest
+from arcaea_offline.constants.enums.arcaea import (
+    ArcaeaPlayResultClearType,
+    ArcaeaPlayResultModifier,
+    ArcaeaRatingClass,
+)
+from arcaea_offline.external.importers.arcaea.st3 import St3Parser
+
+import tests.resources
+
+
+class TestSt3Parser:
+    DB_PATH = files(tests.resources).joinpath("st3-test.db")
+
+    @property
+    def play_results(self):
+        conn = sqlite3.connect(str(self.DB_PATH))
+        return St3Parser.parse(conn)
+
+    def test_basic(self):
+        play_results = self.play_results
+
+        assert len(play_results) == 4
+
+        test1 = next(filter(lambda x: x.song_id == "test1", play_results))
+        assert test1.rating_class is ArcaeaRatingClass.FUTURE
+        assert test1.score == 9441167
+        assert test1.pure == 895
+        assert test1.far == 32
+        assert test1.lost == 22
+        assert test1.date == datetime.fromtimestamp(1722100000).astimezone()
+        assert test1.clear_type is ArcaeaPlayResultClearType.TRACK_LOST
+        assert test1.modifier is ArcaeaPlayResultModifier.HARD
+
+    def test_corrupt_handling(self):
+        play_results = self.play_results
+
+        corrupt1 = filter(lambda x: x.song_id == "corrupt1", play_results)
+        # `rating_class` out of range, so this should be ignored during parsing,
+        # thus does not present in the result.
+        assert len(list(corrupt1)) == 0
+
+        corrupt2 = next(filter(lambda x: x.song_id == "corrupt2", play_results))
+        assert corrupt2.clear_type is None
+        assert corrupt2.modifier is None
+
+        date1 = next(filter(lambda x: x.song_id == "date1", play_results))
+        assert date1.date is None
+
+    def test_invalid_input(self):
+        pytest.raises(TypeError, St3Parser.parse, "abcdefghijklmn")
+        pytest.raises(TypeError, St3Parser.parse, 123456)

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
+    }
+  ]
+}