refactor!: remove v4 database

This commit is contained in:
283375 2024-10-02 12:51:22 +08:00
parent 779fe0130e
commit 6e8ac3dee7
Signed by: 283375
SSH Key Fingerprint: SHA256:UcX0qg6ZOSDOeieKPGokA5h7soykG61nz2uxuQgVLSk
34 changed files with 59 additions and 1113 deletions

View File

@ -0,0 +1 @@
DATABASE_VERSION = 5

View File

@ -1,3 +0,0 @@
from .db import Database
__all__ = ["Database"]

View File

@ -1,404 +0,0 @@
import logging
import math
from typing import Iterable, Optional, Type, Union
from sqlalchemy import Engine, func, inspect, select
from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, sessionmaker
from arcaea_offline.singleton import Singleton
from .models.v4.config import ConfigBase, Property
from .models.v4.scores import (
CalculatedPotential,
Score,
ScoreBest,
ScoreCalculated,
ScoresBase,
ScoresViewBase,
)
from .models.v4.songs import (
Chart,
ChartInfo,
Difficulty,
DifficultyLocalized,
Pack,
PackLocalized,
Song,
SongLocalized,
SongsBase,
SongsViewBase,
)
logger = logging.getLogger(__name__)
class Database(metaclass=Singleton):
def __init__(self, engine: Optional[Engine]):
try:
self.__engine
except AttributeError:
self.__engine = None
if engine is None:
if isinstance(self.engine, Engine):
return
raise ValueError("No sqlalchemy.Engine instance specified before.")
if not isinstance(engine, Engine):
raise ValueError(
f"A sqlalchemy.Engine instance expected, not {repr(engine)}"
)
if isinstance(self.engine, Engine):
logger.warning(
"A sqlalchemy.Engine instance %r has been specified "
"and will be replaced to %r",
self.engine,
engine,
)
self.engine = engine
@property
def engine(self) -> Engine:
return self.__engine # type: ignore
@engine.setter
def engine(self, value: Engine):
if not isinstance(value, Engine):
raise ValueError("Database.engine only accepts sqlalchemy.Engine")
self.__engine = value
self.__sessionmaker = sessionmaker(self.__engine)
@property
def sessionmaker(self):
return self.__sessionmaker
# region init
def init(self, *, checkfirst: bool = True):
# create tables & views
if checkfirst:
# > https://github.com/kvesteri/sqlalchemy-utils/issues/396
# > view.create_view() causes DuplicateTableError on
# > Base.metadata.create_all(checkfirst=True)
# so if `checkfirst` is True, drop these views before creating
SongsViewBase.metadata.drop_all(self.engine)
ScoresViewBase.metadata.drop_all(self.engine)
SongsBase.metadata.create_all(self.engine, checkfirst=checkfirst)
SongsViewBase.metadata.create_all(self.engine)
ScoresBase.metadata.create_all(self.engine, checkfirst=checkfirst)
ScoresViewBase.metadata.create_all(self.engine)
ConfigBase.metadata.create_all(self.engine, checkfirst=checkfirst)
# insert version property
with self.sessionmaker() as session:
stmt = select(Property.value).where(Property.key == "version")
result = session.execute(stmt).fetchone()
if not checkfirst or not result:
session.add(Property(key="version", value="4"))
session.commit()
def check_init(self) -> bool:
# check table exists
expect_tables = (
list(SongsBase.metadata.tables.keys())
+ list(ScoresBase.metadata.tables.keys())
+ list(ConfigBase.metadata.tables.keys())
+ [
Chart.__tablename__,
ScoreCalculated.__tablename__,
ScoreBest.__tablename__,
CalculatedPotential.__tablename__,
]
)
return all(inspect(self.engine).has_table(t) for t in expect_tables)
# endregion
def version(self) -> Union[int, None]:
stmt = select(Property).where(Property.key == "version")
with self.sessionmaker() as session:
result = session.scalar(stmt)
return None if result is None else int(result.value)
# region Pack
def get_packs(self):
stmt = select(Pack)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_pack(self, pack_id: str):
stmt = select(Pack).where(Pack.id == pack_id)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
def get_pack_localized(self, pack_id: str):
stmt = select(PackLocalized).where(PackLocalized.id == pack_id)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# endregion
# region Song
def get_songs(self):
stmt = select(Song)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_songs_by_pack_id(self, pack_id: str):
stmt = select(Song).where(Song.set == pack_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_song(self, song_id: str):
stmt = select(Song).where(Song.id == song_id)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
def get_song_localized(self, song_id: str):
stmt = select(SongLocalized).where(SongLocalized.id == song_id)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# endregion
# region Difficulty
def get_difficulties(self):
stmt = select(Difficulty)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_difficulties_by_song_id(self, song_id: str):
stmt = select(Difficulty).where(Difficulty.song_id == song_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_difficulties_localized_by_song_id(self, song_id: str):
stmt = select(DifficultyLocalized).where(DifficultyLocalized.song_id == song_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_difficulty(self, song_id: str, rating_class: int):
stmt = select(Difficulty).where(
(Difficulty.song_id == song_id) & (Difficulty.rating_class == rating_class)
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
def get_difficulty_localized(self, song_id: str, rating_class: int):
stmt = select(DifficultyLocalized).where(
(DifficultyLocalized.song_id == song_id)
& (DifficultyLocalized.rating_class == rating_class)
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# endregion
# region ChartInfo
def get_chart_infos(self):
stmt = select(ChartInfo)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_chart_infos_by_song_id(self, song_id: str):
stmt = select(ChartInfo).where(ChartInfo.song_id == song_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_chart_info(self, song_id: str, rating_class: int):
stmt = select(ChartInfo).where(
(ChartInfo.song_id == song_id) & (ChartInfo.rating_class == rating_class)
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# endregion
# region Chart
def get_charts_by_pack_id(self, pack_id: str):
stmt = select(Chart).where(Chart.set == pack_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_charts_by_song_id(self, song_id: str):
stmt = select(Chart).where(Chart.song_id == song_id)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_charts_by_constant(self, constant: int):
stmt = select(Chart).where(Chart.constant == constant)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_chart(self, song_id: str, rating_class: int):
stmt = select(Chart).where(
(Chart.song_id == song_id) & (Chart.rating_class == rating_class)
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# endregion
# region Score
def get_scores(self):
stmt = select(Score)
with self.sessionmaker() as session:
results = list(session.scalars(stmt))
return results
def get_score(self, score_id: int):
stmt = select(Score).where(Score.id == score_id)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
def get_score_best(self, song_id: str, rating_class: int):
stmt = select(ScoreBest).where(
(ScoreBest.song_id == song_id) & (ScoreBest.rating_class == rating_class)
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
def insert_score(self, score: Score):
with self.sessionmaker() as session:
session.add(score)
session.commit()
def insert_scores(self, scores: Iterable[Score]):
with self.sessionmaker() as session:
session.add_all(scores)
session.commit()
def update_score(self, score: Score):
if score.id is None:
raise ValueError(
"Cannot determine which score to update, please specify `score.id`"
)
with self.sessionmaker() as session:
session.merge(score)
session.commit()
def delete_score(self, score: Score):
with self.sessionmaker() as session:
session.delete(score)
session.commit()
def recommend_charts(self, play_result: float, bounds: float = 0.1):
base_constant = math.ceil(play_result * 10)
results = []
results_id = []
with self.sessionmaker() as session:
for constant in range(base_constant - 20, base_constant + 1):
# from Pure Memory(EX+) to AA
score_modifier = (play_result * 10 - constant) / 10
if score_modifier >= 2.0: # noqa: PLR2004
min_score = 10000000
elif score_modifier >= 1.0:
min_score = 200000 * (score_modifier - 1) + 9800000
else:
min_score = 300000 * score_modifier + 9500000
min_score = int(min_score)
charts = self.get_charts_by_constant(constant)
for chart in charts:
score_best_stmt = select(ScoreBest).where(
(ScoreBest.song_id == chart.song_id)
& (ScoreBest.rating_class == chart.rating_class)
& (ScoreBest.score >= min_score)
& (play_result - bounds < ScoreBest.potential)
& (ScoreBest.potential < play_result + bounds)
)
if session.scalar(score_best_stmt):
chart_id = f"{chart.song_id},{chart.rating_class}"
if chart_id not in results_id:
results.append(chart)
results_id.append(chart_id)
return results
# endregion
def get_b30(self):
stmt = select(CalculatedPotential.b30).select_from(CalculatedPotential)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result
# region COUNT
def __count_table(self, base: Type[DeclarativeBase]):
stmt = select(func.count()).select_from(base)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result or 0
def __count_column(self, column: InstrumentedAttribute):
stmt = select(func.count(column))
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result or 0
def count_packs(self):
return self.__count_column(Pack.id)
def count_songs(self):
return self.__count_column(Song.id)
def count_difficulties(self):
return self.__count_table(Difficulty)
def count_chart_infos(self):
return self.__count_table(ChartInfo)
def count_complete_chart_infos(self):
stmt = (
select(func.count())
.select_from(ChartInfo)
.where((ChartInfo.constant != None) & (ChartInfo.notes != None)) # noqa: E711
)
with self.sessionmaker() as session:
result = session.scalar(stmt)
return result or 0
def count_charts(self):
return self.__count_table(Chart)
def count_scores(self):
return self.__count_column(Score.id)
def count_scores_calculated(self):
return self.__count_table(ScoreCalculated)
def count_scores_best(self):
return self.__count_table(ScoreBest)
# endregion

View File

@ -0,0 +1,38 @@
from .arcaea import (
Chart,
ChartInfo,
Difficulty,
DifficultyLocalized,
Pack,
PackLocalized,
Song,
SongLocalized,
SongSearchWord,
)
from .base import ModelsV5Base, ModelsV5ViewBase
from .config import Property
from .play_results import (
CalculatedPotential,
PlayResult,
PlayResultBest,
PlayResultCalculated,
)
__all__ = [
"CalculatedPotential",
"Chart",
"ChartInfo",
"Difficulty",
"DifficultyLocalized",
"ModelsV5Base",
"ModelsV5ViewBase",
"Pack",
"PackLocalized",
"PlayResult",
"PlayResultBest",
"PlayResultCalculated",
"Property",
"Song",
"SongLocalized",
"SongSearchWord",
]

View File

@ -1,3 +1,4 @@
from datetime import datetime
from typing import List, Optional
from sqlalchemy import Enum, ForeignKey, and_, func, select
@ -72,7 +73,7 @@ class Song(ModelsV5Base, ReprHelper):
audio_preview_end: Mapped[Optional[int]]
side: Mapped[Optional[ArcaeaSongSide]]
version: Mapped[Optional[str]]
date: Mapped[Optional[int]]
date: Mapped[Optional[datetime]]
bg: Mapped[Optional[str]]
bg_inverse: Mapped[Optional[str]]
bg_day: Mapped[Optional[str]]
@ -143,7 +144,7 @@ class Difficulty(ModelsV5Base, ReprHelper):
bpm: Mapped[Optional[str]]
bpm_base: Mapped[Optional[float]]
version: Mapped[Optional[str]]
date: Mapped[Optional[int]]
date: Mapped[Optional[datetime]]
song: Mapped[Song] = relationship(back_populates="difficulties", viewonly=True)
chart_info: Mapped[Optional["ChartInfo"]] = relationship(
@ -225,7 +226,7 @@ class Chart(ModelsV5ViewBase, ReprHelper):
audio_preview_end: Mapped[Optional[int]]
side: Mapped[Optional[int]]
version: Mapped[Optional[str]]
date: Mapped[Optional[int]]
date: Mapped[Optional[datetime]]
bg: Mapped[Optional[str]]
bg_inverse: Mapped[Optional[str]]
bg_day: Mapped[Optional[str]]

View File

@ -10,7 +10,7 @@ from arcaea_offline.constants.enums import (
ArcaeaSongSide,
)
from .._custom_types import DbIntEnum, TZDateTime
from ._custom_types import DbIntEnum, TZDateTime
TYPE_ANNOTATION_MAP = {
datetime: TZDateTime,

View File

@ -1,42 +0,0 @@
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,
)
__all__ = [
"CalculatedPotential",
"Chart",
"ChartInfo",
"ConfigBase",
"Difficulty",
"DifficultyLocalized",
"Pack",
"PackLocalized",
"Property",
"Score",
"ScoreBest",
"ScoreCalculated",
"ScoresBase",
"ScoresViewBase",
"Song",
"SongLocalized",
"SongsBase",
"SongsViewBase",
]

View File

@ -1,36 +0,0 @@
# 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__()

View File

@ -1,22 +0,0 @@
# 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())

View File

@ -1,188 +0,0 @@
# 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__ = [
"CalculatedPotential",
"Score",
"ScoreBest",
"ScoreCalculated",
"ScoresBase",
"ScoresViewBase",
]
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), # noqa: PLR2004
(
Score.score >= 9800000, # noqa: PLR2004
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,
)

View File

@ -1,241 +0,0 @@
# 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__ = [
"Chart",
"ChartInfo",
"Difficulty",
"DifficultyLocalized",
"Pack",
"PackLocalized",
"Song",
"SongLocalized",
"SongsBase",
"SongsViewBase",
]
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,
)

View File

@ -1,38 +0,0 @@
from .arcaea import (
Chart,
ChartInfo,
Difficulty,
DifficultyLocalized,
Pack,
PackLocalized,
Song,
SongLocalized,
SongSearchWord,
)
from .base import ModelsV5Base, ModelsV5ViewBase
from .config import Property
from .play_results import (
CalculatedPotential,
PlayResult,
PlayResultBest,
PlayResultCalculated,
)
__all__ = [
"CalculatedPotential",
"Chart",
"ChartInfo",
"Difficulty",
"DifficultyLocalized",
"ModelsV5Base",
"ModelsV5ViewBase",
"Pack",
"PackLocalized",
"PlayResult",
"PlayResultBest",
"PlayResultCalculated",
"Property",
"Song",
"SongLocalized",
"SongSearchWord",
]

View File

@ -6,7 +6,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import (
from arcaea_offline.database.models import (
PlayResultBest,
PlayResultCalculated,
)

View File

@ -6,7 +6,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from arcaea_offline.constants.enums.arcaea import ArcaeaLanguage
from arcaea_offline.database.models.v5 import Difficulty, Pack, Song
from arcaea_offline.database.models import Difficulty, Pack, Song
from .definitions import ArcsongJsonDifficultyItem, ArcsongJsonRoot, ArcsongJsonSongItem

View File

@ -1,6 +1,6 @@
from typing import List
from arcaea_offline.database.models.v5 import PlayResult
from arcaea_offline.database.models import PlayResult
from .definitions import (
ArcaeaOfflineDEFv2PlayResultItem,

View File

@ -4,7 +4,7 @@ from sqlalchemy import func
from sqlalchemy.orm import Session
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import (
from arcaea_offline.database.models import (
ChartInfo,
Difficulty,
PlayResultBest,

View File

@ -10,7 +10,7 @@ from arcaea_offline.constants.enums import (
ArcaeaRatingClass,
ArcaeaSongSide,
)
from arcaea_offline.database.models.v5 import (
from arcaea_offline.database.models import (
Difficulty,
DifficultyLocalized,
Pack,

View File

@ -8,7 +8,7 @@ from arcaea_offline.constants.enums import (
ArcaeaPlayResultModifier,
ArcaeaRatingClass,
)
from arcaea_offline.database.models.v5 import PlayResult
from arcaea_offline.database.models import PlayResult
from .common import fix_timestamp

View File

@ -12,7 +12,7 @@ from arcaea_offline.constants.enums import (
ArcaeaPlayResultModifier,
ArcaeaRatingClass,
)
from arcaea_offline.database.models.v5 import PlayResult
from arcaea_offline.database.models import PlayResult
from .common import fix_timestamp

View File

@ -2,7 +2,7 @@ import sqlite3
from typing import List, overload
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import ChartInfo
from arcaea_offline.database.models import ChartInfo
class ArcsongDatabaseImporter:

View File

@ -3,7 +3,7 @@ from contextlib import closing
from typing import List, overload
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import ChartInfo
from arcaea_offline.database.models import ChartInfo
class ChartInfoDatabaseParser:

View File

@ -1,12 +0,0 @@
from typing import Generic, TypeVar
T = TypeVar("T")
class Singleton(type, Generic[T]):
_instance = None
def __call__(cls, *args, **kwargs) -> T:
if cls._instance is None:
cls._instance = super().__call__(*args, **kwargs)
return cls._instance

View File

@ -11,7 +11,7 @@ Database model v5 common relationships
"""
from arcaea_offline.constants.enums import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import (
from arcaea_offline.database.models import (
ChartInfo,
Difficulty,
ModelsV5Base,

View File

@ -5,7 +5,7 @@ Pack <> PackLocalized
"""
from arcaea_offline.constants.enums import ArcaeaLanguage
from arcaea_offline.database.models.v5 import ModelsV5Base, Pack, PackLocalized
from arcaea_offline.database.models import ModelsV5Base, Pack, PackLocalized
class TestPackRelationships:

View File

@ -7,7 +7,7 @@ Chart functionalities
"""
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import (
from arcaea_offline.database.models import (
Chart,
ChartInfo,
Difficulty,

View File

@ -1,108 +0,0 @@
from arcaea_offline.database.models.v4.songs import (
Chart,
ChartInfo,
Difficulty,
Pack,
Song,
SongsBase,
SongsViewBase,
)
def _song(**kw):
defaults = {"artist": "test"}
defaults.update(kw)
return Song(**defaults)
def _difficulty(**kw):
defaults = {"rating_plus": False, "audio_override": False, "jacket_override": False}
defaults.update(kw)
return Difficulty(**defaults)
class TestChart:
def init_db(self, session):
SongsBase.metadata.create_all(session.bind, checkfirst=False)
SongsViewBase.metadata.create_all(session.bind, checkfirst=False)
def test_chart_info(self, db_session):
self.init_db(db_session)
pre_entites = [
Pack(id="test", name="Test Pack"),
_song(idx=0, id="song0", set="test", title="Full Chart Info"),
_song(idx=1, id="song1", set="test", title="Partial Chart Info"),
_song(idx=2, id="song2", set="test", title="No Chart Info"),
_difficulty(song_id="song0", rating_class=2, rating=9),
_difficulty(song_id="song1", rating_class=2, rating=9),
_difficulty(song_id="song2", rating_class=2, rating=9),
ChartInfo(song_id="song0", rating_class=2, constant=90, notes=1234),
ChartInfo(song_id="song1", rating_class=2, constant=90),
]
db_session.add_all(pre_entites)
db_session.commit()
chart_song0_ratingclass2 = (
db_session.query(Chart)
.where((Chart.song_id == "song0") & (Chart.rating_class == 2))
.one()
)
assert chart_song0_ratingclass2.constant == 90
assert chart_song0_ratingclass2.notes == 1234
chart_song1_ratingclass2 = (
db_session.query(Chart)
.where((Chart.song_id == "song1") & (Chart.rating_class == 2))
.one()
)
assert chart_song1_ratingclass2.constant == 90
assert chart_song1_ratingclass2.notes is None
chart_song2_ratingclass2 = (
db_session.query(Chart)
.where((Chart.song_id == "song2") & (Chart.rating_class == 2))
.first()
)
assert chart_song2_ratingclass2 is None
def test_difficulty_title_override(self, db_session):
self.init_db(db_session)
pre_entites = [
Pack(id="test", name="Test Pack"),
_song(idx=0, id="test", set="test", title="Test"),
_difficulty(song_id="test", rating_class=0, rating=2),
_difficulty(song_id="test", rating_class=1, rating=5),
_difficulty(song_id="test", rating_class=2, rating=8),
_difficulty(
song_id="test", rating_class=3, rating=10, title="TEST ~REVIVE~"
),
ChartInfo(song_id="test", rating_class=0, constant=10),
ChartInfo(song_id="test", rating_class=1, constant=10),
ChartInfo(song_id="test", rating_class=2, constant=10),
ChartInfo(song_id="test", rating_class=3, constant=10),
]
db_session.add_all(pre_entites)
db_session.commit()
charts_original_title = (
db_session.query(Chart)
.where((Chart.song_id == "test") & (Chart.rating_class in [0, 1, 2]))
.all()
)
assert all(chart.title == "Test" for chart in charts_original_title)
chart_overrided_title = (
db_session.query(Chart)
.where((Chart.song_id == "test") & (Chart.rating_class == 3))
.one()
)
assert chart_overrided_title.title == "TEST ~REVIVE~"

View File

@ -6,7 +6,7 @@ from arcaea_offline.constants.enums.arcaea import (
ArcaeaPlayResultModifier,
ArcaeaRatingClass,
)
from arcaea_offline.database.models.v5.play_results import PlayResult
from arcaea_offline.database.models.play_results import PlayResult
from arcaea_offline.external.importers.arcaea.online import ArcaeaOnlineApiParser
API_RESULT = {

View File

@ -2,7 +2,7 @@ import sqlite3
import tests.resources
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import ChartInfo
from arcaea_offline.database.models import ChartInfo
from arcaea_offline.external.importers.arcsong import (
ArcsongDatabaseImporter,
)

View File

@ -2,7 +2,7 @@ import sqlite3
import tests.resources
from arcaea_offline.constants.enums.arcaea import ArcaeaRatingClass
from arcaea_offline.database.models.v5 import ChartInfo
from arcaea_offline.database.models import ChartInfo
from arcaea_offline.external.importers.chart_info_database import (
ChartInfoDatabaseParser,
)