# Released under the MIT License. See LICENSE for details.
from __future__ import annotations
from bacommon.restapi.v1.accounts import AccountResponse
import _thread
import json
import threading
import time
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from efro.dataclassio import dataclass_from_json
import _babase
import _bascenev1
import babase
import bascenev1 as bs
from babase._general import CallPartial
from features import profanity
from playersdata import pdata
from repository import profiles
from serverdata import serverdata
from tools import logger
import setting

# Get the blacklist from pdata
blacklist = pdata.get_blacklist()

# Get settings
settings = setting.get_settings_data()

# Track IP join statistics for rate limiting/spam protection
ipjoin: Dict[str, IPJoin] = {}


@dataclass
class PlayerData:
    """A dataclass to hold player-related data."""
    pbid: str
    player_data: Dict[str, Any]
    ip: str
    device_id: str
    last_join: float = field(default_factory=time.time)
    join_count: int = 0
    client_id: int = -1
    display_string: str = ""


@dataclass
class IPJoin:
    """A dataclass to hold IP join information."""
    last_join: float = field(default_factory=time.time)
    count: int = 0


class ServerCheck:
    """A class to check for new players and handle their joining process."""

    def __init__(self):
        self.players: List[str] = []
        self.ip_client_map: Dict[str, List[int]] = {}
        self.device_client_map: Dict[str, List[int]] = {}
        self.ip_join: Dict[str, IPJoin] = {}
        self.timer = bs.AppTimer(1, babase.CallStrict(self.check), repeat=True)

    def check(self) -> None:
        """
        Checks for new players, handles their joining process, and performs various checks.
        """
        new_players = []
        self.ip_client_map.clear()
        self.device_client_map.clear()

        for ros in bs.get_game_roster():
            client_id = ros["client_id"]
            if client_id == -1:
                continue

            ip = _bascenev1.get_client_ip(client_id)
            device_id = self._get_device_id(client_id)
            account_id = ros["account_id"]

            if not self._handle_connection_limits(ip, device_id, client_id, account_id):
                continue

            new_players.append(account_id)

            if account_id not in self.players:
                self._handle_new_player(ros, ip, device_id)

        self.players = new_players

    def _get_device_id(self, client_id: int) -> Optional[str]:
        """
        Returns the public or private device UUID for a given client_id.
        """
        device_id = _bascenev1.get_client_public_device_uuid(client_id)
        if device_id is None:
            device_id = _bascenev1.get_client_device_uuid(client_id)
        return device_id

    def _handle_connection_limits(self, ip: str, device_id: str, client_id: int, account_id: str) -> bool:
        """
        Handles connection limits per IP and device.
        """
        if not self._check_limit(self.device_client_map, device_id, client_id, "device", account_id):
            return False
        if not self._check_limit(self.ip_client_map, ip, client_id, "IP", account_id):
            return False
        return True

    def _check_limit(self, client_map: Dict[str, List[int]], key: str, client_id: int, limit_type: str, account_id: str) -> bool:
        """
        Checks if the connection limit for a given type (IP or device) has been reached.
        """
        if key not in client_map:
            client_map[key] = [client_id]
        else:
            client_map[key].append(client_id)
            if len(client_map[key]) >= settings['maxAccountPerIP']:
                self._disconnect_player(
                    client_id,
                    f"Only {settings['maxAccountPerIP']} players per {limit_type} allowed, disconnecting this device.",
                    f'Player disconnected, reached max players per {limit_type} || {account_id}',
                )
                return False
        return True

    def _disconnect_player(self, client_id: int, message: str, log_message: str) -> None:
        """

        Disconnects a player with a given message and logs the disconnection.
        """
        bs.chatmessage(message, clients=[client_id])
        bs.disconnect_client(client_id)
        logger.log(log_message, "playerjoin")

    def _handle_new_player(self, ros: Dict[str, Any], ip: str, device_id: str) -> None:
        """
        Handles the joining process for a new player.
        """
        account_id = ros["account_id"]
        display_string = ros["display_string"]
        client_id = ros["client_id"]

        censored_name = profanity.censor(display_string)
        if censored_name != display_string:
            self._disconnect_player(
                client_id,
                "Profanity in ID, change your ID and join back",
                f'{display_string} || {account_id} || kicked by profanity check',
            )
            return

        if settings["whitelist"] and account_id not in pdata.CacheData.whitelist:
            self._disconnect_player(
                client_id,
                "Not in whitelist, contact admin",
                f'{display_string} || {account_id} | kicked > not in whitelist',
            )
            return

        logger.log(
            f'{display_string}  || {account_id} || joined server', "playerjoin")
        logger.log(f'{account_id} {ip} {device_id}')

        if account_id in serverdata.clients:
            on_player_join_server(
                account_id, serverdata.clients[account_id], ip, device_id)
        else:
            LoadProfile(account_id, ip, device_id).start()


