diff --git a/src/arcaea_offline/external/arcaea/__init__.py b/src/arcaea_offline/external/arcaea/__init__.py index c18f1fb..8fe4ea9 100644 --- a/src/arcaea_offline/external/arcaea/__init__.py +++ b/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 diff --git a/src/arcaea_offline/external/arcaea/common.py b/src/arcaea_offline/external/arcaea/common.py index f1c8e4c..200c652 100644 --- a/src/arcaea_offline/external/arcaea/common.py +++ b/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() / 1000))) + 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 diff --git a/src/arcaea_offline/external/arcaea/online.py b/src/arcaea_offline/external/arcaea/online.py new file mode 100644 index 0000000..835b5ff --- /dev/null +++ b/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