feat(db): v4 to v5 migration

- Literally reverting f19ac4d8d5

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       - id: trailing-whitespace
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.8
+    rev: v0.11.12
     hooks:
       - id: ruff
         args: ["--fix"]

src/arcaea_offline/constants/enums/arcaea.py

@@ -13,6 +13,7 @@ class ArcaeaSongSide(IntEnum):
     LIGHT = 0
     CONFLICT = 1
     COLORLESS = 2
+    LEPHON = 3
 
 
 class ArcaeaPlayResultModifier(IntEnum):

src/arcaea_offline/database/migrations/README.md

@@ -0,0 +1 @@
+Generic single-database configuration.

src/arcaea_offline/database/migrations/env.py

@@ -0,0 +1,82 @@
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+from arcaea_offline.database.models._base import ModelBase
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# Interpret the config file for Python logging.
+# This line sets up loggers basically.
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+# add your model's MetaData object here
+# for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
+target_metadata = [ModelBase.metadata]
+
+# other values from the config, defined by the needs of env.py,
+# can be acquired:
+# my_important_option = config.get_main_option("my_important_option")
+# ... etc.
+
+
+def run_migrations_offline() -> None:
+    """Run migrations in 'offline' mode.
+
+    This configures the context with just a URL
+    and not an Engine, though an Engine is acceptable
+    here as well.  By skipping the Engine creation
+    we don't even need a DBAPI to be available.
+
+    Calls to context.execute() here emit the given string to the
+    script output.
+
+    """
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+        render_as_batch=True,
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    """Run migrations in 'online' mode.
+
+    In this scenario we need to create an Engine
+    and associate a connection with the context.
+
+    """
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+
+    with connectable.connect() as connection:
+        context.configure(
+            connection=connection,
+            target_metadata=target_metadata,
+            render_as_batch=True,
+            transaction_per_migration=True,
+        )
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

src/arcaea_offline/database/migrations/legacies/v5.py

@@ -0,0 +1,28 @@
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import DateTime
+from sqlalchemy.types import TypeDecorator
+
+
+class ForceTimezoneDateTime(TypeDecorator):
+    """
+    Store timezone aware timestamps as timezone naive UTC
+
+    https://docs.sqlalchemy.org/en/20/core/custom_types.html#store-timezone-aware-timestamps-as-timezone-naive-utc
+    """
+
+    impl = DateTime
+    cache_ok = True
+
+    def process_bind_param(self, value: Optional[datetime], dialect):
+        if value is not None:
+            if not value.tzinfo or value.tzinfo.utcoffset(value) is None:
+                raise TypeError("datetime tzinfo is required")
+            value = value.astimezone(timezone.utc).replace(tzinfo=None)
+        return value
+
+    def process_result_value(self, value: Optional[datetime], dialect):
+        if value is not None:
+            value = value.replace(tzinfo=timezone.utc)
+        return value

src/arcaea_offline/database/migrations/script.py.mako

@@ -0,0 +1,26 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from typing import Sequence, Union
+
+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)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

src/arcaea_offline/database/migrations/versions/0ca6733e40dc_v4_to_v5.py