def on_player_join_server(pbid: str, player_data: Optional[Dict[str, Any]], ip: str, device_id: str) -> None:
    """
    Handles the joining process for a player on the server.
    """
    global ipjoin
    now = time.time()
    client_id = -1
    display_string = ""

    for ros in bs.get_game_roster():
        if ros["account_id"] == pbid:
            client_id = ros["client_id"]
            display_string = ros['display_string']
            break

    if client_id == -1:
        return

    if ip in ipjoin:
        ip_info = ipjoin[ip]
        last_join = ip_info.last_join
        join_count = ip_info.count
        if now - last_join < 15:
            join_count += 1
            if join_count > 2:
                bs.broadcastmessage(
                    "Joining too fast, slow down dude",
                    color=(1, 0, 1),
                    transient=True,
                    clients=[client_id],
                )
                logger.log(f'{pbid} || kicked for joining too fast')
                bs.disconnect_client(client_id)
                _thread.start_new_thread(report_spam, (pbid,))
                return
        else:
            join_count = 0
        ip_info.count = join_count
        ip_info.last_join = now
    else:
        ipjoin[ip] = IPJoin(last_join=now, count=0)

    if pbid in serverdata.clients:
        serverdata.clients[pbid]["lastJoin"] = now

    if player_data is not None:
        handle_existing_player(pbid, player_data, ip,
                               device_id, client_id, display_string)
    else:
        handle_new_player_data(pbid, display_string, client_id)


def handle_existing_player(pbid: str, player_data: Dict[str, Any], ip: str, device_id: str, client_id: int, display_string: str) -> None:
    """
    Handles the joining process for an existing player.
    """
    serverdata.recents.append(
        {
            "client_id": client_id,
            "deviceId": display_string,
            "pbid": pbid,
            "ip": ip,
            "device_uuid": device_id,
        }
    )
    serverdata.recents = serverdata.recents[-20:]

    if check_ban(ip, device_id, pbid):
        _bascenev1.chatmessage(
            'sad, your account is flagged contact server owner for unban', clients=[client_id])
        bs.disconnect_client(client_id)
        return

    if get_account_age(player_data["creationDate"]) < settings["minAgeToJoinInHours"]:
        bs.broadcastmessage(
            "New Accounts not allowed here, come back later",
            color=(1, 0, 0),
            transient=True,
            clients=[client_id],
        )
        logger.log(pbid + " | kicked > reason:Banned account")
        bs.disconnect_client(client_id)
        return

    current_time = datetime.now()
    if pbid not in serverdata.clients:
        serverdata.clients[pbid] = player_data
        serverdata.clients[pbid]["warnCount"] = 0
        serverdata.clients[pbid]["lastWarned"] = time.time()
        serverdata.clients[pbid]["verified"] = False
        serverdata.clients[pbid]["rejoincount"] = 1
        serverdata.clients[pbid]["lastJoin"] = time.time()
        if pbid in blacklist["kick-vote-disabled"] and current_time < datetime.strptime(
            blacklist["kick-vote-disabled"][pbid]["till"], "%Y-%m-%d %H:%M:%S"
        ):
            _bascenev1.disable_kickvote(pbid)

    serverdata.clients[pbid]["lastIP"] = ip
    serverdata.clients[pbid]["deviceUUID"] = device_id
    verify_account(pbid, player_data)
    logger.log(
        f'{pbid} ip: {serverdata.clients[pbid]["lastIP"]}, Device id: {device_id}')
    bs.broadcastmessage(
        settings["regularWelcomeMsg"] + " " + display_string,
        color=(0.6, 0.8, 0.6),
        transient=True,
        clients=[client_id],
    )
    if settings["ballistica_web"]["enable"]:
        from . import notification_manager
        notification_manager.player_joined(pbid)


def handle_new_player_data(pbid: str, display_string: str, client_id: int) -> None:
    """
    Handles the joining process for a player with no existing data.
    """

    thread = FetchThread(
        target=my_acc_age,
        callback=save_age,
        pb_id=pbid,
        display_string=display_string,
    )
    thread.start()
    bs.broadcastmessage(
        settings["firstTimeJoinMsg"],
        color=(0.6, 0.8, 0.6),
        transient=True,
        clients=[client_id],
    )
    if settings["ballistica_web"]["enable"]:
        from . import notification_manager
        notification_manager.player_joined(pbid)


