artemis/titles/pokken/base.py

from datetime import datetime, timedelta
import json, logging
from typing import Any, Dict, List
import random

from core.data import Data
from core import CoreConfig
from .config import PokkenConfig
from .proto import jackal_pb2
from .database import PokkenData
from .const import PokkenConstants


class PokkenBase:
    def __init__(self, core_cfg: CoreConfig, game_cfg: PokkenConfig) -> None:
        self.core_cfg = core_cfg
        self.game_cfg = game_cfg
        self.version = 0
        self.logger = logging.getLogger("pokken")
        self.data = PokkenData(core_cfg)
        self.SUPPORT_SET_NONE = 4294967295

    def handle_noop(self, request: Any) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = request.type

        return res.SerializeToString()

    def handle_ping(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.PING

        return res.SerializeToString()

    def handle_register_pcb(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.REGISTER_PCB
        self.logger.info(f"Register PCB {request.register_pcb.pcb_id}")

        regist_pcb = jackal_pb2.RegisterPcbResponseData()
        regist_pcb.server_time = int(datetime.now().timestamp())
        biwa_setting = {
            "MatchingServer": {
                "host": f"https://{self.game_cfg.server.hostname}",
                "port": self.game_cfg.ports.game,
                "url": "/SDAK/100/matching",
            },
            "StunServer": {
                "addr": self.game_cfg.server.stun_server_host,
                "port": self.game_cfg.server.stun_server_port,
            },
            "TurnServer": {
                "addr": self.game_cfg.server.stun_server_host,
                "port": self.game_cfg.server.stun_server_port,
            },
            "AdmissionUrl": f"ws://{self.game_cfg.server.hostname}:{self.game_cfg.ports.admission}",
            "locationId": 123, # FIXME: Get arcade's ID from the database
            "logfilename": "JackalMatchingLibrary.log",
            "biwalogfilename": "./biwa.log",
        }
        regist_pcb.bnp_baseuri = f"{self.core_cfg.title.hostname}/bna"
        regist_pcb.biwa_setting = json.dumps(biwa_setting)

        res.register_pcb.CopyFrom(regist_pcb)

        return res.SerializeToString()

    def handle_save_ads(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SAVE_ADS

        return res.SerializeToString()

    def handle_save_client_log(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SAVE_CLIENT_LOG

        return res.SerializeToString()

    def handle_check_diagnosis(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.CHECK_DIAGNOSIS

        return res.SerializeToString()

    def handle_load_client_settings(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.LOAD_CLIENT_SETTINGS
        settings = jackal_pb2.LoadClientSettingsResponseData()

        # TODO: Make configurable
        settings.money_magnification = 1
        settings.continue_bonus_exp = 100
        settings.continue_fight_money = 100
        settings.event_bonus_exp = 100
        settings.level_cap = 999
        settings.op_movie_flag = 0xFFFFFFFF
        settings.lucky_bonus_rate = 1
        settings.fail_support_num = 10
        settings.chara_open_flag = 0xFFFFFFFF
        settings.chara_open_date = int(datetime.now().timestamp() / 1000)
        settings.chara_pre_open_date = int(datetime.now().timestamp() / 1000)
        settings.search_id = 123
        res.load_client_settings.CopyFrom(settings)

        return res.SerializeToString()

    def handle_load_ranking(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.LOAD_RANKING
        ranking = jackal_pb2.LoadRankingResponseData()

        ranking.ranking_id = 1
        ranking.ranking_start = 0
        ranking.ranking_end = 1
        ranking.event_end = True
        ranking.modify_date = int(datetime.now().timestamp() / 1000)
        res.load_ranking.CopyFrom(ranking)
        return res.SerializeToString()

    def handle_load_user(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.LOAD_USER
        access_code = request.load_user.access_code
        load_usr = jackal_pb2.LoadUserResponseData()
        user_id = self.data.card.get_user_id_from_card(access_code)

        if user_id is None and self.game_cfg.server.auto_register:
            user_id = self.data.user.create_user()
            card_id = self.data.card.create_card(user_id, access_code)

            self.logger.info(
                f"Register new card {access_code} (UserId {user_id}, CardId {card_id})"
            )

        elif user_id is None:
            self.logger.info(f"Registration of card {access_code} blocked!")
            res.load_user.CopyFrom(load_usr)
            return res.SerializeToString()

        """ 
        TODO: Add repeated values
        tutorial_progress_flag
        rankmatch_progress
        support_pokemon_list
        support_set_1
        support_set_2
        support_set_3
        aid_skill_list
        achievement_flag
        event_achievement_flag
        event_achievement_param
        """
        profile = self.data.profile.get_profile(user_id)
        load_usr.commidserv_result = 1
        load_usr.load_hash = 1
        load_usr.cardlock_status = False
        load_usr.banapass_id = user_id
        load_usr.access_code = access_code
        load_usr.precedent_release_flag = 0xFFFFFFFF

        if profile is None:
            profile_id = self.data.profile.create_profile(user_id)
            profile_dict = {"id": profile_id, "user": user_id}
            pokemon_data = []
            tutorial_progress = []
            rankmatch_progress = []
            achievement_flag = []
            event_achievement_flag = []
            event_achievement_param = []
            load_usr.new_card_flag = True

        else:
            profile_dict = {k: v for k, v in profile._asdict().items() if v is not None}
            self.logger.info(
                f"Card-in user {user_id} (Trainer name {profile_dict.get('trainer_name', '')})"
            )
            pokemon_data = self.data.profile.get_all_pokemon_data(user_id)
            tutorial_progress = []
            rankmatch_progress = []
            achievement_flag = []
            event_achievement_flag = []
            event_achievement_param = []
            load_usr.new_card_flag = False

        load_usr.navi_newbie_flag = profile_dict.get("navi_newbie_flag", True)
        load_usr.navi_enable_flag = profile_dict.get("navi_enable_flag", True)
        load_usr.pad_vibrate_flag = profile_dict.get("pad_vibrate_flag", True)
        load_usr.home_region_code = profile_dict.get("home_region_code", 0)
        load_usr.home_loc_name = profile_dict.get("home_loc_name", "")
        load_usr.pref_code = profile_dict.get("pref_code", 0)
        load_usr.trainer_name = profile_dict.get(
            "trainer_name", f"Newb{str(user_id).zfill(4)}"
        )
        load_usr.trainer_rank_point = profile_dict.get("trainer_rank_point", 0)
        load_usr.wallet = profile_dict.get("wallet", 0)
        load_usr.fight_money = profile_dict.get("fight_money", 0)
        load_usr.score_point = profile_dict.get("score_point", 0)
        load_usr.grade_max_num = profile_dict.get("grade_max_num", 0)
        load_usr.extra_counter = profile_dict.get("extra_counter", 0)
        load_usr.total_play_days = profile_dict.get("total_play_days", 0)
        load_usr.play_date_time = profile_dict.get("play_date_time", 0)
        load_usr.lucky_box_fail_num = profile_dict.get("lucky_box_fail_num", 0)
        load_usr.event_reward_get_flag = profile_dict.get("event_reward_get_flag", 0)
        load_usr.rank_pvp_all = profile_dict.get("rank_pvp_all", 0)
        load_usr.rank_pvp_loc = profile_dict.get("rank_pvp_loc", 0)
        load_usr.rank_cpu_all = profile_dict.get("rank_cpu_all", 0)
        load_usr.rank_cpu_loc = profile_dict.get("rank_cpu_loc", 0)
        load_usr.rank_event = profile_dict.get("rank_event", 0)
        load_usr.awake_num = profile_dict.get("awake_num", 0)
        load_usr.use_support_num = profile_dict.get("use_support_num", 0)
        load_usr.rankmatch_flag = profile_dict.get("rankmatch_flag", 0)
        load_usr.rankmatch_max = profile_dict.get("rankmatch_max", 0)
        load_usr.rankmatch_success = profile_dict.get("rankmatch_success", 0)
        load_usr.beat_num = profile_dict.get("beat_num", 0)
        load_usr.title_text_id = profile_dict.get("title_text_id", 0)
        load_usr.title_plate_id = profile_dict.get("title_plate_id", 0)
        load_usr.title_decoration_id = profile_dict.get("title_decoration_id", 0)
        load_usr.navi_trainer = profile_dict.get("navi_trainer", 0)
        load_usr.navi_version_id = profile_dict.get("navi_version_id", 0)
        load_usr.aid_skill = profile_dict.get("aid_skill", 0)
        load_usr.comment_text_id = profile_dict.get("comment_text_id", 0)
        load_usr.comment_word_id = profile_dict.get("comment_word_id", 0)
        load_usr.latest_use_pokemon = profile_dict.get("latest_use_pokemon", 0)
        load_usr.ex_ko_num = profile_dict.get("ex_ko_num", 0)
        load_usr.wko_num = profile_dict.get("wko_num", 0)
        load_usr.timeup_win_num = profile_dict.get("timeup_win_num", 0)
        load_usr.cool_ko_num = profile_dict.get("cool_ko_num", 0)
        load_usr.perfect_ko_num = profile_dict.get("perfect_ko_num", 0)
        load_usr.record_flag = profile_dict.get("record_flag", 0)
        load_usr.site_register_status = profile_dict.get("site_register_status", 0)
        load_usr.continue_num = profile_dict.get("continue_num", 0)

        load_usr.avatar_body = profile_dict.get("avatar_body", 0)
        load_usr.avatar_gender = profile_dict.get("avatar_gender", 0)
        load_usr.avatar_background = profile_dict.get("avatar_background", 0)
        load_usr.avatar_head = profile_dict.get("avatar_head", 0)
        load_usr.avatar_battleglass = profile_dict.get("avatar_battleglass", 0)
        load_usr.avatar_face0 = profile_dict.get("avatar_face0", 0)
        load_usr.avatar_face1 = profile_dict.get("avatar_face1", 0)
        load_usr.avatar_face2 = profile_dict.get("avatar_face2", 0)
        load_usr.avatar_bodyall = profile_dict.get("avatar_bodyall", 0)
        load_usr.avatar_wear = profile_dict.get("avatar_wear", 0)
        load_usr.avatar_accessory = profile_dict.get("avatar_accessory", 0)
        load_usr.avatar_stamp = profile_dict.get("avatar_stamp", 0)

        load_usr.event_state = profile_dict.get("event_state", 0)
        load_usr.event_id = profile_dict.get("event_id", 0)
        load_usr.sp_bonus_category_id_1 = profile_dict.get("sp_bonus_category_id_1", 0)
        load_usr.sp_bonus_key_value_1 = profile_dict.get("sp_bonus_key_value_1", 0)
        load_usr.sp_bonus_category_id_2 = profile_dict.get("sp_bonus_category_id_2", 0)
        load_usr.sp_bonus_key_value_2 = profile_dict.get("sp_bonus_key_value_2", 0)
        load_usr.last_play_event_id = profile_dict.get("last_play_event_id", 0)

        if pokemon_data is not None:
            for pkmn in pokemon_data:
                pkmn_d = pkmn._asdict()
                pkm = jackal_pb2.LoadUserResponseData.PokemonData()

                pkm.char_id = pkmn_d.get('char_id', 0)
                pkm.illustration_book_no = pkmn_d.get('illustration_book_no', 0)
                pkm.pokemon_exp = pkmn_d.get('pokemon_exp', 0)
                pkm.battle_num_vs_wan = pkmn_d.get('battle_num_vs_wan', 0)
                pkm.win_vs_wan = pkmn_d.get('win_vs_wan', 0)
                pkm.battle_num_vs_lan = pkmn_d.get('battle_num_vs_lan', 0)
                pkm.win_vs_lan = pkmn_d.get('win_vs_lan', 0)
                pkm.battle_num_vs_cpu = pkmn_d.get('battle_num_vs_cpu', 0)
                pkm.win_cpu = pkmn_d.get('win_cpu', 0)
                pkm.battle_all_num_tutorial = pkmn_d.get('battle_all_num_tutorial', 0)
                pkm.battle_num_tutorial = pkmn_d.get('battle_num_tutorial', 0)
                pkm.bp_point_atk = pkmn_d.get('bp_point_atk', 0)
                pkm.bp_point_res = pkmn_d.get('bp_point_res', 0)
                pkm.bp_point_def = pkmn_d.get('bp_point_def', 0)
                pkm.bp_point_sp = pkmn_d.get('bp_point_sp', 0)

                load_usr.pokemon_data.append(pkm)

        res.load_user.CopyFrom(load_usr)
        return res.SerializeToString()

    def handle_set_bnpassid_lock(self, data: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SET_BNPASSID_LOCK
        return res.SerializeToString()

    def handle_save_user(self, request: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SAVE_USER

        req = request.save_user
        user_id = req.banapass_id
        
        tut_flgs: List[int] = []
        ach_flgs: List[int] = []
        evt_flgs: List[int] = []
        evt_params: List[int] = []

        get_rank_pts: int = req.get_trainer_rank_point if req.get_trainer_rank_point else 0
        get_money: int = req.get_money
        get_score_pts: int = req.get_score_point if req.get_score_point else 0
        grade_max: int = req.grade_max_num
        extra_counter: int = req.extra_counter
        evt_reward_get_flg: int = req.event_reward_get_flag
        num_continues: int = req.continue_num
        total_play_days: int = req.total_play_days
        awake_num: int = req.awake_num  # ?
        use_support_ct: int = req.use_support_num
        beat_num: int = req.beat_num # ?
        evt_state: int = req.event_state
        aid_skill: int = req.aid_skill
        last_evt: int = req.last_play_event_id

        battle = req.battle_data
        mon = req.pokemon_data

        p = self.data.profile.touch_profile(user_id)
        if p is None or not p:
            self.data.profile.create_profile(user_id)

        if req.trainer_name_pending is not None and req.trainer_name_pending: # we're saving for the first time
            self.data.profile.set_profile_name(user_id, req.trainer_name_pending, req.avatar_gender if req.avatar_gender else None)

        for tut_flg in req.tutorial_progress_flag:
            tut_flgs.append(tut_flg)
        
        self.data.profile.update_profile_tutorial_flags(user_id, tut_flgs)

        for ach_flg in req.achievement_flag:
            ach_flgs.append(ach_flg)
        
        self.data.profile.update_profile_tutorial_flags(user_id, ach_flg)

        for evt_flg in req.event_achievement_flag:
            evt_flgs.append(evt_flg)

        for evt_param in req.event_achievement_param:
            evt_params.append(evt_param)

        self.data.profile.update_profile_event(user_id, evt_state, evt_flgs, evt_params, req.last_play_event_id)
        
        for reward in req.reward_data:
            self.data.item.add_reward(user_id, reward.get_category_id, reward.get_content_id, reward.get_type_id)
        
        self.data.profile.add_profile_points(user_id, get_rank_pts, get_money, get_score_pts, grade_max)
        
        self.data.profile.update_support_team(user_id, 1, req.support_set_1[0], req.support_set_1[1])
        self.data.profile.update_support_team(user_id, 2, req.support_set_2[0], req.support_set_2[1])
        self.data.profile.update_support_team(user_id, 3, req.support_set_3[0], req.support_set_3[1])

        self.data.profile.put_pokemon(user_id, mon.char_id, mon.illustration_book_no, mon.bp_point_atk, mon.bp_point_res, mon.bp_point_def, mon.bp_point_sp)
        self.data.profile.add_pokemon_xp(user_id, mon.char_id, mon.get_pokemon_exp)
        
        for x in range(len(battle.play_mode)):
            self.data.profile.put_pokemon_battle_result(
                user_id, 
                mon.char_id, 
                PokkenConstants.BATTLE_TYPE(battle.play_mode[x]), 
                PokkenConstants.BATTLE_RESULT(battle.result[x])
            )

        self.data.profile.put_stats(
            user_id,
            battle.ex_ko_num,
            battle.wko_num,
            battle.timeup_win_num,
            battle.cool_ko_num,
            battle.perfect_ko_num,
            num_continues
        )

        self.data.profile.put_extra(
            user_id,
            extra_counter,
            evt_reward_get_flg,
            total_play_days,
            awake_num,
            use_support_ct,
            beat_num,
            aid_skill,
            last_evt
        )


        return res.SerializeToString()

    def handle_save_ingame_log(self, data: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SAVE_INGAME_LOG
        return res.SerializeToString()

    def handle_save_charge(self, data: jackal_pb2.Request) -> bytes:
        res = jackal_pb2.Response()
        res.result = 1
        res.type = jackal_pb2.MessageType.SAVE_CHARGE
        return res.SerializeToString()

    def handle_matching_noop(
        self, data: Dict = {}, client_ip: str = "127.0.0.1"
    ) -> Dict:
        return {}

    def handle_matching_start_matching(
        self, data: Dict = {}, client_ip: str = "127.0.0.1"
    ) -> Dict:
        return {}

    def handle_matching_is_matching(
        self, data: Dict = {}, client_ip: str = "127.0.0.1"
    ) -> Dict:
        """
                "sessionId":"12345678",
                "A":{
                    "pcb_id": data["data"]["must"]["pcb_id"],
                    "gip": client_ip
                },
        """
        return {
            "data": {
                "sessionId":"12345678",
                "A":{
                    "pcb_id": data["data"]["must"]["pcb_id"],
                    "gip": client_ip
                },
                "list":[]
            }
        }

    def handle_matching_stop_matching(
        self, data: Dict = {}, client_ip: str = "127.0.0.1"
    ) -> Dict:
        return {}

    def handle_admission_noop(self, data: Dict, req_ip: str = "127.0.0.1") -> Dict:
        return {}
    
    def handle_admission_joinsession(self, data: Dict, req_ip: str = "127.0.0.1") -> Dict:
        self.logger.info(f"Admission: JoinSession from {req_ip}")
        return {
            'data': {
                "id": 12345678
            }
        }