961 lines
37 KiB
Python
961 lines
37 KiB
Python
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
|
|
from sqlalchemy.exc import IntegrityError # type: ignore
|
|
from sqlalchemy.types import String, Integer, JSON # type: ignore
|
|
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
|
|
from typing import Optional, Dict, List, Tuple, Any
|
|
|
|
from bemani.common import GameConstants, Time
|
|
from bemani.data.exceptions import ScoreSaveException
|
|
from bemani.data.mysql.base import BaseData, metadata
|
|
from bemani.data.types import Score, Attempt, Song, UserID
|
|
|
|
"""
|
|
Table for storing a score for a particular game. This is keyed by userid and
|
|
musicid, as a user can only have one score for a particular song/chart combo.
|
|
This has a JSON blob for any data the game wishes to store, such as points, medals,
|
|
ghost, etc.
|
|
|
|
Note that this is NOT keyed by game song id and chart, but by an internal musicid
|
|
managed by the music table. This is so we can support keeping the same score across
|
|
multiple games, even if the game changes the ID it refers to the song by.
|
|
"""
|
|
score = Table(
|
|
"score",
|
|
metadata,
|
|
Column("id", Integer, nullable=False, primary_key=True),
|
|
Column("userid", BigInteger(unsigned=True), nullable=False),
|
|
Column("musicid", Integer, nullable=False, index=True),
|
|
Column("points", Integer, nullable=False, index=True),
|
|
Column("timestamp", Integer, nullable=False, index=True),
|
|
Column("update", Integer, nullable=False, index=True),
|
|
Column("lid", Integer, nullable=False, index=True),
|
|
Column("data", JSON, nullable=False),
|
|
UniqueConstraint("userid", "musicid", name="userid_musicid"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
"""
|
|
Table for storing score history for a particular game. Every entry that is stored
|
|
or updated in score will be written into this table as well, for looking up history
|
|
over time.
|
|
"""
|
|
score_history = Table(
|
|
"score_history",
|
|
metadata,
|
|
Column("id", Integer, nullable=False, primary_key=True),
|
|
Column("userid", BigInteger(unsigned=True), nullable=False),
|
|
Column("musicid", Integer, nullable=False, index=True),
|
|
Column("points", Integer, nullable=False),
|
|
Column("timestamp", Integer, nullable=False, index=True),
|
|
Column("lid", Integer, nullable=False, index=True),
|
|
Column("new_record", Integer, nullable=False),
|
|
Column("data", JSON, nullable=False),
|
|
UniqueConstraint("userid", "musicid", "timestamp", name="userid_musicid_timestamp"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
"""
|
|
Table for storing the mapping between game songid/chart and musicid for the score
|
|
and score_history table. To find scores, you will want to join this table with
|
|
the score table where id = score.musicid and game/version/songid/chart matches.
|
|
|
|
NOTE that it is expected to see the same songid/chart present multiple times as long
|
|
as the game version changes. In this way, a song which is in multiple versions of
|
|
the game can be found when playing each version.
|
|
"""
|
|
music = Table(
|
|
"music",
|
|
metadata,
|
|
Column("id", Integer, nullable=False, index=True),
|
|
Column("songid", Integer, nullable=False),
|
|
Column("chart", Integer, nullable=False),
|
|
Column("game", String(32), nullable=False, index=True),
|
|
Column("version", Integer, nullable=False, index=True),
|
|
Column("name", String(255)),
|
|
Column("artist", String(255)),
|
|
Column("genre", String(255)),
|
|
Column("data", JSON),
|
|
UniqueConstraint(
|
|
"songid", "chart", "game", "version", name="songid_chart_game_version"
|
|
),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
|
|
class MusicData(BaseData):
|
|
def __get_musicid(
|
|
self, game: GameConstants, version: int, songid: int, songchart: int
|
|
) -> int:
|
|
"""
|
|
Given a game/version/songid/chart, look up the unique music ID for this song.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
songid - ID of the song according to the game.
|
|
songchart - Chart number according to the game.
|
|
|
|
Returns:
|
|
Integer representing music ID if found or raises an exception otherwise.
|
|
"""
|
|
sql = "SELECT id FROM music WHERE songid = :songid AND chart = :chart AND game = :game AND version = :version"
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"songid": songid,
|
|
"chart": songchart,
|
|
"game": game.value,
|
|
"version": version,
|
|
},
|
|
)
|
|
if cursor.rowcount != 1:
|
|
# music doesn't exist
|
|
raise Exception(
|
|
f"Song {songid} chart {songchart} doesn't exist for game {game} version {version}"
|
|
)
|
|
result = cursor.fetchone()
|
|
return result["id"]
|
|
|
|
def put_score(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
userid: UserID,
|
|
songid: int,
|
|
songchart: int,
|
|
location: int,
|
|
points: int,
|
|
data: Dict[str, Any],
|
|
new_record: bool,
|
|
timestamp: Optional[int] = None,
|
|
) -> None:
|
|
"""
|
|
Given a game/version/song/chart and user ID, save a new/updated high score.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
songid - ID of the song according to the game.
|
|
songchart - Chart number according to the game.
|
|
location - Machine ID where this score was earned.
|
|
points - Points obtained on this song.
|
|
data - Data that the game wishes to record along with the score.
|
|
new_record - Whether this score was a new record or not.
|
|
timestamp - Optional integer specifying when the high score happened.
|
|
"""
|
|
# First look up the song/chart from the music DB
|
|
musicid = self.__get_musicid(game, version, songid, songchart)
|
|
ts = timestamp if timestamp is not None else Time.now()
|
|
|
|
# Add to user score
|
|
if new_record:
|
|
# We want to update the timestamp/location to now if its a new record.
|
|
sql = (
|
|
"INSERT INTO `score` (`userid`, `musicid`, `points`, `data`, `timestamp`, `update`, `lid`) "
|
|
+ "VALUES (:userid, :musicid, :points, :data, :timestamp, :update, :location) "
|
|
+ "ON DUPLICATE KEY UPDATE data = VALUES(data), points = VALUES(points), "
|
|
+ "timestamp = VALUES(timestamp), `update` = VALUES(`update`), lid = VALUES(lid)"
|
|
)
|
|
else:
|
|
# We only want to add the timestamp if it is new.
|
|
sql = (
|
|
"INSERT INTO `score` (`userid`, `musicid`, `points`, `data`, `timestamp`, `update`, `lid`) "
|
|
+ "VALUES (:userid, :musicid, :points, :data, :timestamp, :update, :location) "
|
|
+ "ON DUPLICATE KEY UPDATE data = VALUES(data), points = VALUES(points), `update` = VALUES(`update`)"
|
|
)
|
|
self.execute(
|
|
sql,
|
|
{
|
|
"userid": userid,
|
|
"musicid": musicid,
|
|
"points": points,
|
|
"data": self.serialize(data),
|
|
"timestamp": ts,
|
|
"update": ts,
|
|
"location": location,
|
|
},
|
|
)
|
|
|
|
def put_attempt(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
userid: Optional[UserID],
|
|
songid: int,
|
|
songchart: int,
|
|
location: int,
|
|
points: int,
|
|
data: Dict[str, Any],
|
|
new_record: bool,
|
|
timestamp: Optional[int] = None,
|
|
) -> None:
|
|
"""
|
|
Given a game/version/song/chart and user ID, save a single score attempt.
|
|
|
|
Note that this is different than put_score above, because a user may have only one score
|
|
per song/chart in a given game, but they can have as many history entries as times played.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
songid - ID of the song according to the game.
|
|
songchart - Chart number according to the game.
|
|
location - Machine ID where this score was earned.
|
|
points - Points obtained on this song.
|
|
data - Optional data that the game wishes to record along with the score.
|
|
new_record - Whether this score was a new record or not.
|
|
timestamp - Optional integer specifying when the attempt happened.
|
|
"""
|
|
# First look up the song/chart from the music DB
|
|
musicid = self.__get_musicid(game, version, songid, songchart)
|
|
ts = timestamp if timestamp is not None else Time.now()
|
|
|
|
# Add to score history
|
|
sql = (
|
|
"INSERT INTO `score_history` (userid, musicid, timestamp, lid, new_record, points, data) "
|
|
+ "VALUES (:userid, :musicid, :timestamp, :location, :new_record, :points, :data)"
|
|
)
|
|
try:
|
|
self.execute(
|
|
sql,
|
|
{
|
|
"userid": userid if userid is not None else 0,
|
|
"musicid": musicid,
|
|
"timestamp": ts,
|
|
"location": location,
|
|
"new_record": 1 if new_record else 0,
|
|
"points": points,
|
|
"data": self.serialize(data),
|
|
},
|
|
)
|
|
except IntegrityError:
|
|
raise ScoreSaveException(
|
|
f"There is already an attempt by {userid if userid is not None else 0} for music id {musicid} at {ts}"
|
|
)
|
|
|
|
def get_score(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
userid: UserID,
|
|
songid: int,
|
|
songchart: int,
|
|
) -> Optional[Score]:
|
|
"""
|
|
Look up a user's previous high score.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
songid - ID of the song according to the game.
|
|
songchart - Chart number according to the game.
|
|
|
|
Returns:
|
|
The optional data stored by the game previously, or None if no score exists.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, music.chart AS chart, score.id AS scorekey, score.timestamp AS timestamp, score.update AS `update`, score.lid AS lid, "
|
|
+ "(select COUNT(score_history.timestamp) FROM score_history WHERE score_history.musicid = music.id AND score_history.userid = :userid) AS plays, "
|
|
+ "score.points AS points, score.data AS data FROM score, music WHERE score.userid = :userid AND score.musicid = music.id "
|
|
+ "AND music.game = :game AND music.version = :version AND music.songid = :songid AND music.chart = :songchart"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"userid": userid,
|
|
"game": game.value,
|
|
"version": version,
|
|
"songid": songid,
|
|
"songchart": songchart,
|
|
},
|
|
)
|
|
if cursor.rowcount != 1:
|
|
# score doesn't exist
|
|
return None
|
|
|
|
result = cursor.fetchone()
|
|
return Score(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["update"],
|
|
result["lid"],
|
|
result["plays"],
|
|
self.deserialize(result["data"]),
|
|
)
|
|
|
|
def get_score_by_key(
|
|
self, game: GameConstants, version: int, key: int
|
|
) -> Optional[Tuple[UserID, Score]]:
|
|
"""
|
|
Look up previous high score by key.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
key - Integer representing a unique key fetched in a previous Score lookup.
|
|
|
|
Returns:
|
|
The optional data stored by the game previously, or None if no score exists.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, music.chart AS chart, score.id AS scorekey, score.timestamp AS timestamp, score.update AS `update`, "
|
|
+ "score.userid AS userid, score.lid AS lid, "
|
|
+ "(select COUNT(score_history.timestamp) FROM score_history WHERE score_history.musicid = music.id AND score_history.userid = score.userid) AS plays, "
|
|
+ "score.points AS points, score.data AS data FROM score, music WHERE score.id = :scorekey AND score.musicid = music.id "
|
|
+ "AND music.game = :game AND music.version = :version"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"scorekey": key,
|
|
},
|
|
)
|
|
if cursor.rowcount != 1:
|
|
# score doesn't exist
|
|
return None
|
|
|
|
result = cursor.fetchone()
|
|
return (
|
|
UserID(result["userid"]),
|
|
Score(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["update"],
|
|
result["lid"],
|
|
result["plays"],
|
|
self.deserialize(result["data"]),
|
|
),
|
|
)
|
|
|
|
def get_scores(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
userid: UserID,
|
|
since: Optional[int] = None,
|
|
until: Optional[int] = None,
|
|
) -> List[Score]:
|
|
"""
|
|
Look up all of a user's previous high scores.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
|
|
Returns:
|
|
A list of Score objects representing all high scores for a game.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, music.chart AS chart, score.id AS scorekey, score.timestamp AS timestamp, score.update AS `update`, score.lid AS lid, "
|
|
+ "(select COUNT(score_history.timestamp) FROM score_history WHERE score_history.musicid = music.id AND score_history.userid = :userid) AS plays, "
|
|
+ "score.points AS points, score.data AS data FROM score, music WHERE score.userid = :userid AND score.musicid = music.id "
|
|
+ "AND music.game = :game AND music.version = :version"
|
|
)
|
|
if since is not None:
|
|
sql = sql + " AND score.update >= :since"
|
|
if until is not None:
|
|
sql = sql + " AND score.update < :until"
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"userid": userid,
|
|
"game": game.value,
|
|
"version": version,
|
|
"since": since,
|
|
"until": until,
|
|
},
|
|
)
|
|
|
|
scores = []
|
|
for result in cursor.fetchall():
|
|
scores.append(
|
|
Score(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["update"],
|
|
result["lid"],
|
|
result["plays"],
|
|
self.deserialize(result["data"]),
|
|
)
|
|
)
|
|
|
|
return scores
|
|
|
|
def get_most_played(
|
|
self, game: GameConstants, version: int, userid: UserID, count: int
|
|
) -> List[Tuple[int, int]]:
|
|
"""
|
|
Look up a user's most played songs.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
count - Number of scores to look up.
|
|
|
|
Returns:
|
|
A list of tuples, containing the songid and the number of plays across all charts for that song.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, COUNT(score_history.timestamp) AS plays FROM score_history, music "
|
|
+ "WHERE score_history.userid = :userid AND score_history.musicid = music.id "
|
|
+ "AND music.game = :game AND music.version = :version "
|
|
+ "GROUP BY songid ORDER BY plays DESC LIMIT :count"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{"userid": userid, "game": game.value, "version": version, "count": count},
|
|
)
|
|
|
|
most_played = []
|
|
for result in cursor.fetchall():
|
|
most_played.append((result["songid"], result["plays"]))
|
|
|
|
return most_played
|
|
|
|
def get_last_played(
|
|
self, game: GameConstants, version: int, userid: UserID, count: int
|
|
) -> List[Tuple[int, int]]:
|
|
"""
|
|
Look up a user's last played songs.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userid - Integer representing a user. Usually looked up with UserData.
|
|
count - Number of scores to look up.
|
|
|
|
Returns:
|
|
A list of tuples, containing the songid and the last played time for this song.
|
|
"""
|
|
sql = (
|
|
"SELECT DISTINCT(music.songid) AS songid, score_history.timestamp AS timestamp FROM score_history, music "
|
|
+ "WHERE score_history.userid = :userid AND score_history.musicid = music.id "
|
|
+ "AND music.game = :game AND music.version = :version "
|
|
+ "ORDER BY timestamp DESC LIMIT :count"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{"userid": userid, "game": game.value, "version": version, "count": count},
|
|
)
|
|
|
|
last_played = []
|
|
for result in cursor.fetchall():
|
|
last_played.append((result["songid"], result["timestamp"]))
|
|
|
|
return last_played
|
|
|
|
def get_hit_chart(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
count: int,
|
|
days: Optional[int] = None,
|
|
) -> List[Tuple[int, int]]:
|
|
"""
|
|
Look up a game's most played songs.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
count - Number of scores to look up.
|
|
|
|
Returns:
|
|
A list of tuples, containing the songid and the number of plays across all charts for that song.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, COUNT(score_history.timestamp) AS plays FROM score_history, music "
|
|
+ "WHERE score_history.musicid = music.id AND music.game = :game AND music.version = :version "
|
|
)
|
|
timestamp: Optional[int] = None
|
|
if days is not None:
|
|
# Only select the last X days of hit chart
|
|
sql = sql + "AND score_history.timestamp > :timestamp "
|
|
timestamp = Time.now() - (Time.SECONDS_IN_DAY * days)
|
|
|
|
sql = sql + "GROUP BY songid ORDER BY plays DESC LIMIT :count"
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"count": count,
|
|
"timestamp": timestamp,
|
|
},
|
|
)
|
|
|
|
most_played = []
|
|
for result in cursor.fetchall():
|
|
most_played.append((result["songid"], result["plays"]))
|
|
|
|
return most_played
|
|
|
|
def get_song(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
songid: int,
|
|
songchart: int,
|
|
) -> Optional[Song]:
|
|
"""
|
|
Given a game/version/songid/chart, look up the name, artist and genre of that song.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
songid - Integer representing the ID (from the game) for this song.
|
|
songchart - Integer representing the chart for this song.
|
|
|
|
Returns:
|
|
A Song object representing the song details
|
|
"""
|
|
sql = (
|
|
"SELECT music.name AS name, music.artist AS artist, music.genre AS genre, music.data AS data "
|
|
+ "FROM music WHERE music.game = :game AND music.version = :version AND "
|
|
+ "music.songid = :songid AND music.chart = :songchart"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"songid": songid,
|
|
"songchart": songchart,
|
|
},
|
|
)
|
|
if cursor.rowcount != 1:
|
|
# music doesn't exist
|
|
return None
|
|
result = cursor.fetchone()
|
|
return Song(
|
|
game,
|
|
version,
|
|
songid,
|
|
songchart,
|
|
result["name"],
|
|
result["artist"],
|
|
result["genre"],
|
|
self.deserialize(result["data"]),
|
|
)
|
|
|
|
def get_all_songs(
|
|
self,
|
|
game: GameConstants,
|
|
version: Optional[int] = None,
|
|
) -> List[Song]:
|
|
"""
|
|
Given a game and a version, look up all song/chart combos associated with that game.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
|
|
Returns:
|
|
A list of Song objects detailing the song information for each song.
|
|
"""
|
|
sql = (
|
|
"SELECT version, songid, chart, name, artist, genre, data FROM music "
|
|
"WHERE music.game = :game"
|
|
)
|
|
params: Dict[str, Any] = {"game": game.value}
|
|
if version is not None:
|
|
sql += " AND music.version = :version"
|
|
params["version"] = version
|
|
else:
|
|
sql += " ORDER BY music.version DESC"
|
|
cursor = self.execute(sql, params)
|
|
|
|
all_songs = []
|
|
for result in cursor.fetchall():
|
|
all_songs.append(
|
|
Song(
|
|
game,
|
|
result["version"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["name"],
|
|
result["artist"],
|
|
result["genre"],
|
|
self.deserialize(result["data"]),
|
|
)
|
|
)
|
|
|
|
return all_songs
|
|
|
|
def get_all_versions_of_song(
|
|
self,
|
|
game: GameConstants,
|
|
version: int,
|
|
songid: int,
|
|
songchart: int,
|
|
interested_versions: Optional[List[int]] = None,
|
|
) -> List[Song]:
|
|
"""
|
|
Given a game/version/songid/chart, look up all versions of that song across all game versions.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
songid - Integer representing the ID (from the game) for this song.
|
|
songchart - Integer representing the chart for this song.
|
|
|
|
Returns:
|
|
A list of Song objects representing all song versions.
|
|
"""
|
|
musicid = self.__get_musicid(game, version, songid, songchart)
|
|
sql = (
|
|
"SELECT version, songid, chart, name, artist, genre, data FROM music "
|
|
"WHERE music.id = :musicid"
|
|
)
|
|
if interested_versions is not None:
|
|
sql += f" AND music.version in ({','.join(str(int(v)) for v in interested_versions)})"
|
|
cursor = self.execute(sql, {"musicid": musicid})
|
|
all_songs = []
|
|
for result in cursor.fetchall():
|
|
all_songs.append(
|
|
Song(
|
|
game,
|
|
result["version"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["name"],
|
|
result["artist"],
|
|
result["genre"],
|
|
self.deserialize(result["data"]),
|
|
)
|
|
)
|
|
return all_songs
|
|
|
|
def get_all_scores(
|
|
self,
|
|
game: GameConstants,
|
|
version: Optional[int] = None,
|
|
userid: Optional[UserID] = None,
|
|
songid: Optional[int] = None,
|
|
songchart: Optional[int] = None,
|
|
since: Optional[int] = None,
|
|
until: Optional[int] = None,
|
|
) -> List[Tuple[UserID, Score]]:
|
|
"""
|
|
Look up all of a game's high scores for all users.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
|
|
Returns:
|
|
A list of UserID, Score objects representing all high scores for a game.
|
|
"""
|
|
# First, construct the queries for grabbing the songid/chart
|
|
if version is not None:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score.musicid AND game = :game AND version = :version"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score.musicid AND game = :game AND version = :version"
|
|
else:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
|
|
# Select statement for getting play count
|
|
playselect = "SELECT COUNT(timestamp) FROM score_history WHERE score_history.musicid = score.musicid AND score_history.userid = score.userid"
|
|
|
|
# Now, construct the inner select statement so we can choose which scores we care about
|
|
innerselect = "SELECT DISTINCT(id) FROM music WHERE game = :game"
|
|
if version is not None:
|
|
innerselect = innerselect + " AND version = :version"
|
|
if songid is not None:
|
|
innerselect = innerselect + " AND songid = :songid"
|
|
if songchart is not None:
|
|
innerselect = innerselect + " AND chart = :songchart"
|
|
|
|
# Finally, construct the full query
|
|
sql = (
|
|
"SELECT ({}) AS songid, ({}) AS chart, id AS scorekey, points, timestamp, `update`, lid, data, userid, ({}) AS plays "
|
|
"FROM score WHERE musicid IN ({})"
|
|
).format(songidquery, chartquery, playselect, innerselect)
|
|
|
|
# Now, limit the query
|
|
if userid is not None:
|
|
sql = sql + " AND userid = :userid"
|
|
if since is not None:
|
|
sql = sql + " AND score.update >= :since"
|
|
if until is not None:
|
|
sql = sql + " AND score.update < :until"
|
|
|
|
# Now, query itself
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"userid": userid,
|
|
"songid": songid,
|
|
"songchart": songchart,
|
|
"since": since,
|
|
"until": until,
|
|
},
|
|
)
|
|
|
|
# Objectify result
|
|
scores = []
|
|
for result in cursor.fetchall():
|
|
scores.append(
|
|
(
|
|
UserID(result["userid"]),
|
|
Score(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["update"],
|
|
result["lid"],
|
|
result["plays"],
|
|
self.deserialize(result["data"]),
|
|
),
|
|
)
|
|
)
|
|
|
|
return scores
|
|
|
|
def get_all_records(
|
|
self,
|
|
game: GameConstants,
|
|
version: Optional[int] = None,
|
|
userlist: Optional[List[UserID]] = None,
|
|
locationlist: Optional[List[int]] = None,
|
|
) -> List[Tuple[UserID, Score]]:
|
|
"""
|
|
Look up all of a game's records, only returning the top score for each song. For score ties,
|
|
king-of-the-hill rules are in effect, so for two players with an identical top score, the player
|
|
that got the score last wins. If a list of user IDs is given, we will only look up records pertaining
|
|
to those users. So if another user has a higher record, we will ignore this. This can be used to
|
|
display area-local high scores, etc.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
userlist - List of UserIDs to limit the search to.
|
|
locationlist - A list of location IDs to limit searches to.
|
|
|
|
Returns:
|
|
A list of UserID, Score objects representing all high scores for a game.
|
|
"""
|
|
# First, construct the queries for grabbing the songid/chart
|
|
if version is not None:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score.musicid AND game = :game AND version = :version"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score.musicid AND game = :game AND version = :version"
|
|
else:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
|
|
# Next, get a list of all songs that were played given the input criteria
|
|
musicid_sql = "SELECT DISTINCT(score.musicid) FROM score, music WHERE score.musicid = music.id AND music.game = :game"
|
|
params: Dict[str, Any] = {"game": game.value}
|
|
if version is not None:
|
|
musicid_sql = musicid_sql + " AND music.version = :version"
|
|
params["version"] = version
|
|
|
|
# Figure out where the record was earned
|
|
if locationlist is not None:
|
|
if len(locationlist) == 0:
|
|
# We don't have any locations, but SQL will shit the bed, so lets add a default one.
|
|
locationlist.append(-1)
|
|
location_sql = "AND score.lid IN :locationlist"
|
|
params["locationlist"] = tuple(locationlist)
|
|
else:
|
|
location_sql = ""
|
|
|
|
# Figure out who got the record
|
|
if userlist is not None:
|
|
if len(userlist) == 0:
|
|
# We don't have any users, but SQL will shit the bed, so lets add a fake one.
|
|
userlist.append(UserID(-1))
|
|
user_sql = f"SELECT userid FROM score WHERE score.musicid = played.musicid AND score.userid IN :userlist {location_sql} ORDER BY points DESC, timestamp DESC LIMIT 1"
|
|
params["userlist"] = tuple(userlist)
|
|
else:
|
|
user_sql = f"SELECT userid FROM score WHERE score.musicid = played.musicid {location_sql} ORDER BY points DESC, timestamp DESC LIMIT 1"
|
|
records_sql = (
|
|
f"SELECT ({user_sql}) AS userid, musicid FROM ({musicid_sql}) played"
|
|
)
|
|
|
|
# Now, join it up against the score and music table to grab the info we need
|
|
sql = (
|
|
"SELECT ({}) AS songid, ({}) AS chart, score.points AS points, score.userid AS userid, score.id AS scorekey, score.data AS data, "
|
|
+ "score.timestamp AS timestamp, score.update AS `update`, "
|
|
+ "score.lid AS lid, (select COUNT(score_history.timestamp) FROM score_history WHERE score_history.musicid = score.musicid) AS plays "
|
|
+ "FROM score, ({}) records WHERE records.userid = score.userid AND records.musicid = score.musicid"
|
|
).format(songidquery, chartquery, records_sql)
|
|
cursor = self.execute(sql, params)
|
|
|
|
scores = []
|
|
for result in cursor.fetchall():
|
|
scores.append(
|
|
(
|
|
UserID(result["userid"]),
|
|
Score(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["update"],
|
|
result["lid"],
|
|
result["plays"],
|
|
self.deserialize(result["data"]),
|
|
),
|
|
)
|
|
)
|
|
|
|
return scores
|
|
|
|
def get_attempt_by_key(
|
|
self, game: GameConstants, version: int, key: int
|
|
) -> Optional[Tuple[UserID, Attempt]]:
|
|
"""
|
|
Look up a previous attempt by key.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
key - Integer representing a unique key fetched in a previous Attempt lookup.
|
|
|
|
Returns:
|
|
The optional data stored by the game previously, or None if no score exists.
|
|
"""
|
|
sql = (
|
|
"SELECT music.songid AS songid, music.chart AS chart, score_history.id AS scorekey, score_history.timestamp AS timestamp, score_history.userid AS userid, "
|
|
+ "score_history.lid AS lid, score_history.new_record AS new_record, score_history.points AS points, score_history.data AS data FROM score_history, music "
|
|
+ "WHERE score_history.id = :scorekey AND score_history.musicid = music.id AND music.game = :game AND music.version = :version"
|
|
)
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"scorekey": key,
|
|
},
|
|
)
|
|
if cursor.rowcount != 1:
|
|
# score doesn't exist
|
|
return None
|
|
|
|
result = cursor.fetchone()
|
|
return (
|
|
UserID(result["userid"]),
|
|
Attempt(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["lid"],
|
|
True if result["new_record"] == 1 else False,
|
|
self.deserialize(result["data"]),
|
|
),
|
|
)
|
|
|
|
def get_all_attempts(
|
|
self,
|
|
game: GameConstants,
|
|
version: Optional[int] = None,
|
|
userid: Optional[UserID] = None,
|
|
songid: Optional[int] = None,
|
|
songchart: Optional[int] = None,
|
|
timelimit: Optional[int] = None,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None,
|
|
) -> List[Tuple[Optional[UserID], Attempt]]:
|
|
"""
|
|
Look up all of the attempts to score for a particular game.
|
|
|
|
Parameters:
|
|
game - Enum value representing a game series.
|
|
version - Integer representing which version of the game.
|
|
|
|
Returns:
|
|
A list of UserID, Attempt objects representing all score attempts for a game, sorted newest to oldest attempts.
|
|
"""
|
|
# First, construct the queries for grabbing the songid/chart
|
|
if version is not None:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score_history.musicid AND game = :game AND version = :version"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score_history.musicid AND game = :game AND version = :version"
|
|
else:
|
|
songidquery = "SELECT songid FROM music WHERE music.id = score_history.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
chartquery = "SELECT chart FROM music WHERE music.id = score_history.musicid AND game = :game ORDER BY version DESC LIMIT 1"
|
|
|
|
# Now, construct the inner select statement so we can choose which scores we care about
|
|
innerselect = "SELECT DISTINCT(id) FROM music WHERE game = :game"
|
|
if version is not None:
|
|
innerselect = innerselect + " AND version = :version"
|
|
if songid is not None:
|
|
innerselect = innerselect + " AND songid = :songid"
|
|
if songchart is not None:
|
|
innerselect = innerselect + " AND chart = :songchart"
|
|
|
|
# Finally, construct the full query
|
|
sql = (
|
|
"SELECT ({}) AS songid, ({}) AS chart, id AS scorekey, timestamp, points, new_record, lid, data, userid "
|
|
"FROM score_history WHERE musicid IN ({})"
|
|
).format(songidquery, chartquery, innerselect)
|
|
|
|
# Now, limit the query
|
|
if userid is not None:
|
|
sql = sql + " AND userid = :userid"
|
|
if timelimit is not None:
|
|
sql = sql + " AND timestamp >= :timestamp"
|
|
sql = sql + " ORDER BY timestamp DESC"
|
|
if limit is not None:
|
|
sql = sql + " LIMIT :limit"
|
|
if offset is not None:
|
|
sql = sql + " OFFSET :offset"
|
|
|
|
# Now, query itself
|
|
cursor = self.execute(
|
|
sql,
|
|
{
|
|
"game": game.value,
|
|
"version": version,
|
|
"userid": userid,
|
|
"songid": songid,
|
|
"songchart": songchart,
|
|
"timestamp": timelimit,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
},
|
|
)
|
|
|
|
# Now objectify the attempts
|
|
attempts = []
|
|
for result in cursor.fetchall():
|
|
attempts.append(
|
|
(
|
|
UserID(result["userid"]) if result["userid"] > 0 else None,
|
|
Attempt(
|
|
result["scorekey"],
|
|
result["songid"],
|
|
result["chart"],
|
|
result["points"],
|
|
result["timestamp"],
|
|
result["lid"],
|
|
True if result["new_record"] == 1 else False,
|
|
self.deserialize(result["data"]),
|
|
),
|
|
)
|
|
)
|
|
|
|
return attempts
|