mirror of
https://github.com/wlinator/luminara.git
synced 2024-10-02 18:23:12 +00:00
Add daily service
This commit is contained in:
parent
09ab8db099
commit
bdbc5d0e73
1 changed files with 117 additions and 0 deletions
117
services/daily_service.py
Normal file
117
services/daily_service.py
Normal file
|
@ -0,0 +1,117 @@
|
|||
from datetime import datetime, timedelta
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
import pytz
|
||||
|
||||
from db import database
|
||||
from lib.const import CONST
|
||||
from services.currency_service import Currency
|
||||
|
||||
|
||||
class Dailies:
|
||||
def __init__(self, user_id: int) -> None:
|
||||
self.user_id: int = user_id
|
||||
self.amount: int = 0
|
||||
self.tz = pytz.timezone("US/Eastern")
|
||||
self.time_now: datetime = datetime.now(tz=self.tz)
|
||||
self.reset_time: datetime = self.time_now.replace(
|
||||
hour=7,
|
||||
minute=0,
|
||||
second=0,
|
||||
microsecond=0,
|
||||
)
|
||||
|
||||
data: Tuple[Optional[str], int] = Dailies.get_data(user_id)
|
||||
|
||||
if data[0] is not None:
|
||||
self.claimed_at: datetime = datetime.fromisoformat(data[0])
|
||||
else:
|
||||
# set date as yesterday to mock a valid claimed_at.
|
||||
self.claimed_at: datetime = datetime.now(tz=self.tz) - timedelta(days=2)
|
||||
|
||||
self.streak: int = int(data[1])
|
||||
|
||||
def refresh(self) -> None:
|
||||
if self.amount == 0:
|
||||
self.amount = CONST.DAILY_REWARD
|
||||
query: str = """
|
||||
INSERT INTO dailies (user_id, amount, claimed_at, streak)
|
||||
VALUES (%s, %s, %s, %s)
|
||||
"""
|
||||
values: Tuple[int, int, str, int] = (
|
||||
self.user_id,
|
||||
self.amount,
|
||||
self.claimed_at.isoformat(),
|
||||
self.streak,
|
||||
)
|
||||
database.execute_query(query, values)
|
||||
|
||||
cash = Currency(self.user_id)
|
||||
cash.add_balance(self.amount)
|
||||
cash.push()
|
||||
|
||||
def can_be_claimed(self) -> bool:
|
||||
if self.claimed_at is None:
|
||||
return True
|
||||
|
||||
if self.time_now < self.reset_time:
|
||||
self.reset_time -= timedelta(days=1)
|
||||
|
||||
return self.claimed_at < self.reset_time <= self.time_now
|
||||
|
||||
def streak_check(self) -> bool:
|
||||
"""
|
||||
Three checks are performed, only one has to return True.
|
||||
1. the daily was claimed yesterday
|
||||
2. the daily was claimed the day before yesterday (users no longer lose their dailies as fast)
|
||||
3. the daily was claimed today but before the reset time (see __init__)
|
||||
:return:
|
||||
"""
|
||||
|
||||
check_1: bool = (
|
||||
self.claimed_at.date() == (self.time_now - timedelta(days=1)).date()
|
||||
)
|
||||
check_2: bool = (
|
||||
self.claimed_at.date() == (self.time_now - timedelta(days=2)).date()
|
||||
)
|
||||
check_3: bool = (
|
||||
self.claimed_at.date() == self.time_now.date()
|
||||
and self.claimed_at < self.reset_time
|
||||
)
|
||||
|
||||
return check_1 or check_2 or check_3
|
||||
|
||||
@staticmethod
|
||||
def get_data(user_id: int) -> Tuple[Optional[str], int]:
|
||||
query: str = """
|
||||
SELECT claimed_at, streak
|
||||
FROM dailies
|
||||
WHERE id = (
|
||||
SELECT MAX(id)
|
||||
FROM dailies
|
||||
WHERE user_id = %s
|
||||
)
|
||||
"""
|
||||
|
||||
try:
|
||||
(claimed_at, streak) = database.select_query(query, (user_id,))[0]
|
||||
except (IndexError, TypeError):
|
||||
(claimed_at, streak) = None, 0
|
||||
|
||||
return claimed_at, streak
|
||||
|
||||
@staticmethod
|
||||
def load_leaderboard() -> List[Tuple[int, int, str, int]]:
|
||||
query: str = """
|
||||
SELECT user_id, MAX(streak), claimed_at
|
||||
FROM dailies
|
||||
GROUP BY user_id
|
||||
ORDER BY MAX(streak) DESC;
|
||||
"""
|
||||
|
||||
data: List[Tuple[int, int, str]] = database.select_query(query)
|
||||
|
||||
leaderboard: List[Tuple[int, int, str, int]] = [
|
||||
(row[0], row[1], row[2], rank) for rank, row in enumerate(data, start=1)
|
||||
]
|
||||
return leaderboard
|
Loading…
Reference in a new issue