def check_ban(ip: str, device_id: str, pbid: str, log: bool = True) -> bool | str:
    """
    Checks if a player is banned based on their IP, device ID, or player ID.
    """
    current_time = datetime.now()

    def check_ban_list(ban_list: Dict[str, Any], key: str, ban_type: str) -> Optional[str]:
        if key in ban_list and current_time < datetime.strptime(
            ban_list[key]["till"], "%Y-%m-%d %H:%M:%S"
        ):
            return f'reason: matched {ban_type} | {ban_list[key]["reason"]}, Till: {ban_list[key]["till"]}'
        return None

    ban_msg = check_ban_list(blacklist["ban"]["ips"], ip, "IP")
    if ban_msg is None:
        ban_msg = check_ban_list(
            blacklist["ban"]["deviceids"], device_id, "deviceId")
    if ban_msg is None:
        ban_msg = check_ban_list(blacklist["ban"]["ids"], pbid, "ID")

    if ban_msg:
        if log:
            logger.log(f'{pbid} | kicked > {ban_msg}')
            return True
        return ban_msg
    return False


def verify_account(pb_id: str, p_data: Dict[str, Any]) -> None:
    """
    Verifies a player's account by checking their display string against their device accounts.
    """
    if _bascenev1.protocol_version() > 35:
        serverdata.clients[pb_id]["verified"] = True
        return

    display_string = ""
    for ros in bs.get_game_roster():
        if ros['account_id'] == pb_id:
            display_string = ros['display_string']
            break

    if display_string not in p_data.get('display_string', []):
        thread2 = FetchThread(
            target=get_device_accounts,
            callback=save_ids,
            pb_id=pb_id,
            display_string=display_string,
        )
        thread2.start()
    else:
        serverdata.clients[pb_id]["verified"] = True


def _make_request_safe(request: Callable, retries: int = 2, raise_err: bool = True) -> Any:
    """
    A wrapper to make a request safely with retries.
    """
    try:
        return request()
    except Exception:
        if retries > 0:
            time.sleep(1)
            return _make_request_safe(request, retries=retries - 1, raise_err=raise_err)
        if raise_err:
            raise


def get_account_creation_date(pb_id: str) -> Optional[str]:
    """
    Gets the account creation date for a given player ID.
    """
    if _bascenev1.protocol_version() > 35:

        try:
            req = urllib.request.Request(
                f"https://www.ballistica.net/api/v1/accounts/{pb_id}",
                headers={
                    "Authorization": f"Bearer {settings['accountApiToken']}"
                },
            )
            with urllib.request.urlopen(req) as response:
                response_json_str = response.read().decode('utf-8')
                account = json.loads(response_json_str)

                if account:
                    account = dataclass_from_json(
                        AccountResponse, response_json_str)
                    return str(account.create_time)
        except (urllib.error.URLError, ValueError) as e:
            logger.log(
                f"Error getting account creation date for {pb_id}: {e}", "error")
            return None
    else:
        account_creation_url = f"http://bombsquadgame.com/accountquery?id={pb_id}"
        try:
            with urllib.request.urlopen(account_creation_url) as response:
                account_creation = json.loads(response.read())
                creation_time = datetime.strptime(
                    "/".join(map(str, account_creation["created"])), "%Y/%m/%d/%H/%M/%S")
                # Convert to IST
                creation_time += timedelta(hours=5, minutes=30)
                return str(creation_time)
        except (urllib.error.URLError, ValueError) as e:
            logger.log(
                f"Error getting account creation date for {pb_id}: {e}", "error")
            return None
    return None


def get_device_accounts(pb_id: str) -> List[str]:
    """
    Gets the device accounts associated with a given player ID.
    """
    url = f"http://bombsquadgame.com/bsAccountInfo?buildNumber=20258&accountID={pb_id}"
    try:
        with urllib.request.urlopen(url) as response:
            return json.loads(response.read())["accountDisplayStrings"]
    except (urllib.error.URLError, ValueError) as e:
        logger.log(f"Error getting device accounts for {pb_id}: {e}", "error")
        return ['???']


class LoadProfile(threading.Thread):
    """
    A thread to load a player's profile from pdata.
    """

    def __init__(self, pb_id: str, ip: str, device_id: str):
        super().__init__()
        self.pbid = pb_id
        self.ip = ip
        self.device_id = device_id

    def run(self) -> None:
        player_data = pdata.get_info(self.pbid)
        _babase.pushcall(
            CallPartial(on_player_join_server, self.pbid,
                 player_data, self.ip, self.device_id),
            from_other_thread=True,
        )


