impr(external): St3ScoreParser improvements

src/arcaea_offline/external/arcaea/__init__.py

@@ -1,3 +1,4 @@
+from .online import ArcaeaOnlineParser
 from .packlist import PacklistParser
 from .songlist import SonglistDifficultiesParser, SonglistParser
 from .st3 import St3ScoreParser

src/arcaea_offline/external/arcaea/common.py

@@ -1,11 +1,41 @@
 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(timestamp)) + 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

src/arcaea_offline/external/arcaea/online.py

@@ -0,0 +1,75 @@
+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 __init__(self, filepath):
+        super().__init__(filepath)
+
+    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/st3.py

@@ -6,7 +6,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from ...models.scores import Score
-from .common import ArcaeaParser
+from .common import ArcaeaParser, fix_timestamp
 
 logger = logging.getLogger(__name__)
 
@@ -37,9 +37,6 @@ class St3ScoreParser(ArcaeaParser):
                     (song_id, rating_class),
                 ).fetchone()[0]
 
-                date_str = str(date)
-                date = None if len(date_str) < 7 else int(date_str.ljust(10, "0"))
-
                 items.append(
                     Score(
                         song_id=song_id,
@@ -48,10 +45,10 @@ class St3ScoreParser(ArcaeaParser):
                         pure=pure,
                         far=far,
                         lost=lost,
-                        date=date,
+                        date=fix_timestamp(date),
                         modifier=modifier,
                         clear_type=clear_type,
-                        comment="Imported from st3",
+                        comment=f"Parsed from st3",
                     )
                 )
 

src/arcaea_offline/models/scores.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from sqlalchemy import TEXT, case, func, inspect, select
+from sqlalchemy import TEXT, case, func, inspect, select, text
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 from sqlalchemy_utils import create_view
 
@@ -78,10 +78,16 @@ class ScoreCalculated(ScoresViewBase):
             Score.score,
             Score.pure,
             (
-                Score.score
-                - func.floor(
-                    (Score.pure * 10000000.0 / ChartInfo.notes)
-                    + (Score.far * 0.5 * 10000000.0 / ChartInfo.notes)
+                case(
+                    (
+                        (ChartInfo.notes.isnot(None) & ChartInfo.notes != 0),
+                        Score.score
+                        - func.floor(
+                            (Score.pure * 10000000.0 / ChartInfo.notes)
+                            + (Score.far * 0.5 * 10000000.0 / ChartInfo.notes)
+                        ),
+                    ),
+                    else_=text("NULL"),
                 )
             ).label("shiny_pure"),
             Score.far,

src/arcaea_offline/utils/partner.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from enum import IntEnum
+
+
+class KanaeDayNight(IntEnum):
+    Day = 0
+    Night = 1
+
+
+def kanae_day_night(timestamp: int) -> KanaeDayNight:
+    """
+    :param timestamp: POSIX timestamp, which is passed to `datetime.fromtimestamp(timestamp)`.
+    """
+    dt = datetime.fromtimestamp(timestamp)
+    return KanaeDayNight.Day if 6 <= dt.hour <= 19 else KanaeDayNight.Night

src/arcaea_offline/utils/score.py

@@ -2,6 +2,15 @@ from typing import Any, Sequence
 
 SCORE_GRADE_FLOOR = [9900000, 9800000, 9500000, 9200000, 8900000, 8600000, 0]
 SCORE_GRADE_TEXTS = ["EX+", "EX", "AA", "A", "B", "C", "D"]
+MODIFIER_TEXTS = ["NORMAL", "EASY", "HARD"]
+CLEAR_TYPE_TEXTS = [
+    "TRACK LOST",
+    "NORMAL CLEAR",
+    "FULL RECALL",
+    "PURE MEMORY",
+    "EASY CLEAR",
+    "HARD CLEAR",
+]
 
 
 def zip_score_grade(score: int, __seq: Sequence, default: Any = "__PRESERVE__"):
@@ -27,3 +36,11 @@ def zip_score_grade(score: int, __seq: Sequence, default: Any = "__PRESERVE__"):
 
 def score_to_grade_text(score: int) -> str:
     return zip_score_grade(score, SCORE_GRADE_TEXTS)
+
+
+def modifier_to_text(modifier: int) -> str:
+    return MODIFIER_TEXTS[modifier]
+
+
+def clear_type_to_text(clear_type: int) -> str:
+    return CLEAR_TYPE_TEXTS[clear_type]