refactor!: sqlalchemy database models

src/arcaea_offline/database/db.py

@@ -11,8 +11,10 @@ from arcaea_offline.external.exports import (
     ScoreExport,
     exporters,
 )
-from arcaea_offline.models.config import ConfigBase, Property
-from arcaea_offline.models.scores import (
+from arcaea_offline.singleton import Singleton
+
+from .models.v4.config import ConfigBase, Property
+from .models.v4.scores import (
     CalculatedPotential,
     Score,
     ScoreBest,
@@ -20,7 +22,7 @@ from arcaea_offline.models.scores import (
     ScoresBase,
     ScoresViewBase,
 )
-from arcaea_offline.models.songs import (
+from .models.v4.songs import (
     Chart,
     ChartInfo,
     Difficulty,
@@ -32,7 +34,6 @@ from arcaea_offline.models.songs import (
     SongsBase,
     SongsViewBase,
 )
-from arcaea_offline.singleton import Singleton
 
 logger = logging.getLogger(__name__)
 

src/arcaea_offline/database/models/v4/__init__.py

@@ -0,0 +1,21 @@
+from .config import ConfigBase, Property
+from .scores import (
+    CalculatedPotential,
+    Score,
+    ScoreBest,
+    ScoreCalculated,
+    ScoresBase,
+    ScoresViewBase,
+)
+from .songs import (
+    Chart,
+    ChartInfo,
+    Difficulty,
+    DifficultyLocalized,
+    Pack,
+    PackLocalized,
+    Song,
+    SongLocalized,
+    SongsBase,
+    SongsViewBase,
+)

src/arcaea_offline/database/models/v4/common.py

@@ -0,0 +1,36 @@
+# pylint: disable=too-few-public-methods
+
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm.exc import DetachedInstanceError
+
+
+class ReprHelper:
+    # pylint: disable=no-member
+
+    def _repr(self, **kwargs) -> str:
+        """
+        Helper for __repr__
+
+        https://stackoverflow.com/a/55749579/16484891
+
+        CC BY-SA 4.0
+        """
+        field_strings = []
+        at_least_one_attached_attribute = False
+        for key, field in kwargs.items():
+            try:
+                field_strings.append(f"{key}={field!r}")
+            except DetachedInstanceError:
+                field_strings.append(f"{key}=DetachedInstanceError")
+            else:
+                at_least_one_attached_attribute = True
+        if at_least_one_attached_attribute:
+            return f"<{self.__class__.__name__}({','.join(field_strings)})>"
+        return f"<{self.__class__.__name__} {id(self)}>"
+
+    def __repr__(self):
+        if isinstance(self, DeclarativeBase):
+            return self._repr(
+                **{c.key: getattr(self, c.key) for c in self.__table__.columns}
+            )
+        return super().__repr__()

src/arcaea_offline/database/models/v4/config.py

@@ -0,0 +1,22 @@
+# pylint: disable=too-few-public-methods
+
+from sqlalchemy import TEXT
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+from .common import ReprHelper
+
+__all__ = [
+    "ConfigBase",
+    "Property",
+]
+
+
+class ConfigBase(DeclarativeBase, ReprHelper):
+    pass
+
+
+class Property(ConfigBase):
+    __tablename__ = "properties"
+
+    key: Mapped[str] = mapped_column(TEXT(), primary_key=True)
+    value: Mapped[str] = mapped_column(TEXT())

src/arcaea_offline/database/models/v4/scores.py

@@ -0,0 +1,188 @@
+# pylint: disable=too-few-public-methods, duplicate-code
+
+from typing import Optional
+
+from sqlalchemy import TEXT, case, func, inspect, select, text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy_utils import create_view
+
+from .common import ReprHelper
+from .songs import ChartInfo, Difficulty
+
+__all__ = [
+    "ScoresBase",
+    "Score",
+    "ScoresViewBase",
+    "ScoreCalculated",
+    "ScoreBest",
+    "CalculatedPotential",
+]
+
+
+class ScoresBase(DeclarativeBase, ReprHelper):
+    pass
+
+
+class Score(ScoresBase):
+    __tablename__ = "scores"
+
+    id: Mapped[int] = mapped_column(autoincrement=True, primary_key=True)
+    song_id: Mapped[str] = mapped_column(TEXT())
+    rating_class: Mapped[int]
+    score: Mapped[int]
+    pure: Mapped[Optional[int]]
+    far: Mapped[Optional[int]]
+    lost: Mapped[Optional[int]]
+    date: Mapped[Optional[int]]
+    max_recall: Mapped[Optional[int]]
+    modifier: Mapped[Optional[int]] = mapped_column(
+        comment="0: NORMAL, 1: EASY, 2: HARD"
+    )
+    clear_type: Mapped[Optional[int]] = mapped_column(
+        comment="0: TRACK LOST, 1: NORMAL CLEAR, 2: FULL RECALL, "
+        "3: PURE MEMORY, 4: EASY CLEAR, 5: HARD CLEAR"
+    )
+    comment: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+# How to create an SQL View with SQLAlchemy?
+# https://stackoverflow.com/a/53253105/16484891
+# CC BY-SA 4.0
+
+
+class ScoresViewBase(DeclarativeBase, ReprHelper):
+    pass
+
+
+class ScoreCalculated(ScoresViewBase):
+    __tablename__ = "scores_calculated"
+
+    id: Mapped[int]
+    song_id: Mapped[str]
+    rating_class: Mapped[int]
+    score: Mapped[int]
+    pure: Mapped[Optional[int]]
+    shiny_pure: Mapped[Optional[int]]
+    far: Mapped[Optional[int]]
+    lost: Mapped[Optional[int]]
+    date: Mapped[Optional[int]]
+    max_recall: Mapped[Optional[int]]
+    modifier: Mapped[Optional[int]]
+    clear_type: Mapped[Optional[int]]
+    potential: Mapped[float]
+    comment: Mapped[Optional[str]]
+
+    __table__ = create_view(
+        name=__tablename__,
+        selectable=select(
+            Score.id,
+            Difficulty.song_id,
+            Difficulty.rating_class,
+            Score.score,
+            Score.pure,
+            (
+                case(
+                    (
+                        (
+                            ChartInfo.notes.is_not(None)
+                            & Score.pure.is_not(None)
+                            & Score.far.is_not(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,
+            Score.lost,
+            Score.date,
+            Score.max_recall,
+            Score.modifier,
+            Score.clear_type,
+            case(
+                (Score.score >= 10000000, ChartInfo.constant / 10.0 + 2),
+                (
+                    Score.score >= 9800000,
+                    ChartInfo.constant / 10.0 + 1 + (Score.score - 9800000) / 200000.0,
+                ),
+                else_=func.max(
+                    (ChartInfo.constant / 10.0) + (Score.score - 9500000) / 300000.0,
+                    0,
+                ),
+            ).label("potential"),
+            Score.comment,
+        )
+        .select_from(Difficulty)
+        .join(
+            ChartInfo,
+            (Difficulty.song_id == ChartInfo.song_id)
+            & (Difficulty.rating_class == ChartInfo.rating_class),
+        )
+        .join(
+            Score,
+            (Difficulty.song_id == Score.song_id)
+            & (Difficulty.rating_class == Score.rating_class),
+        ),
+        metadata=ScoresViewBase.metadata,
+        cascade_on_drop=False,
+    )
+
+
+class ScoreBest(ScoresViewBase):
+    __tablename__ = "scores_best"
+
+    id: Mapped[int]
+    song_id: Mapped[str]
+    rating_class: Mapped[int]
+    score: Mapped[int]
+    pure: Mapped[Optional[int]]
+    shiny_pure: Mapped[Optional[int]]
+    far: Mapped[Optional[int]]
+    lost: Mapped[Optional[int]]
+    date: Mapped[Optional[int]]
+    max_recall: Mapped[Optional[int]]
+    modifier: Mapped[Optional[int]]
+    clear_type: Mapped[Optional[int]]
+    potential: Mapped[float]
+    comment: Mapped[Optional[str]]
+
+    __table__ = create_view(
+        name=__tablename__,
+        selectable=select(
+            *[
+                col
+                for col in inspect(ScoreCalculated).columns
+                if col.name != "potential"
+            ],
+            func.max(ScoreCalculated.potential).label("potential"),
+        )
+        .select_from(ScoreCalculated)
+        .group_by(ScoreCalculated.song_id, ScoreCalculated.rating_class)
+        .order_by(ScoreCalculated.potential.desc()),
+        metadata=ScoresViewBase.metadata,
+        cascade_on_drop=False,
+    )
+
+
+class CalculatedPotential(ScoresViewBase):
+    __tablename__ = "calculated_potential"
+
+    b30: Mapped[float]
+
+    _select_bests_subquery = (
+        select(ScoreBest.potential.label("b30_sum"))
+        .order_by(ScoreBest.potential.desc())
+        .limit(30)
+        .subquery()
+    )
+    __table__ = create_view(
+        name=__tablename__,
+        selectable=select(func.avg(_select_bests_subquery.c.b30_sum).label("b30")),
+        metadata=ScoresViewBase.metadata,
+        cascade_on_drop=False,
+    )

src/arcaea_offline/database/models/v4/songs.py

@@ -0,0 +1,241 @@
+# pylint: disable=too-few-public-methods, duplicate-code
+
+from typing import Optional
+
+from sqlalchemy import TEXT, ForeignKey, func, select
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy_utils import create_view
+
+from .common import ReprHelper
+
+__all__ = [
+    "SongsBase",
+    "Pack",
+    "PackLocalized",
+    "Song",
+    "SongLocalized",
+    "Difficulty",
+    "DifficultyLocalized",
+    "ChartInfo",
+    "SongsViewBase",
+    "Chart",
+]
+
+
+class SongsBase(DeclarativeBase, ReprHelper):
+    pass
+
+
+class Pack(SongsBase):
+    __tablename__ = "packs"
+
+    id: Mapped[str] = mapped_column(TEXT(), primary_key=True)
+    name: Mapped[str] = mapped_column(TEXT())
+    description: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+class PackLocalized(SongsBase):
+    __tablename__ = "packs_localized"
+
+    id: Mapped[str] = mapped_column(ForeignKey("packs.id"), primary_key=True)
+    name_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    name_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    name_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    name_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+    description_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    description_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    description_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    description_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+class Song(SongsBase):
+    __tablename__ = "songs"
+
+    idx: Mapped[int]
+    id: Mapped[str] = mapped_column(TEXT(), primary_key=True)
+    title: Mapped[str] = mapped_column(TEXT())
+    artist: Mapped[str] = mapped_column(TEXT())
+    set: Mapped[str] = mapped_column(TEXT())
+    bpm: Mapped[Optional[str]] = mapped_column(TEXT())
+    bpm_base: Mapped[Optional[float]]
+    audio_preview: Mapped[Optional[int]]
+    audio_preview_end: Mapped[Optional[int]]
+    side: Mapped[Optional[int]]
+    version: Mapped[Optional[str]] = mapped_column(TEXT())
+    date: Mapped[Optional[int]]
+    bg: Mapped[Optional[str]] = mapped_column(TEXT())
+    bg_inverse: Mapped[Optional[str]] = mapped_column(TEXT())
+    bg_day: Mapped[Optional[str]] = mapped_column(TEXT())
+    bg_night: Mapped[Optional[str]] = mapped_column(TEXT())
+    source: Mapped[Optional[str]] = mapped_column(TEXT())
+    source_copyright: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+class SongLocalized(SongsBase):
+    __tablename__ = "songs_localized"
+
+    id: Mapped[str] = mapped_column(ForeignKey("songs.id"), primary_key=True)
+    title_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+    search_title_ja: Mapped[Optional[str]] = mapped_column(TEXT(), comment="JSON array")
+    search_title_ko: Mapped[Optional[str]] = mapped_column(TEXT(), comment="JSON array")
+    search_title_zh_hans: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    search_title_zh_hant: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    search_artist_ja: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    search_artist_ko: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    search_artist_zh_hans: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    search_artist_zh_hant: Mapped[Optional[str]] = mapped_column(
+        TEXT(), comment="JSON array"
+    )
+    source_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    source_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    source_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    source_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+class Difficulty(SongsBase):
+    __tablename__ = "difficulties"
+
+    song_id: Mapped[str] = mapped_column(TEXT(), primary_key=True)
+    rating_class: Mapped[int] = mapped_column(primary_key=True)
+    rating: Mapped[int]
+    rating_plus: Mapped[bool]
+    chart_designer: Mapped[Optional[str]] = mapped_column(TEXT())
+    jacket_desginer: Mapped[Optional[str]] = mapped_column(TEXT())
+    audio_override: Mapped[bool]
+    jacket_override: Mapped[bool]
+    jacket_night: Mapped[Optional[str]] = mapped_column(TEXT())
+    title: Mapped[Optional[str]] = mapped_column(TEXT())
+    artist: Mapped[Optional[str]] = mapped_column(TEXT())
+    bg: Mapped[Optional[str]] = mapped_column(TEXT())
+    bg_inverse: Mapped[Optional[str]] = mapped_column(TEXT())
+    bpm: Mapped[Optional[str]] = mapped_column(TEXT())
+    bpm_base: Mapped[Optional[float]]
+    version: Mapped[Optional[str]] = mapped_column(TEXT())
+    date: Mapped[Optional[int]]
+
+
+class DifficultyLocalized(SongsBase):
+    __tablename__ = "difficulties_localized"
+
+    song_id: Mapped[str] = mapped_column(
+        ForeignKey("difficulties.song_id"), primary_key=True
+    )
+    rating_class: Mapped[str] = mapped_column(
+        ForeignKey("difficulties.rating_class"), primary_key=True
+    )
+    title_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    title_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+    artist_ja: Mapped[Optional[str]] = mapped_column(TEXT())
+    artist_ko: Mapped[Optional[str]] = mapped_column(TEXT())
+    artist_zh_hans: Mapped[Optional[str]] = mapped_column(TEXT())
+    artist_zh_hant: Mapped[Optional[str]] = mapped_column(TEXT())
+
+
+class ChartInfo(SongsBase):
+    __tablename__ = "charts_info"
+
+    song_id: Mapped[str] = mapped_column(
+        ForeignKey("difficulties.song_id"), primary_key=True
+    )
+    rating_class: Mapped[str] = mapped_column(
+        ForeignKey("difficulties.rating_class"), primary_key=True
+    )
+    constant: Mapped[int] = mapped_column(
+        comment="real_constant * 10. For example, Crimson Throne [FTR] is 10.4, then store 104."
+    )
+    notes: Mapped[Optional[int]]
+
+
+class SongsViewBase(DeclarativeBase, ReprHelper):
+    pass
+
+
+class Chart(SongsViewBase):
+    __tablename__ = "charts"
+
+    song_idx: Mapped[int]
+    song_id: Mapped[str]
+    rating_class: Mapped[int]
+    rating: Mapped[int]
+    rating_plus: Mapped[bool]
+    title: Mapped[str]
+    artist: Mapped[str]
+    set: Mapped[str]
+    bpm: Mapped[Optional[str]]
+    bpm_base: Mapped[Optional[float]]
+    audio_preview: Mapped[Optional[int]]
+    audio_preview_end: Mapped[Optional[int]]
+    side: Mapped[Optional[int]]
+    version: Mapped[Optional[str]]
+    date: Mapped[Optional[int]]
+    bg: Mapped[Optional[str]]
+    bg_inverse: Mapped[Optional[str]]
+    bg_day: Mapped[Optional[str]]
+    bg_night: Mapped[Optional[str]]
+    source: Mapped[Optional[str]]
+    source_copyright: Mapped[Optional[str]]
+    chart_designer: Mapped[Optional[str]]
+    jacket_desginer: Mapped[Optional[str]]
+    audio_override: Mapped[bool]
+    jacket_override: Mapped[bool]
+    jacket_night: Mapped[Optional[str]]
+    constant: Mapped[int]
+    notes: Mapped[Optional[int]]
+
+    __table__ = create_view(
+        name=__tablename__,
+        selectable=select(
+            Song.idx.label("song_idx"),
+            Difficulty.song_id,
+            Difficulty.rating_class,
+            Difficulty.rating,
+            Difficulty.rating_plus,
+            func.coalesce(Difficulty.title, Song.title).label("title"),
+            func.coalesce(Difficulty.artist, Song.artist).label("artist"),
+            Song.set,
+            func.coalesce(Difficulty.bpm, Song.bpm).label("bpm"),
+            func.coalesce(Difficulty.bpm_base, Song.bpm_base).label("bpm_base"),
+            Song.audio_preview,
+            Song.audio_preview_end,
+            Song.side,
+            func.coalesce(Difficulty.version, Song.version).label("version"),
+            func.coalesce(Difficulty.date, Song.date).label("date"),
+            func.coalesce(Difficulty.bg, Song.bg).label("bg"),
+            func.coalesce(Difficulty.bg_inverse, Song.bg_inverse).label("bg_inverse"),
+            Song.bg_day,
+            Song.bg_night,
+            Song.source,
+            Song.source_copyright,
+            Difficulty.chart_designer,
+            Difficulty.jacket_desginer,
+            Difficulty.audio_override,
+            Difficulty.jacket_override,
+            Difficulty.jacket_night,
+            ChartInfo.constant,
+            ChartInfo.notes,
+        )
+        .select_from(Difficulty)
+        .join(
+            ChartInfo,
+            (Difficulty.song_id == ChartInfo.song_id)
+            & (Difficulty.rating_class == ChartInfo.rating_class),
+        )
+        .join(Song, Difficulty.song_id == Song.id),
+        metadata=SongsViewBase.metadata,
+        cascade_on_drop=False,
+    )