class FetchThread(threading.Thread):
    """
    A thread to fetch data from a URL and execute a callback with the result.
    """

    def __init__(self, target: Callable, callback: Optional[Callable] = None, pb_id: str = "ji", display_string: str = "XXX"):
        super().__init__(target=self.target_with_callback, args=(pb_id, display_string))
        self.callback = callback
        self.method = target

    def target_with_callback(self, pb_id: str, display_string: str) -> None:
        data = self.method(pb_id)
        if self.callback is not None:
            _babase.pushcall(
                CallPartial(self.callback, data, pb_id, display_string),
                from_other_thread=True,
            )


def my_acc_age(pb_id: str) -> Optional[str]:
    return get_account_creation_date(pb_id)


def save_age(age: Optional[str], pb_id: str, display_string: str) -> None:
    if age:
        pdata.add_profile(pb_id, display_string, display_string, age)
        if _bascenev1.protocol_version() <= 35:
            time.sleep(2)
            thread2 = FetchThread(
                target=get_device_accounts,
                callback=save_ids,
                pb_id=pb_id,
                display_string=display_string,
            )
            thread2.start()
        if get_account_age(age) < settings["minAgeToJoinInHours"]:
            msg = "New Accounts not allowed to play here, come back tmrw."
            logger.log(f"{pb_id} || kicked > new account")
            kick_by_pb_id(pb_id, msg)


def save_ids(ids: List[str], pb_id: str, display_string: str) -> None:
    pdata.update_display_string(pb_id, ids)
    if display_string not in ids:
        msg = "Spoofed Id detected, Goodbye"
        kick_by_pb_id(pb_id, msg)
        serverdata.clients[pb_id]["verified"] = False
        logger.log(f"{pb_id} || kicked, for using spoofed id {display_string}")
    else:
        serverdata.clients[pb_id]["verified"] = True


def kick_by_pb_id(pb_id: str, msg: str) -> None:
    for ros in bs.get_game_roster():
        if ros['account_id'] == pb_id:
            bs.broadcastmessage(msg, transient=True,
                                clients=[ros['client_id']])
            bs.disconnect_client(ros['client_id'])
            break


def get_account_age(ct: str) -> float:
    try:
        creation_time = datetime.strptime(ct, "%Y-%m-%d %H:%M:%S")
        return (datetime.now() - creation_time).total_seconds() / 3600
    except ValueError:
        return float('inf')


def report_spam(pbid: str) -> None:
    now = time.time()
    profiles = pdata.get_profiles()
    if pbid in profiles:
        spam_count = profiles[pbid].get("spamCount", 0)
        last_spam = profiles[pbid].get("lastSpam", 0)
        if now - last_spam < 2 * 24 * 3600:
            spam_count += 1
            if spam_count > 3:
                logger.log(f"{pbid} auto banned for spamming")
                pdata.ban_player(pbid, 1, "auto ban exceed warn count")
        else:
            spam_count = 0
        profiles[pbid]["spamCount"] = spam_count
        profiles[pbid]["lastSpam"] = now


def on_join_request(ip: str) -> None:
    now = time.time()
    if ip in serverdata.ips:
        last_request = serverdata.ips[ip].get("lastRequest", 0)
        count = serverdata.ips[ip].get("count", 0)
        if now - last_request < 5:
            count += 1
            if count > 40:
                _bascenev1.ban_ip(ip)
        else:
            count = 0
        serverdata.ips[ip] = {"lastRequest": now, "count": count}
    else:
        serverdata.ips[ip] = {"lastRequest": now, "count": 0}


def account_check(account_id: str, ip: str, client_id: int) -> None:
    if not account_id.startswith("\ue063"):
        return

    account_id = account_id.replace("\ue063", "")
    profile = profiles.get_profile(account_id)
    enforce_mfa = settings["mfa"]["enforce_for_all_players"] or account_id in settings["mfa"]["enforce_for_accounts"]

    if enforce_mfa:
        if profile is None or profile["lastIP"] != ip:
            try:
                urllib.request.urlopen(
                    f"https://mods.69420555.xyz/verifyownerip?ip={ip}&tag={account_id}").close()
                profiles.upsert_ip(account_id, ip)
            except urllib.error.URLError:
                _babase.pushcall(
                    CallPartial(bs.chatmessage, "Click stats button and login your V2 account, to verify your identity", [
                         client_id]),
                    from_other_thread=True,
                )
                _babase.pushcall(
                    CallPartial(bs.disconnect_client, client_id, 2), from_other_thread=True)


# Instantiate the server check
server_check = ServerCheck()