@@ -0,0 +1,506 @@
+"""v4 to v5
+
+Revision ID: 0ca6733e40dc
+Revises: a3f9d48b7de3
+Create Date: 2025-05-31 11:38:25.575124
+
+"""
+
+from datetime import datetime, timezone
+from typing import Any, Sequence, Union
+from uuid import uuid4
+
+import sqlalchemy as sa
+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
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade(
+    *,
+    data_migration: bool = True,
+    data_migration_options: Any = None,
+) -> None:
+    op.create_table(
+        "property",
+        sa.Column("key", sa.String(), nullable=False),
+        sa.Column("value", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("key", name=op.f("pk_property")),
+    )
+
+    op.create_table(
+        "pack",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("section", sa.String(), nullable=True),
+        sa.Column(
+            "is_world_extend", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.Column("plus_character", sa.Integer(), nullable=True),
+        sa.Column("append_parent_id", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["append_parent_id"],
+            ["pack.id"],
+            name=op.f("fk_pack_append_parent_id_pack"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_pack")),
+    )
+    with op.batch_alter_table("pack", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_pack_name"), ["name"], unique=False)
+
+    op.create_table(
+        "pack_localization",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("lang", sa.String(), nullable=False),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["pack.id"],
+            name=op.f("fk_pack_localization_id_pack"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", "lang", name=op.f("pk_pack_localization")),
+    )
+    op.create_table(
+        "song",
+        sa.Column("pack_id", sa.String(), nullable=False),
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("idx", sa.Integer(), nullable=True),
+        sa.Column("title", sa.String(), nullable=True),
+        sa.Column("artist", sa.String(), nullable=True),
+        sa.Column(
+            "is_deleted", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.Column("added_at", ForceTimezoneDateTime(), nullable=False),
+        sa.Column("version", sa.String(), nullable=True),
+        sa.Column("bpm", sa.String(), nullable=True),
+        sa.Column("bpm_base", sa.Numeric(), nullable=True),
+        sa.Column(
+            "is_remote", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.Column(
+            "is_unlockable_in_world",
+            sa.Boolean(),
+            server_default=sa.text("0"),
+            nullable=False,
+        ),
+        sa.Column(
+            "is_beyond_unlock_state_local",
+            sa.Boolean(),
+            server_default=sa.text("0"),
+            nullable=False,
+        ),
+        sa.Column("purchase", sa.String(), nullable=True),
+        sa.Column("category", sa.String(), nullable=True),
+        sa.Column("side", sa.Integer(), nullable=True),
+        sa.Column("bg", sa.String(), nullable=True),
+        sa.Column("bg_inverse", sa.String(), nullable=True),
+        sa.Column("bg_day", sa.String(), nullable=True),
+        sa.Column("bg_night", sa.String(), nullable=True),
+        sa.Column("source", sa.String(), nullable=True),
+        sa.Column("source_copyright", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["pack_id"],
+            ["pack.id"],
+            name=op.f("fk_song_pack_id_pack"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_song")),
+    )
+    with op.batch_alter_table("song", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_song_added_at"), ["added_at"], unique=False
+        )
+        batch_op.create_index(batch_op.f("ix_song_artist"), ["artist"], unique=False)
+        batch_op.create_index(batch_op.f("ix_song_title"), ["title"], unique=False)
+
+    op.create_table(
+        "difficulty",
+        sa.Column("song_id", sa.String(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("rating", sa.Integer(), nullable=False),
+        sa.Column(
+            "is_rating_plus", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.Column("chart_designer", sa.String(), nullable=True),
+        sa.Column("jacket_designer", sa.String(), nullable=True),
+        sa.Column(
+            "has_overriding_audio",
+            sa.Boolean(),
+            server_default=sa.text("0"),
+            nullable=False,
+        ),
+        sa.Column(
+            "has_overriding_jacket",
+            sa.Boolean(),
+            server_default=sa.text("0"),
+            nullable=False,
+        ),
+        sa.Column("jacket_night", sa.String(), nullable=True),
+        sa.Column("title", sa.String(), nullable=True),
+        sa.Column("artist", sa.String(), nullable=True),
+        sa.Column("bg", sa.String(), nullable=True),
+        sa.Column("bg_inverse", sa.String(), nullable=True),
+        sa.Column("bpm", sa.String(), nullable=True),
+        sa.Column("bpm_base", sa.Numeric(), nullable=True),
+        sa.Column("added_at", ForceTimezoneDateTime(), nullable=True),
+        sa.Column("version", sa.String(), nullable=True),
+        sa.Column(
+            "is_legacy11", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ["song_id"],
+            ["song.id"],
+            name=op.f("fk_difficulty_song_id_song"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("song_id", "rating_class", name=op.f("pk_difficulty")),
+    )
+    op.create_table(
+        "song_localization",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("lang", sa.String(), nullable=False),
+        sa.Column("title", sa.String(), nullable=True),
+        sa.Column("source", sa.String(), nullable=True),
+        sa.Column(
+            "has_jacket", sa.Boolean(), server_default=sa.text("0"), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["song.id"],
+            name=op.f("fk_song_localization_id_song"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", "lang", name=op.f("pk_song_localization")),
+    )
+    op.create_table(
+        "chart_info",
+        sa.Column("song_id", sa.String(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("constant", sa.Numeric(), nullable=False),
+        sa.Column("notes", sa.Integer(), nullable=False),
+        sa.Column(
+            "added_at",
+            ForceTimezoneDateTime(),
+            nullable=False,
+        ),
+        sa.Column("version", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["song_id", "rating_class"],
+            ["difficulty.song_id", "difficulty.rating_class"],
+            name=op.f("fk_chart_info_song_id_difficulty"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint(
+            "song_id", "rating_class", "added_at", name=op.f("pk_chart_info")
+        ),
+    )
+    op.create_table(
+        "difficulty_localization",
+        sa.Column("song_id", sa.String(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("lang", sa.String(), nullable=False),
+        sa.Column("title", sa.String(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["song_id", "rating_class"],
+            ["difficulty.song_id", "difficulty.rating_class"],
+            name=op.f("fk_difficulty_localization_song_id_difficulty"),
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint(
+            "song_id", "rating_class", "lang", name=op.f("pk_difficulty_localization")
+        ),
+    )
+
+    op.drop_table("properties")
+    op.drop_table("packs")
+    op.drop_table("packs_localized")
+    op.drop_table("difficulties")
+    op.drop_table("songs")
+    op.drop_table("songs_localized")
+    op.drop_table("charts")
+    op.drop_table("charts_info")
+    op.drop_table("difficulties_localized")
+
+    op.rename_table("scores", "scores_old")
+    play_result_tbl = op.create_table(
+        "play_result",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("uuid", sa.Uuid(), nullable=False),
+        sa.Column("song_id", sa.String(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("played_at", ForceTimezoneDateTime(), nullable=True),
+        sa.Column("score", sa.Integer(), nullable=False),
+        sa.Column("pure", sa.Integer(), nullable=True),
+        sa.Column("pure_early", sa.Integer(), nullable=True),
+        sa.Column("pure_late", sa.Integer(), nullable=True),
+        sa.Column("far", sa.Integer(), nullable=True),
+        sa.Column("far_early", sa.Integer(), nullable=True),
+        sa.Column("far_late", sa.Integer(), nullable=True),
+        sa.Column("lost", sa.Integer(), nullable=True),
+        sa.Column("max_recall", sa.Integer(), nullable=True),
+        sa.Column("clear_type", sa.Integer(), nullable=True),
+        sa.Column("modifier", sa.Integer(), nullable=True),
+        sa.Column("comment", sa.Text(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_play_result")),
+        sa.UniqueConstraint("uuid", name=op.f("uq_play_result_uuid")),
+    )
+
+    if data_migration:
+        conn = op.get_bind()
+        query = conn.execute(
+            sa.text(
+                "SELECT id, song_id, rating_class, score, pure, far, lost, "
+                " `date`, max_recall, modifier, clear_type, comment "
+                "FROM scores_old"
+            )
+        )
+        batch_size = 30
+
+        while True:
+            rows = query.fetchmany(batch_size)
+            if not rows:
+                break
+
+            rows_to_insert = []
+            for row in rows:
+                result = row._asdict()
+
+                date = result.pop("date")
+                result["uuid"] = uuid4()
+                result["played_at"] = (
+                    datetime.fromtimestamp(date, tz=timezone.utc)
+                    if date is not None
+                    else None
+                )
+                rows_to_insert.append(result)
+
+            conn.execute(sa.insert(play_result_tbl), rows_to_insert)
+
+    op.drop_table("scores_old")
+
+
+def downgrade() -> None:
+    raise NotImplementedError(
+        f"Downgrade not supported! ({context.get_context().get_current_revision()})"
+    )
+
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "difficulties_localized",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.INTEGER(), nullable=False),
+        sa.Column("title_ja", sa.TEXT(), nullable=True),
+        sa.Column("title_ko", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("artist_ja", sa.TEXT(), nullable=True),
+        sa.Column("artist_ko", sa.TEXT(), nullable=True),
+        sa.Column("artist_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("artist_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["rating_class"],
+            ["difficulties.rating_class"],
+            name=op.f("fk_difficulties_localized_rating_class_difficulties"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["song_id"],
+            ["difficulties.song_id"],
+            name=op.f("fk_difficulties_localized_song_id_difficulties"),
+        ),
+        sa.PrimaryKeyConstraint(
+            "song_id", "rating_class", name=op.f("pk_difficulties_localized")
+        ),
+    )
+    op.create_table(
+        "charts_info",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.INTEGER(), nullable=False),
+        sa.Column("constant", sa.INTEGER(), nullable=False),
+        sa.Column("notes", sa.INTEGER(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["rating_class"],
+            ["difficulties.rating_class"],
+            name=op.f("fk_charts_info_rating_class_difficulties"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["song_id"],
+            ["difficulties.song_id"],
+            name=op.f("fk_charts_info_song_id_difficulties"),
+        ),
+        sa.PrimaryKeyConstraint("song_id", "rating_class", name=op.f("pk_charts_info")),
+    )
+    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(), nullable=True),
+        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(), nullable=True),
+        sa.Column("side", sa.INTEGER(), nullable=False),
+        sa.Column("world_unlock", sa.BOOLEAN(), nullable=False),
+        sa.Column("remote_download", sa.BOOLEAN(), nullable=True),
+        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(), nullable=True),
+        sa.Column("jacket_designer", sa.TEXT(), nullable=True),
+        sa.Column("jacket_override", sa.BOOLEAN(), nullable=False),
+        sa.Column("audio_override", sa.BOOLEAN(), nullable=False),
+        sa.PrimaryKeyConstraint("song_id", "rating_class"),
+    )
+    op.create_table(
+        "songs_localized",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("title_ja", sa.TEXT(), nullable=True),
+        sa.Column("title_ko", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("search_title_ja", sa.TEXT(), nullable=True),
+        sa.Column("search_title_ko", sa.TEXT(), nullable=True),
+        sa.Column("search_title_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("search_title_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("search_artist_ja", sa.TEXT(), nullable=True),
+        sa.Column("search_artist_ko", sa.TEXT(), nullable=True),
+        sa.Column("search_artist_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("search_artist_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("source_ja", sa.TEXT(), nullable=True),
+        sa.Column("source_ko", sa.TEXT(), nullable=True),
+        sa.Column("source_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("source_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["id"], ["songs.id"], name=op.f("fk_songs_localized_id_songs")
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_songs_localized")),
+    )
+    op.create_table(
+        "packs",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("name", sa.TEXT(), nullable=False),
+        sa.Column("description", sa.TEXT(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name="fk_packs"),
+    )
+    op.create_table(
+        "properties",
+        sa.Column("key", sa.TEXT(), nullable=False),
+        sa.Column("value", sa.TEXT(), nullable=False),
+        sa.UniqueConstraint("key"),
+    )
+    op.create_table(
+        "songs",
+        sa.Column("idx", sa.INTEGER(), nullable=False),
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("title", sa.TEXT(), nullable=False),
+        sa.Column("artist", sa.TEXT(), nullable=False),
+        sa.Column("set", sa.TEXT(), nullable=False),
+        sa.Column("bpm", sa.TEXT(), nullable=True),
+        sa.Column("bpm_base", sa.FLOAT(), nullable=True),
+        sa.Column("audio_preview", sa.INTEGER(), nullable=True),
+        sa.Column("audio_preview_end", sa.INTEGER(), nullable=True),
+        sa.Column("side", sa.INTEGER(), nullable=True),
+        sa.Column("version", sa.TEXT(), nullable=True),
+        sa.Column("date", sa.INTEGER(), nullable=True),
+        sa.Column("bg", sa.TEXT(), nullable=True),
+        sa.Column("bg_inverse", sa.TEXT(), nullable=True),
+        sa.Column("bg_day", sa.TEXT(), nullable=True),
+        sa.Column("bg_night", sa.TEXT(), nullable=True),
+        sa.Column("source", sa.TEXT(), nullable=True),
+        sa.Column("source_copyright", sa.TEXT(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name=op.f("songs")),
+    )
+    op.create_table(
+        "difficulties",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.INTEGER(), nullable=False),
+        sa.Column("rating", sa.INTEGER(), nullable=False),
+        sa.Column("rating_plus", sa.BOOLEAN(), nullable=False),
+        sa.Column("chart_designer", sa.TEXT(), nullable=True),
+        sa.Column("jacket_desginer", sa.TEXT(), nullable=True),
+        sa.Column("audio_override", sa.BOOLEAN(), nullable=False),
+        sa.Column("jacket_override", sa.BOOLEAN(), nullable=False),
+        sa.Column("jacket_night", sa.TEXT(), nullable=True),
+        sa.Column("title", sa.TEXT(), nullable=True),
+        sa.Column("artist", sa.TEXT(), nullable=True),
+        sa.Column("bg", sa.TEXT(), nullable=True),
+        sa.Column("bg_inverse", sa.TEXT(), nullable=True),
+        sa.Column("bpm", sa.TEXT(), nullable=True),
+        sa.Column("bpm_base", sa.FLOAT(), nullable=True),
+        sa.Column("version", sa.TEXT(), nullable=True),
+        sa.Column("date", sa.INTEGER(), nullable=True),
+        sa.PrimaryKeyConstraint(
+            "song_id", "rating_class", name=op.f("pk_difficulties")
+        ),
+    )
+    op.create_table(
+        "packs_localized",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("name_ja", sa.TEXT(), nullable=True),
+        sa.Column("name_ko", sa.TEXT(), nullable=True),
+        sa.Column("name_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("name_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("description_ja", sa.TEXT(), nullable=True),
+        sa.Column("description_ko", sa.TEXT(), nullable=True),
+        sa.Column("description_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("description_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["id"], ["packs.id"], name=op.f("fk_packs_localized_id_packs")
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_packs_localized")),
+    )
+    op.create_table(
+        "scores",
+        sa.Column("id", sa.INTEGER(), nullable=False),
+        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(), nullable=True),
+        sa.Column("far", sa.INTEGER(), nullable=True),
+        sa.Column("lost", sa.INTEGER(), nullable=True),
+        sa.Column("date", sa.INTEGER(), nullable=True),
+        sa.Column("max_recall", sa.INTEGER(), nullable=True),
+        sa.Column("modifier", sa.INTEGER(), nullable=True),
+        sa.Column("clear_type", sa.INTEGER(), nullable=True),
+        sa.Column("comment", sa.TEXT(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.drop_table("difficulty_localization")
+    op.drop_table("chart_info")
+    op.drop_table("song_localization")
+    op.drop_table("difficulty")
+    with op.batch_alter_table("song", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_song_title"))
+        batch_op.drop_index(batch_op.f("ix_song_artist"))
+        batch_op.drop_index(batch_op.f("ix_song_added_at"))
+
+    op.drop_table("song")
+    op.drop_table("pack_localization")
+    op.drop_table("property")
+    op.drop_table("play_result")
+    with op.batch_alter_table("pack", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_pack_name"))
+
+    op.drop_table("pack")
+    # ### end Alembic commands ###

src/arcaea_offline/database/migrations/versions/a3f9d48b7de3_v1_to_v4.py

@@ -0,0 +1,275 @@
+"""v1 to v4
+
+Revision ID: a3f9d48b7de3
+Revises:
+Create Date: 2024-11-24 00:03:07.697165
+
+"""
+
+from datetime import datetime, timezone
+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
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+class V4DataMigrationOptions(TypedDict):
+    threshold_date: Optional[datetime]
+
+
+def _data_migration_options(user_input: Optional[Mapping]):
+    options: V4DataMigrationOptions = {
+        "threshold_date": datetime(year=2017, month=1, day=23, tzinfo=timezone.utc),
+    }
+
+    if user_input is None:
+        return options
+
+    if not isinstance(user_input, dict):
+        raise TypeError("v4 migration: data migration options should be a dict object")
+
+    threshold_date = user_input.get("threshold_date")
+    if threshold_date is not None and not isinstance(threshold_date, datetime):
+        raise ValueError(
+            "v4 migration: threshold_date should be None or a datetime.datetime object"
+        )
+    options["threshold_date"] = threshold_date
+
+    return options
+
+
+def upgrade(
+    *,
+    data_migration: bool = True,
+    data_migration_options: Optional[V4DataMigrationOptions] = None,
+) -> None:
+    data_migration_options = _data_migration_options(data_migration_options)
+    threshold_date = data_migration_options["threshold_date"]
+
+    op.create_table(
+        "difficulties",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("rating", sa.Integer(), nullable=False),
+        sa.Column("rating_plus", sa.Boolean(), nullable=False),
+        sa.Column("chart_designer", sa.TEXT(), nullable=True),
+        sa.Column("jacket_desginer", sa.TEXT(), nullable=True),
+        sa.Column("audio_override", sa.Boolean(), nullable=False),
+        sa.Column("jacket_override", sa.Boolean(), nullable=False),
+        sa.Column("jacket_night", sa.TEXT(), nullable=True),
+        sa.Column("title", sa.TEXT(), nullable=True),
+        sa.Column("artist", sa.TEXT(), nullable=True),
+        sa.Column("bg", sa.TEXT(), nullable=True),
+        sa.Column("bg_inverse", sa.TEXT(), nullable=True),
+        sa.Column("bpm", sa.TEXT(), nullable=True),
+        sa.Column("bpm_base", sa.Float(), nullable=True),
+        sa.Column("version", sa.TEXT(), nullable=True),
+        sa.Column("date", sa.Integer(), nullable=True),
+        sa.PrimaryKeyConstraint("song_id", "rating_class", name="pk_difficulties"),
+    )
+    op.create_table(
+        "packs",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("name", sa.TEXT(), nullable=False),
+        sa.Column("description", sa.TEXT(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name="fk_packs"),
+    )
+    op.create_table(
+        "songs",
+        sa.Column("idx", sa.Integer(), nullable=False),
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("title", sa.TEXT(), nullable=False),
+        sa.Column("artist", sa.TEXT(), nullable=False),
+        sa.Column("set", sa.TEXT(), nullable=False),
+        sa.Column("bpm", sa.TEXT(), nullable=True),
+        sa.Column("bpm_base", sa.Float(), nullable=True),
+        sa.Column("audio_preview", sa.Integer(), nullable=True),
+        sa.Column("audio_preview_end", sa.Integer(), nullable=True),
+        sa.Column("side", sa.Integer(), nullable=True),
+        sa.Column("version", sa.TEXT(), nullable=True),
+        sa.Column("date", sa.Integer(), nullable=True),
+        sa.Column("bg", sa.TEXT(), nullable=True),
+        sa.Column("bg_inverse", sa.TEXT(), nullable=True),
+        sa.Column("bg_day", sa.TEXT(), nullable=True),
+        sa.Column("bg_night", sa.TEXT(), nullable=True),
+        sa.Column("source", sa.TEXT(), nullable=True),
+        sa.Column("source_copyright", sa.TEXT(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name="songs"),
+    )
+    op.create_table(
+        "charts_info",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column(
+            "constant",
+            sa.Integer(),
+            nullable=False,
+            comment="real_constant * 10. For example, Crimson Throne [FTR] is 10.4, then store 104.",
+        ),
+        sa.Column("notes", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["rating_class"],
+            ["difficulties.rating_class"],
+            name="fk_charts_info_rating_class_difficulties",
+        ),
+        sa.ForeignKeyConstraint(
+            ["song_id"],
+            ["difficulties.song_id"],
+            name="fk_charts_info_song_id_difficulties",
+        ),
+        sa.PrimaryKeyConstraint("song_id", "rating_class", name="pk_charts_info"),
+    )
+    op.create_table(
+        "difficulties_localized",
+        sa.Column("song_id", sa.TEXT(), nullable=False),
+        sa.Column("rating_class", sa.Integer(), nullable=False),
+        sa.Column("title_ja", sa.TEXT(), nullable=True),
+        sa.Column("title_ko", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("artist_ja", sa.TEXT(), nullable=True),
+        sa.Column("artist_ko", sa.TEXT(), nullable=True),
+        sa.Column("artist_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("artist_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["rating_class"],
+            ["difficulties.rating_class"],
+            name="fk_difficulties_localized_rating_class_difficulties",
+        ),
+        sa.ForeignKeyConstraint(
+            ["song_id"],
+            ["difficulties.song_id"],
+            name="fk_difficulties_localized_song_id_difficulties",
+        ),
+        sa.PrimaryKeyConstraint(
+            "song_id", "rating_class", name="pk_difficulties_localized"
+        ),
+    )
+    op.create_table(
+        "packs_localized",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("name_ja", sa.TEXT(), nullable=True),
+        sa.Column("name_ko", sa.TEXT(), nullable=True),
+        sa.Column("name_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("name_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("description_ja", sa.TEXT(), nullable=True),
+        sa.Column("description_ko", sa.TEXT(), nullable=True),
+        sa.Column("description_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("description_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["packs.id"],
+            name="fk_packs_localized_id_packs",
+        ),
+        sa.PrimaryKeyConstraint("id", name="pk_packs_localized"),
+    )
+    op.create_table(
+        "songs_localized",
+        sa.Column("id", sa.TEXT(), nullable=False),
+        sa.Column("title_ja", sa.TEXT(), nullable=True),
+        sa.Column("title_ko", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("title_zh_hant", sa.TEXT(), nullable=True),
+        sa.Column("search_title_ja", sa.TEXT(), nullable=True, comment="JSON array"),
+        sa.Column("search_title_ko", sa.TEXT(), nullable=True, comment="JSON array"),
+        sa.Column(
+            "search_title_zh_hans", sa.TEXT(), nullable=True, comment="JSON array"
+        ),
+        sa.Column(
+            "search_title_zh_hant", sa.TEXT(), nullable=True, comment="JSON array"
+        ),
+        sa.Column("search_artist_ja", sa.TEXT(), nullable=True, comment="JSON array"),
+        sa.Column("search_artist_ko", sa.TEXT(), nullable=True, comment="JSON array"),
+        sa.Column(
+            "search_artist_zh_hans", sa.TEXT(), nullable=True, comment="JSON array"
+        ),
+        sa.Column(
+            "search_artist_zh_hant", sa.TEXT(), nullable=True, comment="JSON array"
+        ),
+        sa.Column("source_ja", sa.TEXT(), nullable=True),
+        sa.Column("source_ko", sa.TEXT(), nullable=True),
+        sa.Column("source_zh_hans", sa.TEXT(), nullable=True),
+        sa.Column("source_zh_hant", sa.TEXT(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["songs.id"],
+            name="fk_songs_localized_id_songs",
+        ),
+        sa.PrimaryKeyConstraint("id", name="pk_songs_localized"),
+    )
+    op.drop_table("aliases")
+    op.drop_table("packages")
+    op.execute(sa.text("DROP VIEW IF EXISTS bests"))
+    op.execute(sa.text("DROP VIEW IF EXISTS calculated"))
+    op.execute(sa.text("DROP VIEW IF EXISTS calculated_potential"))
+    op.execute(sa.text("DROP VIEW IF EXISTS song_id_names"))
+
+    op.rename_table("scores", "scores_old")
+    scores_tbl = op.create_table(
+        "scores",
+        sa.Column("id", sa.Integer(), autoincrement=True, primary_key=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("date", sa.Integer()),
+        sa.Column("max_recall", sa.Integer()),
+        sa.Column("modifier", sa.Integer(), comment="0: NORMAL, 1: EASY, 2: HARD"),
+        sa.Column(
+            "clear_type",
+            sa.Integer(),
+            comment="0: TRACK LOST, 1: NORMAL CLEAR, 2: FULL RECALL, "
+            "3: PURE MEMORY, 4: EASY CLEAR, 5: HARD CLEAR",
+        ),
+        sa.Column("comment", sa.TEXT()),
+    )
+    if data_migration:
+        conn = op.get_bind()
+        query = conn.execute(
+            sa.text(
+                "SELECT id, song_id, rating_class, score, time, pure, far, lost, max_recall, clear_type "
+                "FROM scores_old"
+            )
+        )
+        batch_size = 30
+
+        while True:
+            rows = query.fetchmany(batch_size)
+            if not rows:
+                break
+
+            rows_to_insert = []
+
+            for row in rows:
+                result = row._asdict()
+                result["date"] = datetime.fromtimestamp(
+                    result.pop("time"), tz=timezone.utc
+                )
+
+                if threshold_date is not None and result["date"] <= threshold_date:
+                    result["date"] = None
+
+                result["date"] = (
+                    int(result["date"].timestamp())
+                    if result["date"] is not None
+                    else None
+                )
+                rows_to_insert.append(result)
+
+            conn.execute(sa.insert(scores_tbl), rows_to_insert)
+
+    op.drop_table("scores_old")
+
+
+def downgrade() -> None:
+    raise NotImplementedError(
+        f"Downgrade not supported! ({context.get_context().get_current_revision()})"
+    )

src/arcaea_offline/database/models/__init__.py

@@ -1,38 +1,32 @@
-from .arcaea import (
-    Chart,
-    ChartInfo,
-    Difficulty,
-    DifficultyLocalized,
-    Pack,
-    PackLocalized,
-    Song,
-    SongLocalized,
-    SongSearchWord,
-)
-from .base import ModelsV5Base, ModelsV5ViewBase
+from ._base import ModelBase, ModelViewBase
+from .chart_info import ChartInfo
 from .config import Property
-from .play_results import (
+from .difficulty import Difficulty, DifficultyLocalization
+from .pack import Pack, PackLocalization
+from .song import Song, SongLocalization
+
+from .chart import Chart  # isort: skip
+from .play_result import (
     CalculatedPotential,
     PlayResult,
     PlayResultBest,
     PlayResultCalculated,
-)
+)  # isort: skip
 
 __all__ = [
     "CalculatedPotential",
     "Chart",
     "ChartInfo",
     "Difficulty",
-    "DifficultyLocalized",
-    "ModelsV5Base",
-    "ModelsV5ViewBase",
+    "DifficultyLocalization",
+    "ModelBase",
+    "ModelViewBase",
     "Pack",
-    "PackLocalized",
+    "PackLocalization",
     "PlayResult",
     "PlayResultBest",
     "PlayResultCalculated",
     "Property",
     "Song",
-    "SongLocalized",
-    "SongSearchWord",
+    "SongLocalization",
 ]

src/arcaea_offline/database/models/_base.py

@@ -1,31 +1,30 @@
 from datetime import datetime
 
+from sqlalchemy import MetaData
 from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.orm.exc import DetachedInstanceError
 
-from arcaea_offline.constants.enums import (
-    ArcaeaPlayResultClearType,
-    ArcaeaPlayResultModifier,
-    ArcaeaRatingClass,
-    ArcaeaSongSide,
-)
-
-from ._custom_types import DbIntEnum, TZDateTime
+from ._types import ForceTimezoneDateTime
 
 TYPE_ANNOTATION_MAP = {
-    datetime: TZDateTime,
-    ArcaeaRatingClass: DbIntEnum(ArcaeaRatingClass),
-    ArcaeaSongSide: DbIntEnum(ArcaeaSongSide),
-    ArcaeaPlayResultClearType: DbIntEnum(ArcaeaPlayResultClearType),
-    ArcaeaPlayResultModifier: DbIntEnum(ArcaeaPlayResultModifier),
+    datetime: ForceTimezoneDateTime,
 }
 
 
-class ModelsV5Base(DeclarativeBase):
+class ModelBase(DeclarativeBase):
     type_annotation_map = TYPE_ANNOTATION_MAP
+    metadata = MetaData(
+        naming_convention={
+            "ix": "ix_%(column_0_label)s",
+            "uq": "uq_%(table_name)s_%(column_0_name)s",
+            "ck": "ck_%(table_name)s_`%(constraint_name)s`",
+            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+            "pk": "pk_%(table_name)s",
+        }
+    )
 
 
-class ModelsV5ViewBase(DeclarativeBase):
+class ModelViewBase(DeclarativeBase):
     type_annotation_map = TYPE_ANNOTATION_MAP
 
 
@@ -34,7 +33,7 @@ class ReprHelper:
 
     def _repr(self, **kwargs) -> str:
         """
-        Helper for __repr__
+        SQLAlchemy model __repr__ helper
 
         https://stackoverflow.com/a/55749579/16484891
 

src/arcaea_offline/database/models/_custom_types.py

@@ -1,46 +0,0 @@
-from datetime import datetime, timezone
-from enum import IntEnum
-from typing import Optional, Type
-
-from sqlalchemy import DateTime, Integer
-from sqlalchemy.types import TypeDecorator
-
-
-class DbIntEnum(TypeDecorator):
-    """sqlalchemy `TypeDecorator` for `IntEnum`s"""
-
-    impl = Integer
-    cache_ok = True
-
-    def __init__(self, enum_class: Type[IntEnum]):
-        super().__init__()
-        self.enum_class = enum_class
-
-    def process_bind_param(self, value: Optional[IntEnum], dialect) -> Optional[int]:
-        return None if value is None else value.value
-
-    def process_result_value(self, value: Optional[int], dialect) -> Optional[IntEnum]:
-        return None if value is None else self.enum_class(value)
-
-
-class TZDateTime(TypeDecorator):
-    """
-    Store Timezone Aware Timestamps as Timezone Naive UTC
-
-    https://docs.sqlalchemy.org/en/20/core/custom_types.html#store-timezone-aware-timestamps-as-timezone-naive-utc
-    """
-
-    impl = DateTime
-    cache_ok = True
-
-    def process_bind_param(self, value: Optional[datetime], dialect):
-        if value is not None:
-            if not value.tzinfo or value.tzinfo.utcoffset(value) is None:
-                raise TypeError("tzinfo is required")
-            value = value.astimezone(timezone.utc).replace(tzinfo=None)
-        return value
-
-    def process_result_value(self, value: Optional[datetime], dialect):
-        if value is not None:
-            value = value.replace(tzinfo=timezone.utc)
-        return value

src/arcaea_offline/database/models/_types.py

@@ -0,0 +1,28 @@
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import DateTime
+from sqlalchemy.types import TypeDecorator
+
+
+class ForceTimezoneDateTime(TypeDecorator):
+    """
+    Store timezone aware timestamps as timezone naive UTC
+
+    https://docs.sqlalchemy.org/en/20/core/custom_types.html#store-timezone-aware-timestamps-as-timezone-naive-utc
+    """
+
+    impl = DateTime
+    cache_ok = True
+
+    def process_bind_param(self, value: Optional[datetime], dialect):
+        if value is not None:
+            if not value.tzinfo or value.tzinfo.utcoffset(value) is None:
+                raise TypeError("datetime tzinfo is required")
+            value = value.astimezone(timezone.utc).replace(tzinfo=None)
+        return value
+
+    def process_result_value(self, value: Optional[datetime], dialect):
+        if value is not None:
+            value = value.replace(tzinfo=timezone.utc)
+        return value

src/arcaea_offline/database/models/arcaea.py

@@ -1,285 +0,0 @@
-from datetime import datetime
-from typing import List, Optional
-
-from sqlalchemy import Enum, ForeignKey, and_, func, select
-from sqlalchemy.orm import Mapped, mapped_column, relationship
-from sqlalchemy_utils import create_view
-
-from arcaea_offline.constants.enums.arcaea import (
-    ArcaeaLanguage,
-    ArcaeaRatingClass,
-    ArcaeaSongSide,
-)
-
-from .base import ModelsV5Base, ModelsV5ViewBase, ReprHelper
-
-__all__ = [
-    "Chart",
-    "ChartInfo",
-    "Difficulty",
-    "DifficultyLocalized",
-    "Pack",
-    "PackLocalized",
-    "Song",
-    "SongLocalized",
-]
-
-
-_ArcaeaLanguageEnumType = Enum(
-    ArcaeaLanguage, native_enum=False, values_callable=lambda e: [x.value for x in e]
-)
-
-
-class Pack(ModelsV5Base, ReprHelper):
-    __tablename__ = "packs"
-
-    id: Mapped[str] = mapped_column(primary_key=True)
-    name: Mapped[str]
-    description: Mapped[Optional[str]]
-
-    songs: Mapped[List["Song"]] = relationship(back_populates="pack", viewonly=True)
-    localized_objects: Mapped[List["PackLocalized"]] = relationship(
-        back_populates="parent", viewonly=True
-    )
-
-
-class PackLocalized(ModelsV5Base, ReprHelper):
-    __tablename__ = "packs_localized"
-
-    pkid: Mapped[int] = mapped_column(primary_key=True)
-    id: Mapped[str] = mapped_column(
-        ForeignKey(Pack.id, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    lang: Mapped[ArcaeaLanguage] = mapped_column(_ArcaeaLanguageEnumType)
-    name: Mapped[Optional[str]]
-    description: Mapped[Optional[str]]
-
-    parent: Mapped[Pack] = relationship(viewonly=True)
-
-
-class Song(ModelsV5Base, ReprHelper):
-    __tablename__ = "songs"
-
-    idx: Mapped[int]
-    id: Mapped[str] = mapped_column(primary_key=True)
-    title: Mapped[str]
-    artist: Mapped[str]
-    pack_id: Mapped[str] = mapped_column(
-        ForeignKey(Pack.id, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    bpm: Mapped[Optional[str]]
-    bpm_base: Mapped[Optional[float]]
-    audio_preview: Mapped[Optional[int]]
-    audio_preview_end: Mapped[Optional[int]]
-    side: Mapped[Optional[ArcaeaSongSide]]
-    version: Mapped[Optional[str]]
-    date: Mapped[Optional[datetime]]
-    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]]
-
-    pack: Mapped[Pack] = relationship(viewonly=True)
-    difficulties: Mapped[List["Difficulty"]] = relationship(
-        back_populates="song", viewonly=True
-    )
-    localized_objects: Mapped[List["SongLocalized"]] = relationship(
-        back_populates="parent", viewonly=True
-    )
-
-    @property
-    def charts_info(self):
-        return [d.chart_info for d in self.difficulties if d.chart_info is not None]
-
-
-class SongLocalized(ModelsV5Base, ReprHelper):
-    __tablename__ = "songs_localized"
-
-    pkid: Mapped[int] = mapped_column(primary_key=True)
-    id: Mapped[str] = mapped_column(
-        ForeignKey(Song.id, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    lang: Mapped[ArcaeaLanguage] = mapped_column(_ArcaeaLanguageEnumType)
-    title: Mapped[Optional[str]]
-    source: Mapped[Optional[str]]
-
-    parent: Mapped[Song] = relationship(
-        back_populates="localized_objects", viewonly=True
-    )
-
-
-class SongSearchWord(ModelsV5Base, ReprHelper):
-    __tablename__ = "songs_search_words"
-
-    pkid: Mapped[int] = mapped_column(primary_key=True)
-    id: Mapped[str] = mapped_column(
-        ForeignKey(Song.id, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    lang: Mapped[ArcaeaLanguage] = mapped_column(_ArcaeaLanguageEnumType)
-    type: Mapped[int] = mapped_column(comment="1: title, 2: artist")
-    value: Mapped[str]
-
-
-class Difficulty(ModelsV5Base, ReprHelper):
-    __tablename__ = "difficulties"
-
-    song_id: Mapped[str] = mapped_column(
-        ForeignKey(Song.id, onupdate="CASCADE", ondelete="NO ACTION"),
-        primary_key=True,
-    )
-    rating_class: Mapped[ArcaeaRatingClass] = mapped_column(primary_key=True)
-    rating: Mapped[int]
-    rating_plus: Mapped[bool]
-    chart_designer: Mapped[Optional[str]]
-    jacket_desginer: Mapped[Optional[str]]
-    audio_override: Mapped[bool] = mapped_column(default=False)
-    jacket_override: Mapped[bool] = mapped_column(default=False)
-    jacket_night: Mapped[Optional[str]]
-    title: Mapped[Optional[str]]
-    artist: Mapped[Optional[str]]
-    bg: Mapped[Optional[str]]
-    bg_inverse: Mapped[Optional[str]]
-    bpm: Mapped[Optional[str]]
-    bpm_base: Mapped[Optional[float]]
-    version: Mapped[Optional[str]]
-    date: Mapped[Optional[datetime]]
-
-    song: Mapped[Song] = relationship(back_populates="difficulties", viewonly=True)
-    chart_info: Mapped[Optional["ChartInfo"]] = relationship(
-        primaryjoin=(
-            "and_(Difficulty.song_id==ChartInfo.song_id, "
-            "Difficulty.rating_class==ChartInfo.rating_class)"
-        ),
-        viewonly=True,
-    )
-    localized_objects: Mapped[List["DifficultyLocalized"]] = relationship(
-        primaryjoin=(
-            "and_(Difficulty.song_id==DifficultyLocalized.song_id, "
-            "Difficulty.rating_class==DifficultyLocalized.rating_class)"
-        ),
-        viewonly=True,
-    )
-
-
-class DifficultyLocalized(ModelsV5Base, ReprHelper):
-    __tablename__ = "difficulties_localized"
-
-    pkid: Mapped[int] = mapped_column(primary_key=True)
-    song_id: Mapped[str] = mapped_column(
-        ForeignKey(Difficulty.song_id, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    rating_class: Mapped[ArcaeaRatingClass] = mapped_column(
-        ForeignKey(Difficulty.rating_class, onupdate="CASCADE", ondelete="NO ACTION")
-    )
-    lang: Mapped[ArcaeaLanguage] = mapped_column(_ArcaeaLanguageEnumType)
-    title: Mapped[Optional[str]]
-    artist: Mapped[Optional[str]]
-
-    parent: Mapped[Difficulty] = relationship(
-        primaryjoin=and_(
-            Difficulty.song_id == song_id, Difficulty.rating_class == rating_class
-        ),
-        viewonly=True,
-    )
-
-
-class ChartInfo(ModelsV5Base, ReprHelper):
-    __tablename__ = "charts_info"
-
-    song_id: Mapped[str] = mapped_column(
-        ForeignKey(Difficulty.song_id, onupdate="CASCADE", ondelete="NO ACTION"),
-        primary_key=True,
-    )
-    rating_class: Mapped[ArcaeaRatingClass] = mapped_column(
-        ForeignKey(Difficulty.rating_class, onupdate="CASCADE", ondelete="NO ACTION"),
-        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]]
-
-    difficulty: Mapped[Difficulty] = relationship(
-        primaryjoin=and_(
-            Difficulty.song_id == song_id, Difficulty.rating_class == rating_class
-        ),
-        viewonly=True,
-    )
-
-
-class Chart(ModelsV5ViewBase, ReprHelper):
-    __tablename__ = "charts"
-
-    song_idx: Mapped[int]
-    song_id: Mapped[str]
-    rating_class: Mapped[ArcaeaRatingClass]
-    rating: Mapped[int]
-    rating_plus: Mapped[bool]
-    title: Mapped[str]
-    artist: Mapped[str]
-    pack_id: 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[datetime]]
-    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.pack_id,
-            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=ModelsV5ViewBase.metadata,
-        cascade_on_drop=False,
-    )

src/arcaea_offline/database/models/chart.py

@@ -0,0 +1,85 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Mapped
+from sqlalchemy_utils import create_view
+
+from ._base import ModelBase, ModelViewBase, ReprHelper
+from .chart_info import ChartInfo
+from .difficulty import Difficulty
+from .song import Song
+
+
+class Chart(ModelBase, ReprHelper):
+    __tablename__ = "charts"
+
+    song_idx: Mapped[int]
+    song_id: Mapped[str]
+    rating_class: Mapped[int]
+    rating: Mapped[int]
+    is_rating_plus: Mapped[bool]
+    title: Mapped[str]
+    artist: Mapped[str]
+    pack_id: 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]]
+    added_at: Mapped[Optional[datetime]]
+    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]]
+    has_overriding_audio: Mapped[bool]
+    has_overriding_jacket: 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.is_rating_plus,
+            func.coalesce(Difficulty.title, Song.title).label("title"),
+            func.coalesce(Difficulty.artist, Song.artist).label("artist"),
+            Song.pack_id,
+            func.coalesce(Difficulty.bpm, Song.bpm).label("bpm"),
+            func.coalesce(Difficulty.bpm_base, Song.bpm_base).label("bpm_base"),
+            Song.side,
+            func.coalesce(Difficulty.version, Song.version).label("version"),
+            func.coalesce(Difficulty.added_at, Song.added_at).label("added_at"),
+            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_designer,
+            Difficulty.has_overriding_audio,
+            Difficulty.has_overriding_jacket,
+            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=ModelViewBase.metadata,
+        cascade_on_drop=False,
+    )

src/arcaea_offline/database/models/chart_info.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+from decimal import Decimal
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import ForeignKeyConstraint, Integer, Numeric, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from ._base import ModelBase, ReprHelper
+from ._types import ForceTimezoneDateTime
+
+if TYPE_CHECKING:
+    from .difficulty import Difficulty
+
+
+class ChartInfo(ModelBase, ReprHelper):
+    __tablename__ = "chart_info"
+    __table_args__ = (
+        ForeignKeyConstraint(
+            ["song_id", "rating_class"],
+            ["difficulty.song_id", "difficulty.rating_class"],
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+    )
+
+    difficulty: Mapped["Difficulty"] = relationship(back_populates="chart_info_list")
+
+    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)
+    added_at: Mapped[datetime] = mapped_column(ForceTimezoneDateTime, primary_key=True)
+    version: Mapped[Optional[str]] = mapped_column(String)

src/arcaea_offline/database/models/config.py

@@ -1,12 +1,12 @@
 from sqlalchemy.orm import Mapped, mapped_column
 
-from .base import ModelsV5Base, ReprHelper
+from ._base import ModelBase, ReprHelper
 
 __all__ = ["Property"]
 
 
-class Property(ModelsV5Base, ReprHelper):
-    __tablename__ = "properties"
+class Property(ModelBase, ReprHelper):
+    __tablename__ = "property"
 
     key: Mapped[str] = mapped_column(primary_key=True)
     value: Mapped[str] = mapped_column()

src/arcaea_offline/database/models/difficulty.py

@@ -0,0 +1,92 @@
+from datetime import datetime
+from decimal import Decimal
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import (
+    Boolean,
+    ForeignKey,
+    ForeignKeyConstraint,
+    Integer,
+    Numeric,
+    String,
+    text,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from ._base import ModelBase, ReprHelper
+from ._types import ForceTimezoneDateTime
+
+if TYPE_CHECKING:
+    from .chart_info import ChartInfo
+    from .song import Song
+
+
+class Difficulty(ModelBase, ReprHelper):
+    __tablename__ = "difficulty"
+
+    song_id: Mapped[str] = mapped_column(
+        ForeignKey("song.id", onupdate="CASCADE", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    song: Mapped["Song"] = relationship(back_populates="difficulties")
+    localization_entries: Mapped[list["DifficultyLocalization"]] = relationship(
+        back_populates="difficulty",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+    chart_info_list: Mapped[list["ChartInfo"]] = relationship(
+        back_populates="difficulty",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+
+    rating_class: Mapped[int] = mapped_column(Integer, primary_key=True)
+
+    rating: Mapped[int] = mapped_column(Integer, nullable=False)
+    is_rating_plus: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+
+    chart_designer: Mapped[Optional[str]] = mapped_column(String)
+    jacket_designer: Mapped[Optional[str]] = mapped_column(String)
+
+    has_overriding_audio: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+    has_overriding_jacket: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+    jacket_night: Mapped[Optional[str]] = mapped_column(String)
+
+    title: Mapped[Optional[str]] = mapped_column(String)
+    artist: Mapped[Optional[str]] = mapped_column(String)
+    bg: Mapped[Optional[str]] = mapped_column(String)
+    bg_inverse: Mapped[Optional[str]] = mapped_column(String)
+    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)
+    is_legacy11: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+
+
+class DifficultyLocalization(ModelBase, ReprHelper):
+    __tablename__ = "difficulty_localization"
+    __table_args__ = (
+        ForeignKeyConstraint(
+            ["song_id", "rating_class"],
+            ["difficulty.song_id", "difficulty.rating_class"],
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+    )
+
+    difficulty: Mapped["Difficulty"] = relationship(
+        back_populates="localization_entries"
+    )
+    song_id: Mapped[str] = mapped_column(String, primary_key=True)
+    rating_class: Mapped[int] = mapped_column(Integer, primary_key=True)
+
+    lang: Mapped[str] = mapped_column(String, primary_key=True)
+    title: Mapped[Optional[str]] = mapped_column(String)

src/arcaea_offline/database/models/pack.py

@@ -0,0 +1,61 @@
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import Boolean, ForeignKey, Integer, String, Text, text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from ._base import ModelBase, ReprHelper
+
+if TYPE_CHECKING:
+    from .song import Song
+
+
+class Pack(ModelBase, ReprHelper):
+    __tablename__ = "pack"
+
+    songs: Mapped[list["Song"]] = relationship(
+        back_populates="pack",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+    localized_entries: Mapped[list["PackLocalization"]] = relationship(
+        back_populates="pack",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    name: Mapped[Optional[str]] = mapped_column(String, index=True)
+    description: Mapped[Optional[str]] = mapped_column(Text)
+    section: Mapped[Optional[str]] = mapped_column(String)
+    is_world_extend: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+
+    plus_character: Mapped[Optional[int]] = mapped_column(Integer)
+
+    append_parent_id: Mapped[Optional[str]] = mapped_column(
+        ForeignKey("pack.id", onupdate="CASCADE", ondelete="CASCADE")
+    )
+
+    parent: Mapped["Pack"] = relationship(
+        "Pack",
+        back_populates="appendages",
+        cascade="all, delete",
+        passive_deletes=True,
+        remote_side=[id],
+    )
+    appendages: Mapped[list["Pack"]] = relationship("Pack", back_populates="parent")
+
+
+class PackLocalization(ModelBase, ReprHelper):
+    __tablename__ = "pack_localization"
+
+    pack: Mapped["Pack"] = relationship(back_populates="localized_entries")
+    id: Mapped[str] = mapped_column(
+        ForeignKey("pack.id", onupdate="CASCADE", ondelete="CASCADE"),
+        primary_key=True,
+    )
+
+    lang: Mapped[str] = mapped_column(String, primary_key=True)
+    name: Mapped[Optional[str]] = mapped_column(String)
+    description: Mapped[Optional[str]] = mapped_column(Text)

src/arcaea_offline/database/models/play_result.py

@@ -1,18 +1,14 @@
 from datetime import datetime, timezone
 from typing import Optional
+from uuid import UUID, uuid4
 
-from sqlalchemy import ForeignKey, and_, case, func, inspect, select, text
-from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy import Integer, String, Text, Uuid, case, func, inspect, select, text
+from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy_utils import create_view
 
-from arcaea_offline.constants.enums import (
-    ArcaeaPlayResultClearType,
-    ArcaeaPlayResultModifier,
-    ArcaeaRatingClass,
-)
-
-from .arcaea import ChartInfo, Difficulty
-from .base import ModelsV5Base, ModelsV5ViewBase, ReprHelper
+from ._base import ModelBase, ModelViewBase, ReprHelper
+from .chart_info import ChartInfo
+from .difficulty import Difficulty
 
 __all__ = [
     "CalculatedPotential",
@@ -22,59 +18,53 @@ __all__ = [
 ]
 
 
-class PlayResult(ModelsV5Base, ReprHelper):
-    __tablename__ = "play_results"
+class PlayResult(ModelBase, ReprHelper):
+    __tablename__ = "play_result"
 
     id: Mapped[int] = mapped_column(autoincrement=True, primary_key=True)
-    song_id: Mapped[str] = mapped_column(
-        ForeignKey(Difficulty.song_id, onupdate="CASCADE", ondelete="NO ACTION"),
-        index=True,
+    uuid: Mapped[UUID] = mapped_column(
+        Uuid, nullable=False, unique=True, default=lambda: uuid4()
     )
-    rating_class: Mapped[ArcaeaRatingClass] = mapped_column(
-        ForeignKey(Difficulty.rating_class, onupdate="CASCADE", ondelete="NO ACTION"),
-        index=True,
+    song_id: Mapped[str] = mapped_column(String)
+    rating_class: Mapped[int] = mapped_column(Integer)
+    played_at: Mapped[Optional[datetime]] = mapped_column(
+        default=lambda: datetime.now(timezone.utc)
     )
     score: Mapped[int]
     pure: Mapped[Optional[int]]
+    pure_early: Mapped[Optional[int]]
+    pure_late: Mapped[Optional[int]]
     far: Mapped[Optional[int]]
+    far_early: Mapped[Optional[int]]
+    far_late: Mapped[Optional[int]]
     lost: Mapped[Optional[int]]
-    date: Mapped[Optional[datetime]] = mapped_column(
-        default=lambda: datetime.now(timezone.utc)
-    )
+
     max_recall: Mapped[Optional[int]]
-    modifier: Mapped[Optional[ArcaeaPlayResultModifier]]
-    clear_type: Mapped[Optional[ArcaeaPlayResultClearType]]
-    comment: Mapped[Optional[str]]
-
-    difficulty: Mapped[Difficulty] = relationship(
-        primaryjoin=and_(
-            song_id == Difficulty.song_id,
-            rating_class == Difficulty.rating_class,
-        ),
-        viewonly=True,
-    )
+    clear_type: Mapped[Optional[int]]
+    modifier: Mapped[Optional[int]]
+    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 PlayResultCalculated(ModelsV5ViewBase, ReprHelper):
+class PlayResultCalculated(ModelViewBase, ReprHelper):
     __tablename__ = "play_results_calculated"
 
     id: Mapped[int]
+    uuid: Mapped[UUID]
     song_id: Mapped[str]
-    rating_class: Mapped[ArcaeaRatingClass]
+    rating_class: Mapped[int]
     score: Mapped[int]
     pure: Mapped[Optional[int]]
+    pure_early: Mapped[Optional[int]]
+    pure_late: Mapped[Optional[int]]
     shiny_pure: Mapped[Optional[int]]
     far: Mapped[Optional[int]]
+    far_early: Mapped[Optional[int]]
+    far_late: Mapped[Optional[int]]
     lost: Mapped[Optional[int]]
-    date: Mapped[Optional[datetime]]
+    played_at: Mapped[Optional[datetime]]
     max_recall: Mapped[Optional[int]]
-    modifier: Mapped[Optional[ArcaeaPlayResultModifier]]
-    clear_type: Mapped[Optional[ArcaeaPlayResultClearType]]
+    modifier: Mapped[Optional[int]]
+    clear_type: Mapped[Optional[int]]
     potential: Mapped[float]
     comment: Mapped[Optional[str]]
 
@@ -106,7 +96,7 @@ class PlayResultCalculated(ModelsV5ViewBase, ReprHelper):
             ).label("shiny_pure"),
             PlayResult.far,
             PlayResult.lost,
-            PlayResult.date,
+            PlayResult.played_at,
             PlayResult.max_recall,
             PlayResult.modifier,
             PlayResult.clear_type,
@@ -137,26 +127,31 @@ class PlayResultCalculated(ModelsV5ViewBase, ReprHelper):
             (Difficulty.song_id == PlayResult.song_id)
             & (Difficulty.rating_class == PlayResult.rating_class),
         ),
-        metadata=ModelsV5ViewBase.metadata,
+        metadata=ModelViewBase.metadata,
         cascade_on_drop=False,
     )
 
 
-class PlayResultBest(ModelsV5ViewBase, ReprHelper):
+class PlayResultBest(ModelViewBase, ReprHelper):
     __tablename__ = "play_results_best"
 
     id: Mapped[int]
+    uuid: Mapped[UUID]
     song_id: Mapped[str]
-    rating_class: Mapped[ArcaeaRatingClass]
+    rating_class: Mapped[int]
     score: Mapped[int]
     pure: Mapped[Optional[int]]
+    pure_early: Mapped[Optional[int]]
+    pure_late: Mapped[Optional[int]]
     shiny_pure: Mapped[Optional[int]]
     far: Mapped[Optional[int]]
+    far_early: Mapped[Optional[int]]
+    far_late: Mapped[Optional[int]]
     lost: Mapped[Optional[int]]
-    date: Mapped[Optional[datetime]]
+    played_at: Mapped[Optional[datetime]]
     max_recall: Mapped[Optional[int]]
-    modifier: Mapped[Optional[ArcaeaPlayResultModifier]]
-    clear_type: Mapped[Optional[ArcaeaPlayResultClearType]]
+    modifier: Mapped[Optional[int]]
+    clear_type: Mapped[Optional[int]]
     potential: Mapped[float]
     comment: Mapped[Optional[str]]
 
@@ -173,12 +168,12 @@ class PlayResultBest(ModelsV5ViewBase, ReprHelper):
         .select_from(PlayResultCalculated)
         .group_by(PlayResultCalculated.song_id, PlayResultCalculated.rating_class)
         .order_by(PlayResultCalculated.potential.desc()),
-        metadata=ModelsV5ViewBase.metadata,
+        metadata=ModelViewBase.metadata,
         cascade_on_drop=False,
     )
 
 
-class CalculatedPotential(ModelsV5ViewBase, ReprHelper):
+class CalculatedPotential(ModelViewBase, ReprHelper):
     __tablename__ = "calculated_potential"
 
     b30: Mapped[float]
@@ -192,6 +187,6 @@ class CalculatedPotential(ModelsV5ViewBase, ReprHelper):
     __table__ = create_view(
         name=__tablename__,
         selectable=select(func.avg(_select_bests_subquery.c.b30_sum).label("b30")),
-        metadata=ModelsV5ViewBase.metadata,
+        metadata=ModelViewBase.metadata,
         cascade_on_drop=False,
     )

src/arcaea_offline/database/models/song.py

@@ -0,0 +1,85 @@
+from datetime import datetime
+from decimal import Decimal
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import Boolean, ForeignKey, Integer, Numeric, String, text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from ._base import ModelBase, ReprHelper
+from ._types import ForceTimezoneDateTime
+
+if TYPE_CHECKING:
+    from .difficulty import Difficulty
+    from .pack import Pack
+
+
+class Song(ModelBase, ReprHelper):
+    __tablename__ = "song"
+
+    pack_id: Mapped[str] = mapped_column(
+        ForeignKey("pack.id", onupdate="CASCADE", ondelete="CASCADE")
+    )
+    pack: Mapped["Pack"] = relationship(back_populates="songs")
+    difficulties: Mapped[list["Difficulty"]] = relationship(
+        back_populates="song",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+    localized_entries: Mapped[list["SongLocalization"]] = relationship(
+        back_populates="song",
+        cascade="all, delete",
+        passive_deletes=True,
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    idx: Mapped[Optional[int]] = mapped_column(Integer)
+    title: Mapped[Optional[str]] = mapped_column(String, index=True)
+    artist: Mapped[Optional[str]] = mapped_column(String, index=True)
+    is_deleted: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+
+    added_at: Mapped[datetime] = mapped_column(
+        ForceTimezoneDateTime, nullable=False, index=True
+    )
+    version: Mapped[Optional[str]] = mapped_column(String)
+
+    bpm: Mapped[Optional[str]] = mapped_column(String)
+    bpm_base: Mapped[Optional[Decimal]] = mapped_column(Numeric(asdecimal=True))
+    is_remote: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+    is_unlockable_in_world: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+    is_beyond_unlock_state_local: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )
+    purchase: Mapped[Optional[str]] = mapped_column(String)
+    category: Mapped[Optional[str]] = mapped_column(String)
+
+    side: Mapped[Optional[int]] = mapped_column(Integer)
+    bg: Mapped[Optional[str]] = mapped_column(String)
+    bg_inverse: Mapped[Optional[str]] = mapped_column(String)
+    bg_day: Mapped[Optional[str]] = mapped_column(String)
+    bg_night: Mapped[Optional[str]] = mapped_column(String)
+
+    source: Mapped[Optional[str]] = mapped_column(String)
+    source_copyright: Mapped[Optional[str]] = mapped_column(String)
+
+
+class SongLocalization(ModelBase, ReprHelper):
+    __tablename__ = "song_localization"
+
+    song: Mapped["Song"] = relationship(back_populates="localized_entries")
+    id: Mapped[str] = mapped_column(
+        ForeignKey("song.id", onupdate="CASCADE", ondelete="CASCADE"),
+        primary_key=True,
+    )
+
+    lang: Mapped[str] = mapped_column(String, primary_key=True)
+    title: Mapped[Optional[str]] = mapped_column(String)
+    source: Mapped[Optional[str]] = mapped_column(String)
+    has_jacket: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, insert_default=False, server_default=text("0")
+    )