Added automatic timezone update on location change; Update methods now return changed values
This commit is contained in:
parent
e307d60e8e
commit
1c76c8d911
@ -1,8 +1,9 @@
|
|||||||
import logging
|
import logging
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta
|
||||||
from typing import Any, Union
|
from typing import Any, Mapping, Tuple, Union
|
||||||
|
|
||||||
|
import pytz
|
||||||
from bson import ObjectId
|
from bson import ObjectId
|
||||||
|
|
||||||
from classes.location import Location
|
from classes.location import Location
|
||||||
@ -45,7 +46,7 @@ class PyroUser:
|
|||||||
offset: int = 1,
|
offset: int = 1,
|
||||||
time_hour: int = 16,
|
time_hour: int = 16,
|
||||||
time_minute: int = 0,
|
time_minute: int = 0,
|
||||||
):
|
) -> "PyroUser":
|
||||||
db_entry = await col_users.find_one({"id": id})
|
db_entry = await col_users.find_one({"id": id})
|
||||||
|
|
||||||
if db_entry is None:
|
if db_entry is None:
|
||||||
@ -73,7 +74,7 @@ class PyroUser:
|
|||||||
return cls(**db_entry)
|
return cls(**db_entry)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def from_dict(cls, **kwargs):
|
async def from_dict(cls, **kwargs) -> "PyroUser":
|
||||||
if "location" in kwargs:
|
if "location" in kwargs:
|
||||||
try:
|
try:
|
||||||
kwargs["location"] = await Location.get(kwargs["location"]) # type: ignore
|
kwargs["location"] = await Location.get(kwargs["location"]) # type: ignore
|
||||||
@ -81,46 +82,85 @@ class PyroUser:
|
|||||||
kwargs["location"] = None # type: ignore
|
kwargs["location"] = None # type: ignore
|
||||||
return cls(**kwargs)
|
return cls(**kwargs)
|
||||||
|
|
||||||
async def update_locale(self, locale: Union[str, None]) -> None:
|
async def update_locale(self, locale: Union[str, None]) -> Union[str, None]:
|
||||||
"""Change user's locale stored in the database.
|
"""Change user's locale stored in the database.
|
||||||
|
|
||||||
### Args:
|
### Args:
|
||||||
* locale (`Union[str, None]`): New locale to be set.
|
* locale (`Union[str, None]`): New locale to be set.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
logger.debug("%s's locale has been set to %s", self.id, locale)
|
logger.debug("%s's locale has been set to %s", self.id, locale)
|
||||||
|
|
||||||
await col_users.update_one({"_id": self._id}, {"$set": {"locale": locale}})
|
await col_users.update_one({"_id": self._id}, {"$set": {"locale": locale}})
|
||||||
|
|
||||||
self.locale = locale
|
self.locale = locale
|
||||||
|
|
||||||
async def update_state(self, enabled: bool = False) -> None:
|
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)
|
logger.debug("%s's state has been set to %s", self.id, enabled)
|
||||||
|
|
||||||
await col_users.update_one({"_id": self._id}, {"$set": {"enabled": enabled}})
|
await col_users.update_one({"_id": self._id}, {"$set": {"enabled": enabled}})
|
||||||
|
|
||||||
self.enabled = enabled
|
self.enabled = enabled
|
||||||
|
|
||||||
async def update_location(self, location_id: int = 0) -> None:
|
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)
|
logger.debug("%s's location has been set to %s", self.id, location_id)
|
||||||
|
|
||||||
await col_users.update_one(
|
await col_users.update_one(
|
||||||
{"_id": self._id}, {"$set": {"location": location_id}}
|
{"_id": self._id}, {"$set": {"location": location_id}}
|
||||||
)
|
)
|
||||||
self.location = await Location.get(location_id)
|
|
||||||
|
|
||||||
async def update_offset(self, offset: int = 1) -> None:
|
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)
|
logger.debug("%s's offset has been set to %s", self.id, offset)
|
||||||
|
|
||||||
await col_users.update_one({"_id": self._id}, {"$set": {"offset": offset}})
|
await col_users.update_one({"_id": self._id}, {"$set": {"offset": offset}})
|
||||||
|
|
||||||
self.offset = offset
|
self.offset = offset
|
||||||
|
|
||||||
async def update_time(self, hour: int = 16, minute: int = 0) -> None:
|
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)
|
logger.debug("%s's time has been set to %s h. %s m.", self.id, hour, minute)
|
||||||
|
|
||||||
await col_users.update_one(
|
await col_users.update_one(
|
||||||
{"_id": self._id}, {"$set": {"time_hour": hour, "time_minute": minute}}
|
{"_id": self._id}, {"$set": {"time_hour": hour, "time_minute": minute}}
|
||||||
)
|
)
|
||||||
|
|
||||||
self.time_hour = hour
|
self.time_hour = hour
|
||||||
self.time_minute = minute
|
self.time_minute = minute
|
||||||
|
|
||||||
|
return self.time_hour, self.time_minute
|
||||||
|
|
||||||
async def delete(self) -> None:
|
async def delete(self) -> None:
|
||||||
logger.debug("%s's data has been deleted", self.id)
|
logger.debug("%s's data has been deleted", self.id)
|
||||||
await col_users.delete_one({"_id": self._id})
|
await col_users.delete_one({"_id": self._id})
|
||||||
|
|
||||||
async def checkout(self) -> Any:
|
async def checkout(self) -> Mapping[str, Any]:
|
||||||
logger.debug("%s's data has been checked out", self.id)
|
logger.debug("%s's data has been checked out", self.id)
|
||||||
db_entry = await col_users.find_one({"_id": self._id})
|
db_entry = await col_users.find_one({"_id": self._id})
|
||||||
|
|
||||||
@ -155,7 +195,7 @@ class PyroUser:
|
|||||||
logger.warning("Location %s does not have a timezone set", self.location.id)
|
logger.warning("Location %s does not have a timezone set", self.location.id)
|
||||||
|
|
||||||
return (
|
return (
|
||||||
datetime.now(self.location.timezone or timezone.utc) + timedelta(days=1)
|
datetime.now(self.location.timezone or pytz.utc) + timedelta(days=1)
|
||||||
).replace(hour=0, minute=0, second=0, microsecond=0)
|
).replace(hour=0, minute=0, second=0, microsecond=0)
|
||||||
|
|
||||||
def get_reminder_time(self) -> datetime:
|
def get_reminder_time(self) -> datetime:
|
||||||
@ -180,12 +220,12 @@ class PyroUser:
|
|||||||
logger.warning("Location %s does not have a timezone set", self.location.id)
|
logger.warning("Location %s does not have a timezone set", self.location.id)
|
||||||
|
|
||||||
return (
|
return (
|
||||||
datetime.now(timezone.utc)
|
datetime.now(pytz.utc)
|
||||||
.replace(
|
.replace(
|
||||||
hour=self.time_hour,
|
hour=self.time_hour,
|
||||||
minute=self.time_minute,
|
minute=self.time_minute,
|
||||||
second=0,
|
second=0,
|
||||||
microsecond=0,
|
microsecond=0,
|
||||||
)
|
)
|
||||||
.astimezone(self.location.timezone or timezone.utc)
|
.astimezone(self.location.timezone or pytz.utc)
|
||||||
)
|
)
|
||||||
|
Reference in New Issue
Block a user