From bfa1472b5c3d4a29c8431038a8ad43b97cc9c2e5 Mon Sep 17 00:00:00 2001 From: 283375 Date: Fri, 27 Sep 2024 23:10:48 +0800 Subject: [PATCH] refactor: arcaea online play results importer --- .../external/importers/arcaea/online.py | 97 +++++++++++++++++++ .../external/importers/arcaea/test_online.py | 88 +++++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 src/arcaea_offline/external/importers/arcaea/online.py create mode 100644 tests/external/importers/arcaea/test_online.py diff --git a/src/arcaea_offline/external/importers/arcaea/online.py b/src/arcaea_offline/external/importers/arcaea/online.py new file mode 100644 index 0000000..dbce5ac --- /dev/null +++ b/src/arcaea_offline/external/importers/arcaea/online.py @@ -0,0 +1,97 @@ +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 diff --git a/tests/external/importers/arcaea/test_online.py b/tests/external/importers/arcaea/test_online.py new file mode 100644 index 0000000..6139fa4 --- /dev/null +++ b/tests/external/importers/arcaea/test_online.py @@ -0,0 +1,88 @@ +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)