artemis/titles/ongeki/bright.py

from datetime import date, datetime, timedelta
from typing import Any, Dict
from random import randint
import pytz
import json

from core.config import CoreConfig
from titles.ongeki.base import OngekiBase
from titles.ongeki.const import OngekiConstants
from titles.ongeki.config import OngekiConfig


class OngekiBright(OngekiBase):
    def __init__(self, core_cfg: CoreConfig, game_cfg: OngekiConfig) -> None:
        super().__init__(core_cfg, game_cfg)
        self.version = OngekiConstants.VER_ONGEKI_BRIGHT

    async def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
        ret = await super().handle_get_game_setting_api_request(data)
        ret["gameSetting"]["dataVersion"] = "1.30.00"
        ret["gameSetting"]["onlineDataVersion"] = "1.30.00"
        return ret

    async def handle_cm_get_user_data_api_request(self, data: Dict) -> Dict:
        # check for a bright profile
        p = await self.data.profile.get_profile_data(data["userId"], self.version)
        if p is None:
            return {}

        cards = await self.data.card.get_user_cards(data["userId"])
        if cards is None or len(cards) == 0:
            # This should never happen
            self.logger.error(
                f"handle_get_user_data_api_request: Internal error - No cards found for user id {data['userId']}"
            )
            return {}

        # get the dict representation of the row so we can modify values
        user_data = p._asdict()

        # remove the values the game doesn't want
        user_data.pop("id")
        user_data.pop("user")
        user_data.pop("version")

        # add access code that we don't store
        user_data["accessCode"] = cards[0]["access_code"]

        # add the compatible card maker version from config
        card_maker_ver = self.game_cfg.version.version(self.version)
        if card_maker_ver and card_maker_ver.get("card_maker"):
            # Card Maker 1.30 = 1.30.01+
            # Card Maker 1.35 = 1.35.03+
            user_data["compatibleCmVersion"] = card_maker_ver.get("card_maker")

        return {"userId": data["userId"], "userData": user_data}

    async def handle_printer_login_api_request(self, data: Dict):
        return {"returnCode": 1}

    async def handle_printer_logout_api_request(self, data: Dict):
        return {"returnCode": 1}

    async def handle_cm_get_user_card_api_request(self, data: Dict) -> Dict:
        user_cards = await self.data.item.get_cards(data["userId"])
        if user_cards is None:
            return {}

        max_ct = data["maxCount"]
        next_idx = data["nextIndex"]
        start_idx = next_idx
        end_idx = max_ct + start_idx

        if len(user_cards[start_idx:]) > max_ct:
            next_idx += max_ct
        else:
            next_idx = -1

        card_list = []
        for card in user_cards:
            tmp = card._asdict()
            tmp.pop("id")
            tmp.pop("user")
            card_list.append(tmp)

        return {
            "userId": data["userId"],
            "length": len(card_list[start_idx:end_idx]),
            "nextIndex": next_idx,
            "userCardList": card_list[start_idx:end_idx],
        }

    async def handle_cm_get_user_character_api_request(self, data: Dict) -> Dict:
        user_characters = await self.data.item.get_characters(data["userId"])
        if user_characters is None:
            return {
                "userId": data["userId"],
                "length": 0,
                "nextIndex": 0,
                "userCharacterList": [],
            }

        max_ct = data["maxCount"]
        next_idx = data["nextIndex"]
        start_idx = next_idx
        end_idx = max_ct + start_idx

        if len(user_characters[start_idx:]) > max_ct:
            next_idx += max_ct
        else:
            next_idx = -1

        character_list = []
        for character in user_characters:
            tmp = character._asdict()
            tmp.pop("id")
            tmp.pop("user")
            character_list.append(tmp)

        return {
            "userId": data["userId"],
            "length": len(character_list[start_idx:end_idx]),
            "nextIndex": next_idx,
            "userCharacterList": character_list[start_idx:end_idx],
        }

    async def handle_get_user_gacha_api_request(self, data: Dict) -> Dict:
        user_gachas = await self.data.item.get_user_gachas(data["userId"])
        if user_gachas is None:
            return {"userId": data["userId"], "length": 0, "userGachaList": []}

        user_gacha_list = []
        for gacha in user_gachas:
            tmp = gacha._asdict()
            tmp.pop("id")
            tmp.pop("user")
            tmp["dailyGachaDate"] = datetime.strftime(tmp["dailyGachaDate"], "%Y-%m-%d")
            user_gacha_list.append(tmp)

        return {
            "userId": data["userId"],
            "length": len(user_gacha_list),
            "userGachaList": user_gacha_list,
        }

    async def handle_cm_get_user_item_api_request(self, data: Dict) -> Dict:
        return self.handle_get_user_item_api_request(data)

    async def handle_cm_get_user_gacha_supply_api_request(self, data: Dict) -> Dict:
        # not used for now? not sure what it even does
        user_gacha_supplies = await self.data.item.get_user_gacha_supplies(data["userId"])
        if user_gacha_supplies is None:
            return {"supplyId": 1, "length": 0, "supplyCardList": []}

        supply_list = [gacha["cardId"] for gacha in user_gacha_supplies]

        return {
            "supplyId": 1,
            "length": len(supply_list),
            "supplyCardList": supply_list,
        }

    async def handle_get_game_gacha_api_request(self, data: Dict) -> Dict:
        """
        returns all current active banners (gachas)
        "Select Gacha" requires maxSelectPoint set and isCeiling set to 1
        """
        game_gachas = []
        # for every gacha_id in the OngekiConfig, grab the banner from the db
        for gacha_id in self.game_cfg.gachas.enabled_gachas:
            game_gacha = await self.data.static.get_gacha(self.version, gacha_id)
            if game_gacha:
                game_gachas.append(game_gacha)

        # clean the database rows
        game_gacha_list = []
        for gacha in game_gachas:
            tmp = gacha._asdict()
            tmp.pop("id")
            tmp.pop("version")
            tmp["startDate"] = datetime.strftime(tmp["startDate"], "%Y-%m-%d %H:%M:%S")
            tmp["endDate"] = datetime.strftime(tmp["endDate"], "%Y-%m-%d %H:%M:%S")
            tmp["noticeStartDate"] = datetime.strftime(
                tmp["noticeStartDate"], "%Y-%m-%d %H:%M:%S"
            )
            tmp["noticeEndDate"] = datetime.strftime(
                tmp["noticeEndDate"], "%Y-%m-%d %H:%M:%S"
            )
            tmp["convertEndDate"] = datetime.strftime(
                tmp["convertEndDate"], "%Y-%m-%d %H:%M:%S"
            )

            # make sure to only show gachas for the current version
            # so only up to bright, 1140 is the first bright memory gacha
            if self.version == OngekiConstants.VER_ONGEKI_BRIGHT_MEMORY:
                game_gacha_list.append(tmp)
            elif (
                self.version == OngekiConstants.VER_ONGEKI_BRIGHT
                and tmp["gachaId"] < 1140
            ):
                game_gacha_list.append(tmp)

        return {
            "length": len(game_gacha_list),
            "gameGachaList": game_gacha_list,
            # no clue
            "registIdList": [],
        }

    async def handle_roll_gacha_api_request(self, data: Dict) -> Dict:
        """
        Handle a gacha roll API request
        """
        gacha_id = data["gachaId"]
        num_rolls = data["times"]
        # change_rate is the 5 gacha rool SR gurantee once a week
        change_rate = data["changeRate"]
        # SSR book which guarantees a SSR card, itemKind=15, itemId=1
        book_used = data["bookUseCount"]
        if num_rolls not in {1, 5, 11}:
            return {}

        # https://gamerch.com/ongeki/entry/462978

        # 77% chance of gett ing a R card
        # 20% chance of getting a SR card
        # 3% chance of getting a SSR card
        rarity = [1 for _ in range(77)]
        rarity += [2 for _ in range(20)]
        rarity += [3 for _ in range(3)]

        # gachaId 1011 is "無料ガチャ" (free gacha), which requires GatchaTickets
        # itemKind=11, itemId=1 and appearenty sucks
        # 94% chance of getting a R card
        # 5% chance of getting a SR card
        # 1% chance of getting a SSR card
        if gacha_id == 1011:
            rarity = [1 for _ in range(94)]
            rarity += [2 for _ in range(5)]
            rarity += [3 for _ in range(1)]

        # gachaId 1012 is "無料ガチャ（SR確定）" (SR confirmed! free gacha), which
        # requires GatchaTickets itemKind=11, itemId=4 and always guarantees
        # a SR card or higher
        # 92% chance of getting a SR card
        # 8% chance of getting a SSR card
        elif gacha_id == 1012:
            rarity = [2 for _ in range(92)]
            rarity += [3 for _ in range(8)]

        assert len(rarity) == 100

        # uniform distribution to get the rarity of the card
        rolls = [rarity[randint(0, len(rarity) - 1)] for _ in range(num_rolls)]

        # if SSR book used, make sure you always get one SSR
        if book_used == 1:
            if rolls.count(3) == 0:
                # if there is no SSR, re-roll
                return self.handle_roll_gacha_api_request(data)
        # make sure that 11 rolls always have at least 1 SR or SSR
        elif (num_rolls == 5 and change_rate is True) or num_rolls == 11:
            if rolls.count(2) == 0 and rolls.count(3) == 0:
                # if there is no SR or SSR, re-roll
                return self.handle_roll_gacha_api_request(data)

        # get a list of cards for each rarity
        cards_r = await self.data.static.get_cards_by_rarity(self.version, 1)
        cards_sr, cards_ssr = [], []

        # free gachas are only allowed to get their specific cards! (R irrelevant)
        if gacha_id in {1011, 1012}:
            gacha_cards = await self.data.static.get_gacha_cards(gacha_id)
            for card in gacha_cards:
                if card["rarity"] == 3:
                    cards_sr.append({"cardId": card["cardId"], "rarity": 2})
                elif card["rarity"] == 4:
                    cards_ssr.append({"cardId": card["cardId"], "rarity": 3})
        else:
            cards_sr = await self.data.static.get_cards_by_rarity(self.version, 2)
            cards_ssr = await self.data.static.get_cards_by_rarity(self.version, 3)

            # get the promoted cards for that gacha and add them multiple
            # times to increase chances by factor chances
            chances = 10

            gacha_cards = await self.data.static.get_gacha_cards(gacha_id)
            for card in gacha_cards:
                # make sure to add the cards to the corresponding rarity
                if card["rarity"] == 2:
                    cards_r += [{"cardId": card["cardId"], "rarity": 1}] * chances
                if card["rarity"] == 3:
                    cards_sr += [{"cardId": card["cardId"], "rarity": 2}] * chances
                elif card["rarity"] == 4:
                    cards_ssr += [{"cardId": card["cardId"], "rarity": 3}] * chances

        # get the card id for each roll
        rolled_cards = []
        for i in range(len(rolls)):
            if rolls[i] == 1:
                rolled_cards.append(cards_r[randint(0, len(cards_r) - 1)])
            elif rolls[i] == 2:
                rolled_cards.append(cards_sr[randint(0, len(cards_sr) - 1)])
            elif rolls[i] == 3:
                rolled_cards.append(cards_ssr[randint(0, len(cards_ssr) - 1)])

        game_gacha_card_list = []
        for card in rolled_cards:
            game_gacha_card_list.append(
                {
                    "gachaId": data["gachaId"],
                    "cardId": card["cardId"],
                    # +1 because Card Maker is weird
                    "rarity": card["rarity"] + 1,
                    "weight": 1,
                    "isPickup": False,
                    "isSelect": False,
                }
            )

        return {
            "length": len(game_gacha_card_list),
            "gameGachaCardList": game_gacha_card_list,
        }

    async def handle_cm_upsert_user_gacha_api_request(self, data: Dict):
        upsert = data["cmUpsertUserGacha"]
        user_id = data["userId"]

        gacha_id = data["gachaId"]
        gacha_count = data["gachaCnt"]
        play_date = datetime.strptime(data["playDate"][:10], "%Y-%m-%d")
        select_point = data["selectPoint"]

        total_gacha_count, ceiling_gacha_count = 0, 0
        # 0 = can still use Gacha Select, 1 = already used Gacha Select
        use_select_point = 0
        daily_gacha_cnt, five_gacha_cnt, eleven_gacha_cnt = 0, 0, 0
        daily_gacha_date = datetime.strptime("2000-01-01", "%Y-%m-%d")

        # check if the user previously rolled the exact same gacha
        user_gacha = await self.data.item.get_user_gacha(user_id, gacha_id)
        if user_gacha:
            total_gacha_count = user_gacha["totalGachaCnt"]
            ceiling_gacha_count = user_gacha["ceilingGachaCnt"]
            daily_gacha_cnt = user_gacha["dailyGachaCnt"]
            five_gacha_cnt = user_gacha["fiveGachaCnt"]
            eleven_gacha_cnt = user_gacha["elevenGachaCnt"]
            # if the Gacha Select has been used, make sure to keep it
            if user_gacha["useSelectPoint"] == 1:
                use_select_point = 1
            # parse just the year, month and date
            daily_gacha_date = user_gacha["dailyGachaDate"]

        # if the saved dailyGachaDate is different from the roll,
        # reset dailyGachaCnt and change the date
        if daily_gacha_date != play_date:
            daily_gacha_date = play_date
            daily_gacha_cnt = 0

        await self.data.item.put_user_gacha(
            user_id,
            gacha_id,
            totalGachaCnt=total_gacha_count + gacha_count,
            ceilingGachaCnt=ceiling_gacha_count + gacha_count,
            selectPoint=select_point,
            useSelectPoint=use_select_point,
            dailyGachaCnt=daily_gacha_cnt + gacha_count,
            fiveGachaCnt=five_gacha_cnt + 1 if gacha_count == 5 else five_gacha_cnt,
            elevenGachaCnt=eleven_gacha_cnt + 1
            if gacha_count == 11
            else eleven_gacha_cnt,
            dailyGachaDate=daily_gacha_date,
        )

        if "userData" in upsert and len(upsert["userData"]) > 0:
            # check if the profile is a bright memory profile
            p = await self.data.profile.get_profile_data(data["userId"], self.version)
            if p is not None:
                # save the bright memory profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )
            else:
                # save the bright profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )

        if "userCharacterList" in upsert:
            for x in upsert["userCharacterList"]:
                await self.data.item.put_character(user_id, x)

        if "userItemList" in upsert:
            for x in upsert["userItemList"]:
                await self.data.item.put_item(user_id, x)

        if "userCardList" in upsert:
            for x in upsert["userCardList"]:
                await self.data.item.put_card(user_id, x)

        # TODO?
        # if "gameGachaCardList" in upsert:
        #    for x in upsert["gameGachaCardList"]:

        return {"returnCode": 1, "apiName": "CMUpsertUserGachaApi"}

    async def handle_cm_upsert_user_select_gacha_api_request(self, data: Dict) -> Dict:
        upsert = data["cmUpsertUserSelectGacha"]
        user_id = data["userId"]

        if "userData" in upsert and len(upsert["userData"]) > 0:
            # check if the profile is a bright memory profile
            p = await self.data.profile.get_profile_data(data["userId"], self.version)
            if p is not None:
                # save the bright memory profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )
            else:
                # save the bright profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )

        if "userCharacterList" in upsert:
            for x in upsert["userCharacterList"]:
                await self.data.item.put_character(user_id, x)

        if "userCardList" in upsert:
            for x in upsert["userCardList"]:
                await self.data.item.put_card(user_id, x)

        if "selectGachaLogList" in data:
            for x in data["selectGachaLogList"]:
                await self.data.item.put_user_gacha(
                    user_id,
                    x["gachaId"],
                    selectPoint=0,
                    useSelectPoint=x["useSelectPoint"],
                )

        return {"returnCode": 1, "apiName": "cmUpsertUserSelectGacha"}

    async def handle_get_game_gacha_card_by_id_api_request(self, data: Dict) -> Dict:
        game_gacha_cards = await self.data.static.get_gacha_cards(data["gachaId"])
        if game_gacha_cards == []:
            # fallback to be at least able to select that gacha
            return {
                "gachaId": data["gachaId"],
                "length": 6,
                "isPickup": False,
                "gameGachaCardList": [
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 100984,
                        "rarity": 4,
                        "weight": 1,
                        "isPickup": False,
                        "isSelect": True,
                    },
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 100997,
                        "rarity": 3,
                        "weight": 2,
                        "isPickup": False,
                        "isSelect": True,
                    },
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 100998,
                        "rarity": 3,
                        "weight": 2,
                        "isPickup": False,
                        "isSelect": True,
                    },
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 101020,
                        "rarity": 2,
                        "weight": 3,
                        "isPickup": False,
                        "isSelect": True,
                    },
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 101021,
                        "rarity": 2,
                        "weight": 3,
                        "isPickup": False,
                        "isSelect": True,
                    },
                    {
                        "gachaId": data["gachaId"],
                        "cardId": 101022,
                        "rarity": 2,
                        "weight": 3,
                        "isPickup": False,
                        "isSelect": True,
                    },
                ],
                "emissionList": [],
                "afterCalcList": [],
                "ssrBookCalcList": [],
            }

        game_gacha_card_list = []
        for gacha_card in game_gacha_cards:
            tmp = gacha_card._asdict()
            tmp.pop("id")
            game_gacha_card_list.append(tmp)

        return {
            "gachaId": data["gachaId"],
            "length": len(game_gacha_card_list),
            "isPickup": False,
            "gameGachaCardList": game_gacha_card_list,
            # again no clue
            "emissionList": [],
            "afterCalcList": [],
            "ssrBookCalcList": [],
        }

    async def handle_get_game_theater_api_request(self, data: Dict) -> Dict:
        """
        shows a banner after every print, not sure what its used for
        """

        """
        return {
            "length": 1,
            "gameTheaterList": [{
                "theaterId": 1,
                "theaterName": "theaterName",
                "startDate": "2018-01-01 00:00:00.0",
                "endDate": "2038-01-01 00:00:00.0",
                "gameSubTheaterList": [{
                    "theaterId": 1,
                    "id": 2,
                    "no": 4
                }]
            }
            ],
            "registIdList": []
        }
        """

        return {"length": 0, "gameTheaterList": [], "registIdList": []}

    async def handle_cm_upsert_user_print_playlog_api_request(self, data: Dict) -> Dict:
        return {
            "returnCode": 1,
            "orderId": 0,
            "serialId": "11111111111111111111",
            "apiName": "CMUpsertUserPrintPlaylogApi",
        }

    async def handle_cm_upsert_user_printlog_api_request(self, data: Dict) -> Dict:
        return {
            "returnCode": 1,
            "orderId": 0,
            "serialId": "11111111111111111111",
            "apiName": "CMUpsertUserPrintlogApi",
        }

    async def handle_cm_upsert_user_print_api_request(self, data: Dict) -> Dict:
        user_print_detail = data["userPrintDetail"]

        # generate random serial id
        serial_id = "".join([str(randint(0, 9)) for _ in range(20)])

        # not needed because are either zero or unset
        user_print_detail.pop("orderId")
        user_print_detail.pop("printNumber")
        user_print_detail.pop("serialId")
        user_print_detail["printDate"] = datetime.strptime(
            user_print_detail["printDate"], "%Y-%m-%d"
        )

        # add the entry to the user print table with the random serialId
        await self.data.item.put_user_print_detail(
            data["userId"], serial_id, user_print_detail
        )

        return {
            "returnCode": 1,
            "serialId": serial_id,
            "apiName": "CMUpsertUserPrintApi",
        }

    async def handle_cm_upsert_user_all_api_request(self, data: Dict) -> Dict:
        upsert = data["cmUpsertUserAll"]
        user_id = data["userId"]

        if "userData" in upsert and len(upsert["userData"]) > 0:
            # check if the profile is a bright memory profile
            p = await self.data.profile.get_profile_data(data["userId"], self.version)
            if p is not None:
                # save the bright memory profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )
            else:
                # save the bright profile
                await self.data.profile.put_profile_data(
                    user_id, self.version, upsert["userData"][0]
                )

        if "userActivityList" in upsert:
            for act in upsert["userActivityList"]:
                await self.data.profile.put_profile_activity(
                    user_id,
                    act["kind"],
                    act["id"],
                    act["sortNumber"],
                    act["param1"],
                    act["param2"],
                    act["param3"],
                    act["param4"],
                )

        if "userItemList" in upsert:
            for x in upsert["userItemList"]:
                await self.data.item.put_item(user_id, x)

        if "userCardList" in upsert:
            for x in upsert["userCardList"]:
                await self.data.item.put_card(user_id, x)

        return {"returnCode": 1, "apiName": "cmUpsertUserAll"}