bemaniutils/bemani/frontend/base.py

# vim: set fileencoding=utf-8
import copy
from abc import ABC
from typing import Any, Dict, Iterator, List, Optional, Set, Tuple, cast

from flask_caching import Cache

from bemani.common import GameConstants, Profile, ValidatedDict, ID
from bemani.data import Data, Config, Score, Attempt, Link, Song, UserID, RemoteUser


class FrontendBase(ABC):

    """
    All subclasses should override this attribute with the string
    the game series uses in the DB.
    """

    game: GameConstants

    """
    If a subclass wishes to constrain music searches to a particular
    version, this should be set. If this is left blank, music operations
    such as records and attempts will pull from all versions of the game.
    """
    version: Optional[int] = None

    """
    List of valid chart integers. Should be overridden by the game.
    """
    valid_charts: List[int] = []

    """
    List of valid rival type strings. Should be overridden by the game.
    """
    valid_rival_types: List[str] = []

    def __init__(self, data: Data, config: Config, cache: Cache) -> None:
        self.data = data
        self.config = config
        self.cache = cache

    def make_index(self, songid: int, chart: int) -> str:
        return f"{songid}-{chart}"

    def get_duplicate_id(self, musicid: int, chart: int) -> Optional[Tuple[int, int]]:
        return None

    def format_score(self, userid: UserID, score: Score) -> Dict[str, Any]:
        return {
            "userid": str(userid),
            "songid": score.id,
            "chart": score.chart,
            "plays": score.plays,
            "points": score.points,
        }

    def format_top_score(self, userid: UserID, score: Score) -> Dict[str, Any]:
        return self.format_score(userid, score)

    def format_attempt(self, userid: UserID, attempt: Attempt) -> Dict[str, Any]:
        return {
            "userid": str(userid),
            "songid": attempt.id,
            "chart": attempt.chart,
            "timestamp": attempt.timestamp,
            "raised": attempt.new_record,
            "points": attempt.points,
        }

    def format_rival(self, link: Link, profile: Profile) -> Dict[str, Any]:
        return {
            "type": link.type,
            "userid": str(link.other_userid),
            "remote": RemoteUser.is_remote(link.other_userid),
        }

    def format_profile(
        self, profile: Profile, playstats: ValidatedDict
    ) -> Dict[str, Any]:
        return {
            "name": profile.get_str("name"),
            "extid": ID.format_extid(profile.extid),
            "first_play_time": playstats.get_int("first_play_timestamp"),
            "last_play_time": playstats.get_int("last_play_timestamp"),
        }

    def format_song(self, song: Song) -> Dict[str, Any]:
        return {
            "name": song.name,
            "artist": song.artist,
            "genre": song.genre,
        }

    def merge_song(self, existing: Dict[str, Any], new: Song) -> Dict[str, Any]:
        return existing

    def round_to_ten(self, elems: List[Any]) -> List[Any]:
        num = len(elems)
        if num % 10 == 0:
            return elems
        else:
            return elems[: -(num % 10)]

    def all_games(self) -> Iterator[Tuple[GameConstants, int, str]]:
        """
        Override this to return an interator based on a game series factory.
        """

    def get_all_songs(self, force_db_load: bool = False) -> Dict[int, Dict[str, Any]]:
        if not force_db_load:
            cached_songs = self.cache.get(f"{self.game.value}.sorted_songs")
            if cached_songs is not None:
                # Not sure why mypy insists that this is a str instead of Any.
                return cast(Dict[int, Dict[str, Any]], cached_songs)

        # Find all songs in the game, process notecounts and difficulties
        songs: Dict[int, Dict[str, Any]] = {}
        for song in self.data.local.music.get_all_songs(self.game, self.version):
            if song.chart not in self.valid_charts:
                # No beginner chart support
                continue
            if song.id not in songs:
                songs[song.id] = self.format_song(song)
            else:
                songs[song.id] = self.merge_song(songs[song.id], song)

        self.cache.set(f"{self.game.value}.sorted_songs", songs, timeout=600)
        return songs

    def get_all_player_info(
        self,
        userids: List[UserID],
        limit: Optional[int] = None,
        allow_remote: bool = False,
    ) -> Dict[UserID, Dict[int, Dict[str, Any]]]:
        info: Dict[UserID, Dict[int, Dict[str, Any]]] = {}
        playstats: Dict[UserID, ValidatedDict] = {}

        # Find all versions of the users' profiles, sorted newest to oldest.
        versions = sorted(
            [version for (game, version, name) in self.all_games()], reverse=True
        )
        for userid in userids:
            info[userid] = {}
            userlimit = limit
            for version in versions:
                if allow_remote:
                    profile = self.data.remote.user.get_profile(
                        self.game, version, userid
                    )
                else:
                    profile = self.data.local.user.get_profile(
                        self.game, version, userid
                    )
                if profile is not None:
                    if userid not in playstats:
                        stats = self.data.local.game.get_settings(self.game, userid)
                        if stats is None:
                            stats = ValidatedDict()
                        playstats[userid] = stats
                    info[userid][version] = self.format_profile(
                        profile, playstats[userid]
                    )
                    info[userid][version]["remote"] = RemoteUser.is_remote(userid)
                    # Exit out if we've hit the limit
                    if userlimit is not None:
                        userlimit = userlimit - 1
                        if userlimit == 0:
                            break

        return info

    def get_latest_player_info(
        self, userids: List[UserID]
    ) -> Dict[UserID, Dict[str, Any]]:
        # Grab the latest profile for each user
        all_info = self.get_all_player_info(userids, 1)
        info = {}

        for userid in userids:
            for version in all_info[userid]:
                info[userid] = all_info[userid][version]
                break

        return info

    def get_all_players(self) -> Dict[UserID, Dict[str, Any]]:
        userids: Set[UserID] = set()

        versions = [version for (game, version, name) in self.all_games()]
        for version in versions:
            userids.update(self.data.local.user.get_all_players(self.game, version))

        return self.get_latest_player_info(list(userids))

    def get_network_scores(self, limit: Optional[int] = None) -> Dict[str, Any]:
        userids: List[UserID] = []

        # Find all attempts across all games
        attempts = [
            attempt
            for attempt in self.data.local.music.get_all_attempts(
                game=self.game, version=self.version, limit=limit
            )
            if attempt[0] is not None
        ]
        for attempt in attempts:
            if attempt[0] not in userids:
                userids.append(attempt[0])

        return {
            "attempts": sorted(
                [self.format_attempt(attempt[0], attempt[1]) for attempt in attempts],
                reverse=True,
                key=lambda attempt: (
                    attempt["timestamp"],
                    attempt["songid"],
                    attempt["chart"],
                ),
            ),
            "players": self.get_latest_player_info(userids),
        }

    def get_network_records(self) -> Dict[str, Any]:
        records: Dict[str, Tuple[UserID, Score]] = {}
        userids: List[UserID] = []

        # Find all high-scores across all games
        highscores = self.data.local.music.get_all_records(
            game=self.game, version=self.version
        )
        for score in highscores:
            index = self.make_index(score[1].id, score[1].chart)
            if index not in records:
                records[index] = score
                if score[0] not in userids:
                    userids.append(score[0])
            # Also take care of duplicate IDs (revivals, omnimix, etc)
            alternate = self.get_duplicate_id(score[1].id, score[1].chart)
            if alternate is not None:
                altid, altchart = alternate
                index = self.make_index(altid, altchart)
                if index not in records:
                    newscore = copy.deepcopy(score)
                    newscore[1].id = altid
                    newscore[1].chart = altchart
                    records[index] = newscore

        return {
            "records": [
                self.format_score(records[index][0], records[index][1])
                for index in records
            ],
            "players": self.get_latest_player_info(userids),
        }

    def get_scores(
        self, userid: UserID, limit: Optional[int] = None
    ) -> List[Dict[str, Any]]:
        # Find all attempts across all games
        attempts = [
            attempt
            for attempt in self.data.local.music.get_all_attempts(
                game=self.game, version=self.version, userid=userid, limit=limit
            )
            if attempt[0] is not None
        ]

        return sorted(
            [self.format_attempt(None, attempt[1]) for attempt in attempts],
            reverse=True,
            key=lambda attempt: (
                attempt["timestamp"],
                attempt["songid"],
                attempt["chart"],
            ),
        )

    def get_records(self, userid: UserID) -> List[Dict[str, Any]]:
        records: Dict[str, Tuple[UserID, Score]] = {}

        # Find all high-scores across all games
        highscores = self.data.local.music.get_all_scores(
            game=self.game, version=self.version, userid=userid
        )
        for score in highscores:
            index = self.make_index(score[1].id, score[1].chart)
            if index not in records:
                records[index] = score
            else:
                current_score = records[index][1].points
                current_plays = records[index][1].plays
                new_score = score[1].points
                new_plays = score[1].plays
                if new_score > current_score:
                    records[index] = score
                    records[index][1].plays += current_plays
                else:
                    records[index][1].plays += new_plays

        # Copy over records to duplicate IDs, such as revivals
        indexes = [index for index in records]
        for index in indexes:
            alternate = self.get_duplicate_id(
                records[index][1].id, records[index][1].chart
            )
            if alternate is not None:
                altid, altchart = alternate
                newindex = self.make_index(altid, altchart)
                if newindex not in records:
                    newscore = copy.deepcopy(score)
                    newscore[1].id = altid
                    newscore[1].chart = altchart
                    records[newindex] = newscore

        return [self.format_score(None, records[index][1]) for index in records]

    def get_top_scores(self, musicid: int) -> Dict[str, Any]:
        scores = self.data.local.music.get_all_scores(
            game=self.game, version=self.version, songid=musicid
        )
        userids: List[UserID] = []
        for score in scores:
            if score[1].chart not in self.valid_charts:
                # No beginner chart support
                continue
            if score[0] not in userids:
                userids.append(score[0])

        for score in scores:
            # See if this is a legacy ID
            if score[1].id != musicid:
                alternative = self.get_duplicate_id(score[1].id, score[1].chart)
                if alternative is None:
                    continue

                oldid, oldchart = alternative
                if oldid == musicid:
                    score[1].id = oldid
                    score[1].chart = oldchart

        return {
            "topscores": [
                self.format_top_score(score[0], score[1])
                for score in scores
                if score[1].chart in self.valid_charts
            ],
            "players": self.get_latest_player_info(userids),
        }

    def get_rivals(
        self, userid: UserID
    ) -> Tuple[
        Dict[int, List[Dict[str, Any]]], Dict[UserID, Dict[int, Dict[str, Any]]]
    ]:
        rivals = {}
        userids = set()
        versions = [version for (game, version, name) in self.all_games()]
        profiles = {}
        for version in versions:
            profile = self.data.local.user.get_profile(self.game, version, userid)
            if profile is None:
                # No profile for this version, so no rivals either.
                continue
            profiles[version] = profile
            rivals[version] = [
                link
                for link in self.data.local.user.get_links(self.game, version, userid)
                if link.type in self.valid_rival_types
            ]
            for rival in rivals[version]:
                userids.add(rival.other_userid)

        return (
            {
                version: [
                    self.format_rival(rival, profiles[version])
                    for rival in rivals[version]
                ]
                for version in rivals
            },
            self.get_all_player_info(list(userids), allow_remote=True),
        )