10 Commits

Author SHA1 Message Date
2dc96e06aa refactor(db)!: replace version column using Version 2025-08-04 01:07:16 +08:00
308b087b94 feat(db): new model VersionDate 2025-08-04 01:04:57 +08:00
a8164f37e2 feat: add Version 2025-08-04 00:58:11 +08:00
ab03b27730 refactor(db)!: ChartInfo.notes can be NULL 2025-07-19 00:49:40 +08:00
ff248d1363 feat(constants): ArcaeaLocalizationLanguage 2025-06-06 17:02:37 +08:00
4c46b43d4a chore(db): remove extra migration comments 2025-06-06 12:30:50 +08:00
826e097d2e fix(db): version property in migrations 2025-06-06 00:00:55 +08:00
02788878a5 feat(db): v1 migration 2025-06-05 23:59:37 +08:00
2b8b13ca95 fix: adapt to new model and tests
- I actually forgot I wrote tests lol
2025-05-31 18:12:58 +08:00
743bbe209f fix(db): DifficultyLocalization.artist 2025-05-31 18:12:00 +08:00
25 changed files with 356 additions and 134 deletions

View File

@ -1,5 +1,6 @@
from .arcaea import (
ArcaeaLanguage,
ArcaeaLocalizationLanguage,
ArcaeaPlayResultClearType,
ArcaeaPlayResultModifier,
ArcaeaRatingClass,
@ -8,6 +9,7 @@ from .arcaea import (
__all__ = [
"ArcaeaLanguage",
"ArcaeaLocalizationLanguage",
"ArcaeaPlayResultClearType",
"ArcaeaPlayResultModifier",
"ArcaeaRatingClass",

View File

@ -31,6 +31,13 @@ class ArcaeaPlayResultClearType(IntEnum):
EASY_CLEAR = 5
class ArcaeaLocalizationLanguage(Enum):
JA = "ja"
KO = "ko"
ZH_HANT = "zh-Hant"
ZH_HANS = "zh-Hans"
class ArcaeaLanguage(Enum):
EN = "en"
JA = "ja"

View File

@ -11,7 +11,6 @@ from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}

View File

@ -15,7 +15,6 @@ from alembic import context, op
from arcaea_offline.database.migrations.legacies.v5 import ForceTimezoneDateTime
# revision identifiers, used by Alembic.
revision: str = "0ca6733e40dc"
down_revision: Union[str, None] = "a3f9d48b7de3"
branch_labels: Union[str, Sequence[str], None] = None
@ -27,7 +26,7 @@ def upgrade(
data_migration: bool = True,
data_migration_options: Any = None,
) -> None:
op.create_table(
property_tbl = op.create_table(
"property",
sa.Column("key", sa.String(), nullable=False),
sa.Column("value", sa.String(), nullable=False),
@ -295,6 +294,8 @@ def upgrade(
op.drop_table("scores_old")
op.execute(sa.insert(property_tbl).values(key="version", value="5"))
def downgrade() -> None:
raise NotImplementedError(

View File

@ -1,7 +1,7 @@
"""v1 to v4
Revision ID: a3f9d48b7de3
Revises:
Revises: b7a0accfc17f
Create Date: 2024-11-24 00:03:07.697165
"""
@ -12,9 +12,8 @@ from typing import Mapping, Optional, Sequence, TypedDict, Union
import sqlalchemy as sa
from alembic import context, op
# revision identifiers, used by Alembic.
revision: str = "a3f9d48b7de3"
down_revision: Union[str, None] = None
down_revision: Union[str, None] = "b7a0accfc17f"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
@ -268,6 +267,16 @@ def upgrade(
op.drop_table("scores_old")
op.drop_table("properties", if_exists=True)
properties_tbl = op.create_table(
"properties",
sa.Column("key", sa.TEXT(), nullable=False),
sa.Column("value", sa.TEXT(), nullable=False),
sa.PrimaryKeyConstraint("key", name="pk_properties"),
)
op.execute(sa.insert(properties_tbl).values(key="version", value="4"))
def downgrade() -> None:
raise NotImplementedError(

View File

@ -0,0 +1,106 @@
"""v1
Revision ID: b7a0accfc17f
Revises:
Create Date: 2025-06-05 22:17:13.327742
"""
from typing import Any, Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "b7a0accfc17f"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade(
*,
data_migration: bool = False,
data_migration_options: Any = None,
) -> None:
op.create_table(
"charts",
sa.Column("song_id", sa.TEXT(), nullable=False),
sa.Column("rating_class", sa.INTEGER(), nullable=False),
sa.Column("name_en", sa.TEXT(), nullable=False),
sa.Column("name_jp", sa.TEXT()),
sa.Column("artist", sa.TEXT(), nullable=False),
sa.Column("bpm", sa.TEXT(), nullable=False),
sa.Column("bpm_base", sa.REAL(), nullable=False),
sa.Column("package_id", sa.TEXT(), nullable=False),
sa.Column("time", sa.INTEGER()),
sa.Column("side", sa.INTEGER(), nullable=False),
sa.Column("world_unlock", sa.BOOLEAN(), nullable=False),
sa.Column("remote_download", sa.BOOLEAN()),
sa.Column("bg", sa.TEXT(), nullable=False),
sa.Column("date", sa.INTEGER(), nullable=False),
sa.Column("version", sa.TEXT(), nullable=False),
sa.Column("difficulty", sa.INTEGER(), nullable=False),
sa.Column("rating", sa.INTEGER(), nullable=False),
sa.Column("note", sa.INTEGER(), nullable=False),
sa.Column("chart_designer", sa.TEXT()),
sa.Column("jacket_designer", sa.TEXT()),
sa.Column("jacket_override", sa.BOOLEAN(), nullable=False),
sa.Column("audio_override", sa.BOOLEAN(), nullable=False),
sa.PrimaryKeyConstraint("song_id", "rating_class", name="pk_charts"),
)
op.create_table(
"aliases",
sa.Column("song_id", sa.TEXT(), nullable=False),
sa.Column("alias", sa.TEXT(), nullable=False),
sa.PrimaryKeyConstraint("song_id", "alias", name="pk_aliases"),
)
op.create_table(
"packages",
sa.Column("package_id", sa.TEXT(), nullable=False),
sa.Column("name", sa.TEXT(), nullable=False),
sa.PrimaryKeyConstraint("package_id", name="pk_packages"),
)
op.create_table(
"scores",
sa.Column("id", sa.INTEGER(), autoincrement=True),
sa.Column("song_id", sa.TEXT(), nullable=False),
sa.Column("rating_class", sa.INTEGER(), nullable=False),
sa.Column("score", sa.INTEGER(), nullable=False),
sa.Column("pure", sa.INTEGER()),
sa.Column("far", sa.INTEGER()),
sa.Column("lost", sa.INTEGER()),
sa.Column("time", sa.INTEGER(), nullable=False),
sa.Column("max_recall", sa.INTEGER()),
sa.Column("clear_type", sa.INTEGER()),
sa.PrimaryKeyConstraint("id", name="pk_scores"),
sa.ForeignKeyConstraint(
["song_id", "rating_class"],
["charts.song_id", "charts.rating_class"],
name="fk_scores_song_id_charts",
onupdate="CASCADE",
ondelete="NO ACTION",
),
)
properties_tbl = op.create_table(
"properties",
sa.Column("key", sa.TEXT(), nullable=False),
sa.Column("value", sa.TEXT(), nullable=False),
# according to the commit history this was a unique constraint
# but i remember sqlalchemy complains if you don't add a primary key
# anyways above tables have modifications like this too
# and nobody actualty uses v1 schema so who cares
sa.PrimaryKeyConstraint("key", name="pk_properties"),
)
# there should be CREATE VIEWs here but im skipping them
# because nobody actually checks this file
op.execute(sa.insert(properties_tbl).values(key="db_version", value="1"))
def downgrade() -> None:
raise NotImplementedError("This is the first revision bro")

View File

@ -0,0 +1,31 @@
"""version date
Revision ID: f82ac2f445a5
Revises: 0ca6733e40dc
Create Date: 2025-07-19 17:11:27.448574
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "f82ac2f445a5"
down_revision: Union[str, None] = "0ca6733e40dc"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"version_date",
sa.Column("version", sa.String(), nullable=False),
sa.Column("songlist_at", sa.DateTime(), nullable=False),
sa.Column("published_at", sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint("version", name=op.f("pk_version_date")),
)
def downgrade() -> None:
op.drop_table("version_date")

View File

@ -4,6 +4,7 @@ from .config import Property
from .difficulty import Difficulty, DifficultyLocalization
from .pack import Pack, PackLocalization
from .song import Song, SongLocalization
from .version_date import VersionDate
from .chart import Chart # isort: skip
from .play_result import (
@ -29,4 +30,5 @@ __all__ = [
"Property",
"Song",
"SongLocalization",
"VersionDate",
]

View File

@ -4,10 +4,13 @@ from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm.exc import DetachedInstanceError
from ._types import ForceTimezoneDateTime
from arcaea_offline.utils import Version
from ._types import ForceTimezoneDateTime, VersionDatabaseType
TYPE_ANNOTATION_MAP = {
datetime: ForceTimezoneDateTime,
Version: VersionDatabaseType,
}

View File

@ -1,9 +1,11 @@
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import DateTime
from sqlalchemy import DateTime, String
from sqlalchemy.types import TypeDecorator
from arcaea_offline.utils import Version
class ForceTimezoneDateTime(TypeDecorator):
"""
@ -26,3 +28,23 @@ class ForceTimezoneDateTime(TypeDecorator):
if value is not None:
value = value.replace(tzinfo=timezone.utc)
return value
class VersionDatabaseType(TypeDecorator):
impl = String
cache_ok = True
def process_bind_param(self, value: Optional[Version], dialect):
if value is None:
return None
if not isinstance(value, Version):
raise ValueError("Input is not a Version instance.")
return str(f"{value.first}.{value.second}.{value.third}")
def process_result_value(self, value: Optional[str], dialect):
if value is None:
return None
return Version(*(map(int, value.split("."))))

View File

@ -5,6 +5,8 @@ from typing import TYPE_CHECKING, Optional
from sqlalchemy import ForeignKeyConstraint, Integer, Numeric, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from arcaea_offline.utils import Version
from ._base import ModelBase, ReprHelper
from ._types import ForceTimezoneDateTime
@ -28,6 +30,6 @@ class ChartInfo(ModelBase, ReprHelper):
song_id: Mapped[str] = mapped_column(String, primary_key=True)
rating_class: Mapped[int] = mapped_column(Integer, primary_key=True)
constant: Mapped[Decimal] = mapped_column(Numeric, nullable=False)
notes: Mapped[int] = mapped_column(Integer)
notes: Mapped[Optional[int]] = mapped_column(Integer)
added_at: Mapped[datetime] = mapped_column(ForceTimezoneDateTime, primary_key=True)
version: Mapped[Optional[str]] = mapped_column(String)
version: Mapped[Optional[Version]]

View File

@ -13,6 +13,8 @@ from sqlalchemy import (
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from arcaea_offline.utils import Version
from ._base import ModelBase, ReprHelper
from ._types import ForceTimezoneDateTime
@ -65,7 +67,7 @@ class Difficulty(ModelBase, ReprHelper):
bpm: Mapped[Optional[str]] = mapped_column(String)
bpm_base: Mapped[Optional[Decimal]] = mapped_column(Numeric(asdecimal=True))
added_at: Mapped[Optional[datetime]] = mapped_column(ForceTimezoneDateTime)
version: Mapped[Optional[str]] = mapped_column(String)
version: Mapped[Optional[Version]]
is_legacy11: Mapped[bool] = mapped_column(
Boolean, nullable=False, insert_default=False, server_default=text("0")
)
@ -90,3 +92,4 @@ class DifficultyLocalization(ModelBase, ReprHelper):
lang: Mapped[str] = mapped_column(String, primary_key=True)
title: Mapped[Optional[str]] = mapped_column(String)
artist: Mapped[Optional[str]] = mapped_column(String)

View File

@ -5,6 +5,8 @@ from typing import TYPE_CHECKING, Optional
from sqlalchemy import Boolean, ForeignKey, Integer, Numeric, String, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from arcaea_offline.utils import Version
from ._base import ModelBase, ReprHelper
from ._types import ForceTimezoneDateTime
@ -42,7 +44,7 @@ class Song(ModelBase, ReprHelper):
added_at: Mapped[datetime] = mapped_column(
ForceTimezoneDateTime, nullable=False, index=True
)
version: Mapped[Optional[str]] = mapped_column(String)
version: Mapped[Optional[Version]]
bpm: Mapped[Optional[str]] = mapped_column(String)
bpm_base: Mapped[Optional[Decimal]] = mapped_column(Numeric(asdecimal=True))

View File

@ -0,0 +1,15 @@
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from arcaea_offline.utils import Version
from ._base import ModelBase, ReprHelper
class VersionDate(ModelBase, ReprHelper):
__tablename__ = "version_date"
version: Mapped[Version] = mapped_column(primary_key=True)
songlist_at: Mapped[datetime]
published_at: Mapped[datetime]

View File

@ -3,6 +3,7 @@ packlist and songlist parsers
"""
import json
from datetime import datetime, timezone
from typing import List, Union
from arcaea_offline.constants.enums import (
@ -12,12 +13,11 @@ from arcaea_offline.constants.enums import (
)
from arcaea_offline.database.models import (
Difficulty,
DifficultyLocalized,
DifficultyLocalization,
Pack,
PackLocalized,
PackLocalization,
Song,
SongLocalized,
SongSearchWord,
SongLocalization,
)
@ -27,11 +27,11 @@ class ArcaeaListParser:
class ArcaeaPacklistParser(ArcaeaListParser):
def parse(self) -> List[Union[Pack, PackLocalized]]:
def parse(self) -> List[Union[Pack, PackLocalization]]:
root = json.loads(self.list_text)
packs = root["packs"]
results: List[Union[Pack, PackLocalized]] = [
results: List[Union[Pack, PackLocalization]] = [
Pack(id="single", name="Memory Archive")
]
for item in packs:
@ -48,8 +48,8 @@ class ArcaeaPacklistParser(ArcaeaListParser):
)
if name_localized or description_localized:
pack_localized = PackLocalized(id=pack.id)
pack_localized.lang = ArcaeaLanguage(key.value)
pack_localized = PackLocalization(id=pack.id)
pack_localized.lang = key.value
pack_localized.name = name_localized
pack_localized.description = description_localized
results.append(pack_localized)
@ -58,7 +58,7 @@ class ArcaeaPacklistParser(ArcaeaListParser):
class ArcaeaSonglistParser(ArcaeaListParser):
def parse_songs(self) -> List[Union[Song, SongLocalized, SongSearchWord]]:
def parse_songs(self) -> List[Union[Song, SongLocalization]]:
root = json.loads(self.list_text)
songs = root["songs"]
@ -72,11 +72,9 @@ class ArcaeaSonglistParser(ArcaeaListParser):
song.bpm = item["bpm"]
song.bpm_base = item["bpm_base"]
song.pack_id = item["set"]
song.audio_preview = item["audioPreview"]
song.audio_preview_end = item["audioPreviewEnd"]
song.side = ArcaeaSongSide(item["side"])
song.version = item["version"]
song.date = item["date"]
song.added_at = datetime.fromtimestamp(item["date"], tz=timezone.utc)
song.bg = item.get("bg")
song.bg_inverse = item.get("bg_inverse")
if item.get("bg_daynight"):
@ -95,32 +93,32 @@ class ArcaeaSonglistParser(ArcaeaListParser):
)
if title_localized or source_localized:
song_localized = SongLocalized(id=song.id)
song_localized.lang = ArcaeaLanguage(lang.value)
song_localized = SongLocalization(id=song.id)
song_localized.lang = lang.value
song_localized.title = title_localized
song_localized.source = source_localized
results.append(song_localized)
# SongSearchTitle
search_titles = item.get("search_title", {}).get(lang.value, None)
if search_titles:
for search_title in search_titles:
song_search_word = SongSearchWord(
id=song.id, lang=lang.value, type=1, value=search_title
)
results.append(song_search_word)
# TODO: SongSearchTitle?
# search_titles = item.get("search_title", {}).get(lang.value, None)
# if search_titles:
# for search_title in search_titles:
# song_search_word = SongSearchWord(
# id=song.id, lang=lang.value, type=1, value=search_title
# )
# results.append(song_search_word)
search_artists = item.get("search_artist", {}).get(lang.value, None)
if search_artists:
for search_artist in search_artists:
song_search_word = SongSearchWord(
id=song.id, lang=lang.value, type=2, value=search_artist
)
results.append(song_search_word)
# search_artists = item.get("search_artist", {}).get(lang.value, None)
# if search_artists:
# for search_artist in search_artists:
# song_search_word = SongSearchWord(
# id=song.id, lang=lang.value, type=2, value=search_artist
# )
# results.append(song_search_word)
return results
def parse_difficulties(self) -> List[Union[Difficulty, DifficultyLocalized]]:
def parse_difficulties(self) -> List[Union[Difficulty, DifficultyLocalization]]:
root = json.loads(self.list_text)
songs = root["songs"]
@ -138,11 +136,11 @@ class ArcaeaSonglistParser(ArcaeaListParser):
difficulty.song_id = song["id"]
difficulty.rating_class = ArcaeaRatingClass(item["ratingClass"])
difficulty.rating = item["rating"]
difficulty.rating_plus = item.get("ratingPlus") or False
difficulty.is_rating_plus = item.get("ratingPlus") or False
difficulty.chart_designer = item["chartDesigner"]
difficulty.jacket_desginer = item.get("jacketDesigner") or None
difficulty.audio_override = item.get("audioOverride") or False
difficulty.jacket_override = item.get("jacketOverride") or False
difficulty.jacket_designer = item.get("jacketDesigner") or None
difficulty.has_overriding_audio = item.get("audioOverride") or False
difficulty.has_overriding_jacket = item.get("jacketOverride") or False
difficulty.jacket_night = item.get("jacketNight") or None
difficulty.title = item.get("title_localized", {}).get("en") or None
difficulty.artist = item.get("artist") or None
@ -151,7 +149,11 @@ class ArcaeaSonglistParser(ArcaeaListParser):
difficulty.bpm = item.get("bpm") or None
difficulty.bpm_base = item.get("bpm_base") or None
difficulty.version = item.get("version") or None
difficulty.date = item.get("date") or None
difficulty.added_at = (
datetime.fromtimestamp(item["date"], tz=timezone.utc)
if item.get("date") is not None
else None
)
results.append(difficulty)
for lang in ArcaeaLanguage:
@ -163,11 +165,11 @@ class ArcaeaSonglistParser(ArcaeaListParser):
)
if title_localized or artist_localized:
difficulty_localized = DifficultyLocalized(
difficulty_localized = DifficultyLocalization(
song_id=difficulty.song_id,
rating_class=difficulty.rating_class,
)
difficulty_localized.lang = ArcaeaLanguage(lang.value)
difficulty_localized.lang = lang.value
difficulty_localized.title = title_localized
difficulty_localized.artist = artist_localized
results.append(difficulty_localized)

View File

@ -85,7 +85,7 @@ class ArcaeaOnlineApiParser:
play_result.pure = item["perfect_count"]
play_result.far = item["near_count"]
play_result.lost = item["miss_count"]
play_result.date = date
play_result.played_at = date
play_result.modifier = ArcaeaPlayResultModifier(item["modifier"])
play_result.clear_type = ArcaeaPlayResultClearType(item["clear_type"])

View File

@ -100,19 +100,18 @@ class ArcaeaSt3Parser:
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,
)
)
play_result = PlayResult()
play_result.song_id = song_id
play_result.rating_class = rating_class_enum
play_result.score = score
play_result.pure = pure
play_result.far = far
play_result.lost = lost
play_result.played_at = date
play_result.modifier = modifier_enum
play_result.clear_type = clear_type_enum
play_result.comment = import_comment
entities.append(play_result)
return entities

View File

@ -0,0 +1,5 @@
from .version import Version
__all__ = [
"Version",
]

View File

@ -0,0 +1,27 @@
from typing import NamedTuple
class Version(NamedTuple):
first: int
second: int
third: int
@classmethod
def from_string(cls, version_str: str):
version_str = version_str.removesuffix("c")
parts = version_str.split(".")
if len(parts) not in {2, 3}:
raise ValueError(f"Invalid version string {version_str}")
try:
if len(parts) == 2: # noqa: PLR2004
parts.append("0")
first, second, third = map(int, parts)
except ValueError as e:
raise ValueError(f"Invalid version string {version_str}") from e
return cls(first, second, third)
def __str__(self):
return f"{self.first}.{self.second}.{self.third}"

View File

@ -10,11 +10,13 @@ Database model v5 common relationships
└───────────┘
"""
from datetime import datetime, timezone
from arcaea_offline.constants.enums import ArcaeaRatingClass
from arcaea_offline.database.models import (
ChartInfo,
Difficulty,
ModelsV5Base,
ModelBase,
Pack,
PlayResult,
Song,
@ -24,7 +26,7 @@ from arcaea_offline.database.models import (
class TestSongRelationships:
@staticmethod
def init_db(session):
ModelsV5Base.metadata.create_all(session.bind)
ModelBase.metadata.create_all(session.bind)
def test_relationships(self, db_session):
self.init_db(db_session)
@ -45,52 +47,56 @@ class TestSongRelationships:
title=title_en,
artist=artist_en,
pack_id=pack.id,
added_at=datetime(2024, 7, 5, tzinfo=timezone.utc),
)
difficulty_pst = Difficulty(
song_id=song.id,
rating_class=ArcaeaRatingClass.PAST,
rating=2,
rating_plus=False,
is_rating_plus=False,
)
chart_info_pst = ChartInfo(
song_id=song.id,
rating_class=ArcaeaRatingClass.PAST,
constant=20,
notes=200,
added_at=datetime(2024, 7, 12, tzinfo=timezone.utc),
)
difficulty_prs = Difficulty(
song_id=song.id,
rating_class=ArcaeaRatingClass.PRESENT,
rating=7,
rating_plus=True,
is_rating_plus=True,
)
chart_info_prs = ChartInfo(
song_id=song.id,
rating_class=ArcaeaRatingClass.PRESENT,
constant=78,
notes=780,
added_at=datetime(2024, 7, 12, tzinfo=timezone.utc),
)
difficulty_ftr = Difficulty(
song_id=song.id,
rating_class=ArcaeaRatingClass.FUTURE,
rating=10,
rating_plus=True,
is_rating_plus=True,
)
chart_info_ftr = ChartInfo(
song_id=song.id,
rating_class=ArcaeaRatingClass.FUTURE,
constant=109,
notes=1090,
added_at=datetime(2024, 7, 12, tzinfo=timezone.utc),
)
difficulty_etr = Difficulty(
song_id=song.id,
rating_class=ArcaeaRatingClass.ETERNAL,
rating=9,
rating_plus=True,
is_rating_plus=True,
)
play_result_ftr = PlayResult(
@ -124,20 +130,19 @@ class TestSongRelationships:
difficulty_ftr,
difficulty_etr,
]
assert song.charts_info == [chart_info_pst, chart_info_prs, chart_info_ftr]
assert difficulty_pst.song == song
assert difficulty_prs.song == song
assert difficulty_ftr.song == song
assert difficulty_etr.song == song
assert difficulty_pst.chart_info == chart_info_pst
assert difficulty_prs.chart_info == chart_info_prs
assert difficulty_ftr.chart_info == chart_info_ftr
assert difficulty_etr.chart_info is None
assert difficulty_pst.chart_info_list == [chart_info_pst]
assert difficulty_prs.chart_info_list == [chart_info_prs]
assert difficulty_ftr.chart_info_list == [chart_info_ftr]
assert difficulty_etr.chart_info_list == []
assert chart_info_pst.difficulty == difficulty_pst
assert chart_info_prs.difficulty == difficulty_prs
assert chart_info_ftr.difficulty == difficulty_ftr
assert play_result_ftr.difficulty == difficulty_ftr
# assert play_result_ftr.difficulty == difficulty_ftr

View File

@ -5,13 +5,13 @@ Pack <> PackLocalized
"""
from arcaea_offline.constants.enums import ArcaeaLanguage
from arcaea_offline.database.models import ModelsV5Base, Pack, PackLocalized
from arcaea_offline.database.models import ModelBase, Pack, PackLocalization
class TestPackRelationships:
@staticmethod
def init_db(session):
ModelsV5Base.metadata.create_all(session.bind)
ModelBase.metadata.create_all(session.bind)
def test_localized_objects(self, db_session):
self.init_db(db_session)
@ -26,20 +26,16 @@ class TestPackRelationships:
description=description_en,
)
pkid_ja = 1
description_ja = "普通のパートナー「∅」と一緒に、\n不人気フレームワーク「Arcaea Offline」より、\n一般的なデータベース・モデルを通過する。"
pack_localized_ja = PackLocalized(
pkid=pkid_ja,
pack_localized_ja = PackLocalization(
id=pack_id,
lang=ArcaeaLanguage.JA.value,
name=None,
description=description_ja,
)
pkid_zh_hans = 2
description_zh_hans = "与平凡的「∅」一起,\n在没人用的「Arcaea Offline」框架里\n一同探索随处可见的数据库模型。"
pack_localized_zh_hans = PackLocalized(
pkid=pkid_zh_hans,
pack_localized_zh_hans = PackLocalization(
id=pack_id,
lang=ArcaeaLanguage.ZH_HANS.value,
name=None,
@ -49,11 +45,11 @@ class TestPackRelationships:
db_session.add_all([pack, pack_localized_ja])
db_session.commit()
assert len(pack.localized_objects) == len([pack_localized_ja])
assert len(pack.localized_entries) == len([pack_localized_ja])
assert pack_localized_ja.parent.description == pack.description
assert pack_localized_ja.pack.description == pack.description
# relationships should be viewonly
# test back populates
new_pack = Pack(
id=f"{pack_id}_new",
name="NEW",
@ -61,9 +57,10 @@ class TestPackRelationships:
)
db_session.add(new_pack)
pack_localized_ja.parent = new_pack
pack.localized_objects.append(pack_localized_zh_hans)
pack_localized_ja.pack = new_pack
pack.localized_entries.append(pack_localized_zh_hans)
db_session.commit()
assert pack_localized_ja.parent == pack
assert len(pack.localized_objects) == 1
assert pack_localized_ja.pack.id == new_pack.id
# TODO: this failes but why
assert len(pack.localized_entries) == 2

View File

@ -6,13 +6,15 @@ Chart functionalities
- Difficulty song info overriding
"""
from datetime import datetime, timezone
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models import (
Chart,
ChartInfo,
Difficulty,
ModelsV5Base,
ModelsV5ViewBase,
ModelBase,
ModelViewBase,
Pack,
Song,
)
@ -20,8 +22,8 @@ from arcaea_offline.database.models import (
class TestChart:
def init_db(self, session):
ModelsV5Base.metadata.create_all(session.bind)
ModelsV5ViewBase.metadata.create_all(session.bind)
ModelBase.metadata.create_all(session.bind)
ModelViewBase.metadata.create_all(session.bind)
def test_basic(self, db_session):
self.init_db(db_session)
@ -37,20 +39,23 @@ class TestChart:
title="~TEST~",
artist="~test~",
pack_id=pack_id,
added_at=datetime(2024, 7, 4, tzinfo=timezone.utc),
)
difficulty = Difficulty(
song_id=song_id,
rating_class=rating_class,
rating=9,
rating_plus=True,
is_rating_plus=True,
)
chart_info = ChartInfo(
song_id=song_id,
rating_class=rating_class,
constant=98,
notes=980,
added_at=datetime(2024, 7, 12, tzinfo=timezone.utc),
)
db_session.add_all([pack, song, difficulty, chart_info])
db_session.commit()
chart: Chart = (
db_session.query(Chart)
@ -64,7 +69,7 @@ class TestChart:
assert chart.artist == song.artist
assert chart.pack_id == song.pack_id
assert chart.rating == difficulty.rating
assert chart.rating_plus == difficulty.rating_plus
assert chart.is_rating_plus == difficulty.is_rating_plus
assert chart.constant == chart_info.constant
assert chart.notes == chart_info.notes
@ -82,12 +87,13 @@ class TestChart:
title="~TEST~",
artist="~test~",
pack_id=pack_id,
added_at=datetime(2024, 7, 4, tzinfo=timezone.utc),
)
difficulty = Difficulty(
song_id=song_id,
rating_class=rating_class,
rating=9,
rating_plus=True,
is_rating_plus=True,
title="~TEST DIFF~",
artist="~diff~",
)
@ -96,8 +102,10 @@ class TestChart:
rating_class=rating_class,
constant=98,
notes=980,
added_at=datetime(2024, 7, 12, tzinfo=timezone.utc),
)
db_session.add_all([pack, song, difficulty, chart_info])
db_session.commit()
chart: Chart = (
db_session.query(Chart)

View File

@ -2,10 +2,9 @@ from datetime import datetime, timedelta, timezone
from enum import IntEnum
from typing import Optional
from sqlalchemy import text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from arcaea_offline.database.models._custom_types import DbIntEnum, TZDateTime
from arcaea_offline.database.models._types import ForceTimezoneDateTime
class TestIntEnum(IntEnum):
@ -22,43 +21,19 @@ class TestBase(DeclarativeBase):
id: Mapped[int] = mapped_column(primary_key=True)
class IntEnumTestModel(TestBase):
__tablename__ = "test_int_enum"
value: Mapped[Optional[TestIntEnum]] = mapped_column(DbIntEnum(TestIntEnum))
class TZDatetimeTestModel(TestBase):
class ForceTimezoneDatetimeTestModel(TestBase):
__tablename__ = "test_tz_datetime"
value: Mapped[Optional[datetime]] = mapped_column(TZDateTime)
value: Mapped[Optional[datetime]] = mapped_column(ForceTimezoneDateTime)
class TestCustomTypes:
def test_int_enum(self, db_session):
def _query_value(_id: int):
return db_session.execute(
text(
f"SELECT value FROM {IntEnumTestModel.__tablename__} WHERE id = {_id}"
)
).one()[0]
TestBase.metadata.create_all(db_session.bind, checkfirst=False)
basic_obj = IntEnumTestModel(id=1, value=TestIntEnum.TWO)
null_obj = IntEnumTestModel(id=2, value=None)
db_session.add(basic_obj)
db_session.add(null_obj)
db_session.commit()
assert _query_value(1) == TestIntEnum.TWO.value
assert _query_value(2) is None
def test_tz_datetime(self, db_session):
def test_force_timezone_datetime(self, db_session):
TestBase.metadata.create_all(db_session.bind, checkfirst=False)
dt1 = datetime.now(tz=timezone(timedelta(hours=8)))
basic_obj = TZDatetimeTestModel(id=1, value=dt1)
null_obj = TZDatetimeTestModel(id=2, value=None)
basic_obj = ForceTimezoneDatetimeTestModel(id=1, value=dt1)
null_obj = ForceTimezoneDatetimeTestModel(id=2, value=None)
db_session.add(basic_obj)
db_session.add(null_obj)
db_session.commit()

View File

@ -6,7 +6,7 @@ from arcaea_offline.constants.enums.arcaea import (
ArcaeaPlayResultModifier,
ArcaeaRatingClass,
)
from arcaea_offline.database.models.play_results import PlayResult
from arcaea_offline.database.models.play_result import PlayResult
from arcaea_offline.external.importers.arcaea.online import ArcaeaOnlineApiParser
API_RESULT = {
@ -80,9 +80,9 @@ class TestArcaeaOnlineApiParser:
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.played_at == 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)
assert test2.played_at == datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc)

View File

@ -31,7 +31,7 @@ class TestArcaeaSt3Parser:
assert test1.pure == 895
assert test1.far == 32
assert test1.lost == 22
assert test1.date == datetime.fromtimestamp(1722100000).astimezone()
assert test1.played_at == datetime.fromtimestamp(1722100000).astimezone()
assert test1.clear_type is ArcaeaPlayResultClearType.TRACK_LOST
assert test1.modifier is ArcaeaPlayResultModifier.HARD
@ -48,7 +48,7 @@ class TestArcaeaSt3Parser:
assert corrupt2.modifier is None
date1 = next(filter(lambda x: x.song_id == "date1", play_results))
assert date1.date is None
assert date1.played_at is None
def test_invalid_input(self):
pytest.raises(TypeError, ArcaeaSt3Parser.parse, "abcdefghijklmn")