TelegramBot/classes/pyrouser.py

import logging
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Mapping, Tuple, Union

import pytz
from bson import ObjectId

from classes.location import Location
from modules.database import col_users

logger = logging.getLogger(__name__)


@dataclass
class PyroUser:
    """Dataclass of DB entry of a user"""

    __slots__ = (
        "_id",
        "id",
        "locale",
        "enabled",
        "location",
        "offset",
        "time_hour",
        "time_minute",
    )

    _id: ObjectId
    id: int
    locale: Union[str, None]
    enabled: bool
    location: Union[Location, None]
    offset: int
    time_hour: int
    time_minute: int

    @classmethod
    async def find(
        cls,
        id: int,
        locale: Union[str, None] = None,
        enabled: bool = True,
        location_id: int = 0,
        offset: int = 1,
        time_hour: int = 16,
        time_minute: int = 0,
    ) -> "PyroUser":
        db_entry = await col_users.find_one({"id": id})

        if db_entry is None:
            inserted = await col_users.insert_one(
                {
                    "id": id,
                    "locale": locale,
                    "enabled": enabled,
                    "location": location_id,
                    "offset": offset,
                    "time_hour": time_hour,
                    "time_minute": time_minute,
                }
            )
            db_entry = await col_users.find_one({"_id": inserted.inserted_id})

        if db_entry is None:
            raise RuntimeError("Could not find inserted user entry.")

        try:
            db_entry["location"] = await Location.get(db_entry["location"])  # type: ignore
        except ValueError:
            db_entry["location"] = None  # type: ignore

        return cls(**db_entry)

    @classmethod
    async def from_dict(cls, **kwargs) -> "PyroUser":
        if "location" in kwargs:
            try:
                kwargs["location"] = await Location.get(kwargs["location"])  # type: ignore
            except ValueError:
                kwargs["location"] = None  # type: ignore
        return cls(**kwargs)

    async def update_locale(self, locale: Union[str, None]) -> Union[str, None]:
        """Change user's locale stored in the database.

        ### Args:
            * locale (`Union[str, None]`): New locale to be set.
        """

        logger.debug("%s's locale has been set to %s", self.id, locale)

        await col_users.update_one({"_id": self._id}, {"$set": {"locale": locale}})

        self.locale = locale

        return self.locale

    async def update_state(self, enabled: bool = False) -> bool:
        logger.debug("%s's state has been set to %s", self.id, enabled)

        await col_users.update_one({"_id": self._id}, {"$set": {"enabled": enabled}})

        self.enabled = enabled

        return self.enabled

    async def update_location(self, location_id: int = 0) -> Location:
        logger.debug("%s's location has been set to %s", self.id, location_id)

        await col_users.update_one(
            {"_id": self._id}, {"$set": {"location": location_id}}
        )

        location = await Location.get(location_id)

        # Execute if timezones of old and new locations are different
        if self.location and (self.location.timezone.zone != location.timezone.zone):
            # Get UTC time for selected reminder time
            now_utc = datetime.now(pytz.utc).replace(
                hour=self.time_hour, minute=self.time_minute, second=0, microsecond=0
            )

            # Get the time for the reminder time of old and new location
            local_old = now_utc.astimezone(self.location.timezone)
            local_new = (
                location.timezone.localize(local_old.replace(tzinfo=None))
            ).astimezone(pytz.utc)

            # Update the time to match the new timezone
            await self.update_time(hour=local_new.hour, minute=local_new.minute)

        self.location = location

        return self.location

    async def update_offset(self, offset: int = 1) -> int:
        logger.debug("%s's offset has been set to %s", self.id, offset)

        await col_users.update_one({"_id": self._id}, {"$set": {"offset": offset}})

        self.offset = offset

        return offset

    async def update_time(self, hour: int = 16, minute: int = 0) -> Tuple[int, int]:
        logger.debug("%s's time has been set to %s h. %s m.", self.id, hour, minute)

        await col_users.update_one(
            {"_id": self._id}, {"$set": {"time_hour": hour, "time_minute": minute}}
        )

        self.time_hour = hour
        self.time_minute = minute

        return self.time_hour, self.time_minute

    async def delete(self) -> None:
        logger.debug("%s's data has been deleted", self.id)
        await col_users.delete_one({"_id": self._id})

    async def checkout(self) -> Mapping[str, Any]:
        logger.debug("%s's data has been checked out", self.id)
        db_entry = await col_users.find_one({"_id": self._id})

        if db_entry is None:
            raise KeyError(
                f"DB record with id {self._id} of user {self.id} is not found"
            )

        del db_entry["_id"]  # type: ignore

        return db_entry

    def get_reminder_date(self) -> datetime:
        """Get next reminder date (year, month and day)

        ### Raises:
            * `AttributeError`: Some attribute(s) are missing

        ### Returns:
            * `datetime`: Datetime object of the next reminder date
        """
        if self.location is None:
            logger.warning(
                "Could not get the reminder date for %s: User does not have some attribute(s) set",
                self.id,
            )
            raise AttributeError(
                f"Could not get the reminder date for {self.id}: User does not have some attribute(s) set"
            )

        if not self.location.timezone:
            logger.warning("Location %s does not have a timezone set", self.location.id)

        return (
            datetime.now(self.location.timezone or pytz.utc) + timedelta(days=1)
        ).replace(hour=0, minute=0, second=0, microsecond=0)

    def get_reminder_time(self) -> datetime:
        """Get reminder time (hour and minute)

        ### Raises:
            * `AttributeError`: Some attribute(s) are missing

        ### Returns:
            * `datetime`: Datetime object of the next reminder date
        """
        if self.time_hour is None or self.time_minute is None or self.location is None:
            logger.warning(
                "Could not get the reminder time for %s: User does not have some attribute(s) set",
                self.id,
            )
            raise AttributeError(
                f"Could not get the reminder time for {self.id}: User does not have some attribute(s) set"
            )

        if not self.location.timezone:
            logger.warning("Location %s does not have a timezone set", self.location.id)

        return (
            datetime.now(pytz.utc)
            .replace(
                hour=self.time_hour,
                minute=self.time_minute,
                second=0,
                microsecond=0,
            )
            .astimezone(self.location.timezone or pytz.utc)
        )