220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
from datetime import datetime, timezone
|
|
from random import randint
|
|
from typing import Union, Any
|
|
import discord
|
|
import discord.member
|
|
from modules.utils import config_get
|
|
from modules.database import col_users, col_warnings
|
|
|
|
try:
|
|
from typing import Literal
|
|
except ImportError:
|
|
from typing_extensions import Literal
|
|
|
|
class NotEnoughMoneyError(Exception):
|
|
"""User does not have enough money to do that"""
|
|
pass
|
|
|
|
class UserNotFoundError(Exception):
|
|
"""HoloUser could not find user with such an ID in database"""
|
|
def __init__(self, user, user_id):
|
|
self.user = user
|
|
self.user_id = user_id
|
|
super().__init__(f"User of type {type(self.user)} with id {self.user_id} was not found")
|
|
|
|
class HoloUser():
|
|
|
|
def __init__(self, user: Union[discord.User, discord.Member, discord.member.Member, int]) -> None:
|
|
"""Get an object that has a proper binding between Discord ID and database
|
|
|
|
### Args:
|
|
* `user` (Union[discord.User, discord.Member, discord.member.Member, int]): Object from which ID can be extracted
|
|
|
|
### Raises:
|
|
* `UserNotFoundError`: User with such ID does not seem to exist in database
|
|
"""
|
|
|
|
if hasattr(user, "id"):
|
|
self.id = user.id # type: ignore
|
|
else:
|
|
self.id = user
|
|
|
|
jav_user = col_users.find_one({"user": self.id})
|
|
|
|
if jav_user is None:
|
|
raise UserNotFoundError(user=user, user_id=self.id)
|
|
|
|
self.db_id = jav_user["_id"]
|
|
|
|
# self.xp = jav_user["xp"]
|
|
# self.xp_next = jav_user["xp_next"]
|
|
# self.level = jav_user["level"]
|
|
# self.work_xp = jav_user["work_xp"]
|
|
# self.balance = jav_user["balance"]
|
|
self.customrole = jav_user["customrole"]
|
|
self.customchannel = jav_user["customchannel"]
|
|
# self.married = jav_user["married"]
|
|
# self.marriage_request = jav_user["marriage_request"]
|
|
# self.marriage_request_sent = jav_user["marriage_request_sent"]
|
|
# self.cooldown = jav_user["cooldown"]
|
|
self.warnings = self.warns()
|
|
|
|
# def xp_add(self, amount: int = 1) -> None:
|
|
# """Add some amount of XP points
|
|
|
|
# ### Args:
|
|
# * `amount` (int, optional): Amount of XP points to give. Defaults to 1.
|
|
# """
|
|
# self.xp += amount
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp": self.xp } })
|
|
|
|
# def xp_level_up(self) -> None:
|
|
# """Add 1 to the current XP level"""
|
|
# xp_diff = int(self.xp - self.xp_next)
|
|
# xp_next = int(self.xp_next*configGet("multiplier", "leveling")+configGet("addition", "leveling"))
|
|
# self.xp = xp_diff
|
|
# self.xp_next = xp_next
|
|
# self.level += 1
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp": xp_diff } })
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp_next": xp_next } })
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "level": self.level } })
|
|
|
|
# def balance_set(self, amount: int) -> None:
|
|
# """Set the balance to amount
|
|
|
|
# ### Args:
|
|
# * `amount` (int): Amount of currency to be set
|
|
# """
|
|
# self.balance = amount
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "balance": self.balance } })
|
|
|
|
# def balance_add(self, amount: int) -> None:
|
|
# """Add amount to the balance
|
|
|
|
# ### Args:
|
|
# * `amount` (int): Amount to be added
|
|
# """
|
|
# self.balance_set(self.balance+amount)
|
|
|
|
# def balance_take(self, amount: int) -> bool:
|
|
# """Take amount from the balance
|
|
|
|
# ### Args:
|
|
# * `amount` (int): Amount to be taken
|
|
|
|
# ### Returns:
|
|
# * `bool`: True if successful and False if not
|
|
# """
|
|
# if self.balance >= amount:
|
|
# self.balance_set(self.balance-amount)
|
|
# return True
|
|
# else:
|
|
# return False
|
|
# #raise NotEnoughMoneyError()
|
|
|
|
# def balance_transfer(self, amount: int, destination: Union[Any, int]) -> None:
|
|
# """Transfer money to another user
|
|
|
|
# ### Args:
|
|
# * `amount` (int): Amount to be transferred
|
|
# * `destination` (Union[Any, int]): Destination user of the transfer (should have attribute "id" or be id itself if int)
|
|
|
|
# ### Raises:
|
|
# * `NotEnoughMoneyError`: Not enough money to perform this transaction
|
|
# """
|
|
# if self.balance >= amount:
|
|
# if isinstance(destination, int):
|
|
# destination = HoloUser(destination)
|
|
# self.balance_take(amount)
|
|
# destination.balance_add(amount) # type: ignore
|
|
# else:
|
|
# raise NotEnoughMoneyError()
|
|
|
|
# def salary_get(self) -> int:
|
|
# """Get the salary level depending on work_xp
|
|
|
|
# ### Returns:
|
|
# * `int`: Amount of money to be earned
|
|
# """
|
|
# if self.work_xp >= 100:
|
|
# return randint(configGet("min", "work", "level", "4"), configGet("max", "work", "level", "4"))
|
|
# elif self.work_xp >= 50:
|
|
# return randint(configGet("min", "work", "level", "3"), configGet("max", "work", "level", "3"))
|
|
# elif self.work_xp > 10:
|
|
# return randint(configGet("min", "work", "level", "2"), configGet("max", "work", "level", "2"))
|
|
# else:
|
|
# return randint(configGet("min", "work", "level", "1"), configGet("max", "work", "level", "1"))
|
|
|
|
# def work_xp_add(self) -> None:
|
|
# self.set("work_xp", self.work_xp+1)
|
|
|
|
def warns(self) -> int:
|
|
"""Get number of warnings user has
|
|
|
|
### Returns:
|
|
* `int`: Number of warnings
|
|
"""
|
|
warns = col_warnings.find_one({"user": self.id})
|
|
if warns == None:
|
|
return 0
|
|
else:
|
|
return warns["warns"]
|
|
|
|
def warn(self, count=1, reason: str = "Not provided") -> None:
|
|
"""Warn and add count to warns number
|
|
|
|
### Args:
|
|
* `count` (int, optional): Count of warnings to be added. Defaults to 1.
|
|
"""
|
|
warns = col_warnings.find_one({"user": self.id})
|
|
if warns != None:
|
|
col_warnings.update_one(filter={"_id": self.db_id}, update={ "$set": { "warns": warns["warns"]+count } })
|
|
else:
|
|
col_warnings.insert_one(document={ "user": self.id, "warns": count })
|
|
logWrite(f"User {self.id} was warned {count} times due to: {reason}")
|
|
|
|
# def cooldown_go(self, kind: Literal["work", "daily", "weekly", "monthly", "steal"]) -> None:
|
|
# """Set cooldown start of kind now
|
|
|
|
# ### Args:
|
|
# * `kind` (Literal["work", "daily", "weekly", "monthly", "steal"]): Kind of a cooldown
|
|
# """
|
|
# self.cooldown[kind] = datetime.now(tz=timezone.utc)
|
|
# col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "cooldown": self.cooldown } })
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
"""Set attribute data and save it into database
|
|
|
|
### Args:
|
|
* `key` (str): Attribute to be changed
|
|
* `value` (Any): Value to set
|
|
"""
|
|
if not hasattr(self, key):
|
|
raise AttributeError()
|
|
setattr(self, key, value)
|
|
col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { key: value } }, upsert=True)
|
|
logWrite(f"Set attribute {key} of user {self.id} to {value}")
|
|
|
|
def purge(self) -> None:
|
|
"""Completely remove data from database. Will not remove transactions logs and warnings."""
|
|
col_users.delete_one(filter={"_id": self.db_id})
|
|
self.unauthorize()
|
|
|
|
def unauthorize(self) -> None:
|
|
"""Cancel Oauth2 authorization"""
|
|
col_authorized.find_one_and_delete({"user": self.id})
|
|
|
|
# def is_authorized(self) -> bool:
|
|
# """Check if user provided Oauth2 authorization
|
|
|
|
# ### Returns:
|
|
# * `bool`: True if yes and False if no
|
|
# """
|
|
# if configGet("mode") == "secure":
|
|
# authorized = col_authorized.find_one({"user": self.id})
|
|
# if authorized is not None:
|
|
# return True
|
|
# else:
|
|
# return False
|
|
# else:
|
|
# return True |