This commit is contained in:
2025-07-06 18:43:00 +08:00
parent 155e8b9664
commit bcaa1a268e
9 changed files with 390 additions and 401 deletions

View File

@ -16,19 +16,27 @@ You should have received a copy of the GNU General Public License along with
this program. If not, see <https://www.gnu.org/licenses/>.
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any
from arcaea_offline_schemas.dis.v5.dataclass import (
AodisDifficulty,
AodisDifficultyLocalization,
AodisSong,
AodisSongLocalization,
)
from arcaea_offline_schemas.dis.v5.dataclass.shared import (
AddedAtObject,
Backgrounds,
BpmObject,
)
from arcaea_offline_schemas.dis.v5.dataclass.song import Source
from typeguard import TypeCheckError, check_type
from ._shared import blank_str_to_none
from .defs import (
ARCAEA_LANGUAGE_KEYS,
Difficulty,
DifficultyLocalization,
Song,
SongLocalization,
)
from .defs import ARCAEA_LANGUAGE_KEYS
logger = logging.getLogger(__name__)
@ -39,170 +47,215 @@ class SonglistFormatError(Exception):
class ArcaeaSonglistParser:
@staticmethod
def _assert_root_content(root: Any):
if not isinstance(root, dict):
msg = f"songlist root should be `dict`, got {type(root)}"
raise SonglistFormatError(msg)
songs = root.get("songs")
if songs is None:
msg = "`songs` field does not exist"
raise SonglistFormatError(msg)
if not isinstance(songs, list):
msg = f"`songs` field should be `list`, got {type(songs)}"
raise SonglistFormatError(msg)
def check_root_content_format(root: dict) -> None:
try:
check_type(root, dict)
check_type(root["songs"], list)
except (KeyError, TypeCheckError) as e:
raise SonglistFormatError from e
@staticmethod
def _assert_song_object(obj: Any):
if not isinstance(obj, dict):
msg = f"A song object should be `dict`, got {type(obj)}"
raise SonglistFormatError(msg)
def check_song_object_format(song: dict) -> None:
try:
check_type(song, dict)
check_type(song["id"], str)
id_ = obj.get("id")
if not isinstance(id_, str):
msg = f"`id` attribute of a song should be `str`, got value {id_!r}"
raise SonglistFormatError(msg)
if deleted := song.get("deleted"):
check_type(deleted, bool)
is_deleted = obj.get("deleted", False)
if not isinstance(is_deleted, bool):
msg = f"`deleted` attribute of a song should be `bool`, got value {id_!r}"
raise SonglistFormatError(msg)
bg_daynight = song.get("bg_daynight", {})
bg = blank_str_to_none(song.get("bg"))
bg_inverse = blank_str_to_none(song.get("bg_inverse"))
bg_day = blank_str_to_none(bg_daynight.get("day"))
bg_night = blank_str_to_none(bg_daynight.get("night"))
if (bg_inverse or bg_day or bg_night) and not bg:
msg = "One of bg_inverse, bg_day and bg_night is set without bg"
raise SonglistFormatError(msg)
# TODO: check localized objects and difficulties (strict list[dict]) and booleans
except (KeyError, TypeCheckError) as e:
raise SonglistFormatError from e
# TODO(283375): localized objects & difficulties (strict list[dict]) & booleans # noqa: E501, FIX002, TD003
@staticmethod
def parse_song(song: dict) -> list[AodisSong | AodisSongLocalization]:
id_ = song["id"]
title_localized = song.get("title_localized", {})
bg_daynight = song.get("bg_daynight", {})
source_localized = song.get("source_localized", {})
jacket_localized = song.get("jacket_localized", {})
bg = blank_str_to_none(song.get("bg"))
bg_inverse = blank_str_to_none(song.get("bg_inverse"))
bg_day = blank_str_to_none(bg_daynight.get("day"))
bg_night = blank_str_to_none(bg_daynight.get("night"))
results = []
aodis_song = AodisSong(
pack_id=song["set"],
id=id_,
addedAt=AddedAtObject(
date=datetime.fromtimestamp(song["date"], tz=timezone.utc).isoformat(),
version=blank_str_to_none(song.get("version")),
),
idx=song.get("idx"),
title=blank_str_to_none(title_localized.get("en")),
artist=blank_str_to_none(song.get("artist")),
isDeleted=song.get("deleted", False),
bpm=BpmObject(
value=song["bpm_base"],
display=blank_str_to_none(song.get("bpm")),
)
if song.get("bpm_base") is not None
else None,
isRemote=song.get("remote_dl", False),
isUnlockableInWorld=song.get("world_unlock", False),
isBeyondUnlockStateLocal=song.get("byd_local_unlock", False),
purchase=blank_str_to_none(song.get("purchase")),
category=blank_str_to_none(song.get("category")),
side=song.get("side"),
background=Backgrounds(
base=bg or bg_inverse or bg_day or bg_night,
inverse=blank_str_to_none(song.get("bg_inverse")),
day=blank_str_to_none(bg_daynight.get("day")),
night=blank_str_to_none(bg_daynight.get("night")),
)
if bg is not None
else None,
source=Source(
display=source_localized["en"],
copyright=blank_str_to_none(song.get("source_copyright")),
)
if source_localized.get("en") is not None
else None,
)
results.append(aodis_song)
for lang_key in ARCAEA_LANGUAGE_KEYS:
title_l10n = blank_str_to_none(title_localized.get(lang_key))
source_l10n = blank_str_to_none(source_localized.get(lang_key))
has_localized_jacket = jacket_localized.get(lang_key)
if title_l10n or source_l10n or has_localized_jacket:
results.append(
AodisSongLocalization(
id=id_,
lang=lang_key,
title=title_l10n,
source=source_l10n,
hasJacket=has_localized_jacket,
),
)
return results
@staticmethod
def parse_difficulties(
song: dict,
) -> list[AodisDifficulty | AodisDifficultyLocalization]:
id_ = song["id"]
difficulties = song["difficulties"]
results = []
for difficulty in difficulties:
difficulty_title_localized = difficulty.get("title_localized", {})
difficulty_artist_localized = difficulty.get("artist_localized", {})
rating_class = difficulty["ratingClass"]
results.append(
AodisDifficulty(
songId=id_,
ratingClass=rating_class,
rating=difficulty["rating"],
isRatingPlus=difficulty.get("ratingPlus", False),
chartDesigner=blank_str_to_none(
difficulty.get("chartDesigner"),
),
jacketDesigner=blank_str_to_none(
difficulty.get("jacketDesigner"),
),
hasOverridingAudio=difficulty.get("audioOverride", False),
hasOverridingJacket=difficulty.get("jacketOverride", False),
jacketNight=blank_str_to_none(difficulty.get("jacket_night")),
addedAt=AddedAtObject(
date=datetime.fromtimestamp(
difficulty["date"],
tz=timezone.utc,
).isoformat(),
version=blank_str_to_none(difficulty.get("version")),
)
if difficulty.get("date")
else None,
background=Backgrounds(base=difficulty["bg"])
if difficulty.get("bg") is not None
else None,
title=blank_str_to_none(difficulty_title_localized.get("en")),
artist=blank_str_to_none(difficulty_artist_localized.get("en")),
isLegacy11=difficulty.get("legacy11", False),
),
)
for lang_key in ARCAEA_LANGUAGE_KEYS:
difficuly_title_l10n = blank_str_to_none(
difficulty_title_localized.get(lang_key),
)
difficuly_artist_l10n = blank_str_to_none(
difficulty_artist_localized.get(lang_key),
)
if difficuly_title_l10n:
results.append(
AodisDifficultyLocalization(
songId=id_,
ratingClass=rating_class,
lang=lang_key,
title=difficuly_title_l10n,
artist=difficuly_artist_l10n,
),
)
return results
@classmethod
def items(
cls, songlist_content: dict, *, skip_asserts: bool = False
) -> list[Song | SongLocalization | Difficulty | DifficultyLocalization]:
cls,
songlist_content: dict,
*,
skip_format_checks: bool = False,
) -> list[
AodisSong
| AodisSongLocalization
| AodisDifficulty
| AodisDifficultyLocalization
]:
try:
cls._assert_root_content(songlist_content)
except SonglistFormatError as e:
if skip_asserts:
cls.check_root_content_format(songlist_content)
except SonglistFormatError:
if skip_format_checks:
logger.exception("Error skipped during songlist parsing:")
return []
raise e
raise
songs = songlist_content["songs"]
results = []
for song in songs:
try:
cls._assert_song_object(song)
except SonglistFormatError as e:
if skip_asserts:
cls.check_song_object_format(song)
except SonglistFormatError:
if skip_format_checks:
logger.exception("Error skipped during songlist parsing:")
continue
raise e
raise
if song.get("deleted"):
continue # TODO: none, return...
continue
title_localized = song.get("title_localized", {})
bg_daynight = song.get("bg_daynight", {})
source_localized = song.get("source_localized", {})
jacket_localized = song.get("jacket_localized", {})
id_ = song["id"]
results.append(
Song(
pack_id=song["set"],
id=id_,
added_at=datetime.fromtimestamp(song["date"], tz=timezone.utc),
idx=song.get("idx", None),
title=blank_str_to_none(title_localized.get("en")),
artist=blank_str_to_none(song.get("artist")),
is_deleted=song.get("deleted", False),
version=blank_str_to_none(song.get("version")),
bpm=blank_str_to_none(song.get("bpm")),
bpm_base=Decimal(song["bpm_base"])
if song.get("bpm_base")
else None,
is_remote=song.get("remote_dl", False),
is_unlockable_in_world=song.get("world_unlock", False),
is_beyond_unlock_state_local=song.get("byd_local_unlock", False),
purchase=blank_str_to_none(song.get("purchase")),
category=blank_str_to_none(song.get("category")),
side=song.get("side"),
bg=blank_str_to_none(song.get("bg")),
bg_inverse=blank_str_to_none(song.get("bg_inverse")),
bg_day=blank_str_to_none(bg_daynight.get("day")),
bg_night=blank_str_to_none(bg_daynight.get("night")),
source=blank_str_to_none(source_localized.get("en")),
source_copyright=blank_str_to_none(song.get("source_copyright")),
)
)
for lang_key in ARCAEA_LANGUAGE_KEYS:
title_l10n = blank_str_to_none(title_localized.get(lang_key))
source_l10n = blank_str_to_none(source_localized.get(lang_key))
has_localized_jacket = jacket_localized.get(lang_key)
if title_l10n or source_l10n or has_localized_jacket:
results.append(
SongLocalization(
id=id_,
lang=lang_key,
title=title_l10n,
source=source_l10n,
has_jacket=has_localized_jacket,
)
)
difficulties = song["difficulties"]
for difficulty in difficulties:
difficulty_title_localized = difficulty.get("title_localized", {})
difficulty_artist_localized = difficulty.get("artist_localized", {})
rating_class = difficulty["ratingClass"]
results.append(
Difficulty(
song_id=id_,
rating_class=rating_class,
rating=difficulty["rating"],
is_rating_plus=difficulty.get("ratingPlus", False),
chart_designer=blank_str_to_none(
difficulty.get("chartDesigner")
),
jacket_designer=blank_str_to_none(
difficulty.get("jacketDesigner")
),
has_overriding_audio=difficulty.get("audioOverride", False),
has_overriding_jacket=difficulty.get("jacketOverride", False),
jacket_night=blank_str_to_none(difficulty.get("jacket_night")),
added_at=datetime.fromtimestamp(
difficulty["date"], tz=timezone.utc
)
if difficulty.get("date")
else None,
bg=blank_str_to_none(difficulty.get("bg")),
version=blank_str_to_none(difficulty.get("version")),
title=blank_str_to_none(difficulty_title_localized.get("en")),
artist=blank_str_to_none(difficulty_artist_localized.get("en")),
is_legacy11=difficulty.get("legacy11", False),
)
)
for lang_key in ARCAEA_LANGUAGE_KEYS:
difficuly_title_l10n = blank_str_to_none(
difficulty_title_localized.get(lang_key)
)
difficuly_artist_l10n = blank_str_to_none(
difficulty_artist_localized.get(lang_key)
)
if difficuly_title_l10n:
results.append(
DifficultyLocalization(
song_id=id_,
rating_class=rating_class,
lang=lang_key,
title=difficuly_title_l10n,
artist=difficuly_artist_l10n,
)
)
results.extend(cls.parse_song(song))
results.extend(cls.parse_difficulties(song))
return results