Dev

.env.example

@@ -0,0 +1,14 @@
+TZ=Europe/Moscow  # TZ identifier
+
+BOT_TOKEN=1234343:AABBCCEE  # https://botfather.t.me
+ADMIN_IDS=77000,77001  # https://myidbot.t.me
+CRYPTO_TOKEN=1234343:AABBOOCCEE  # https://send.t.me
+FLYER_TOKEN=FL-lekLE-lekLE-lekLE-lekLE  # https://FlyerServiceBot.t.me
+
+# Database Configuration
+DB_DRIVER=postgresql+asyncpg
+DB_HOST=ip
+DB_PORT=5432
+DB_NAME=folt_copy
+DB_USER=user
+DB_PASSWORD="password"

data/config.json

@@ -29,7 +29,8 @@
       "name": "Карта4",
       "photo": "https://tinypic.host/",
       "points": "10000",
-      "rarity": "Легендарная"
+      "rarity": "Легендарная",
+			"description": "Описание"
       }
   ]
 }

data/text.py

@@ -1,27 +1,27 @@
 name_card = "Карта"
 name_card_rod_padezh = "Карту"
-WELCOME_MESSAGE_PRIVATE = f'''👋 Привет! Тут ты можешь собирать уникальные карточки и соревноваться с другими игроками\n\n<b>Как получить карточки?</b>\n<blockquote>Отправь команду «{name_card}»</blockquote>\n\nУзнать все функции можно по команде /help'''
+WELCOME_MESSAGE_PRIVATE = f"""👋 Привет! Тут ты можешь собирать уникальные карточки и соревноваться с другими игроками\n\n<b>Как получить карточки?</b>\n<blockquote>Отправь команду «{name_card}»</blockquote>\n\nУзнать все функции можно по команде /help"""
 
-WELCOME_MESSAGE = f'''👋 Всем привет! Я бот, в котором ты можешь собирать уникальные карточки и соревноваться с другими игроками\n
+WELCOME_MESSAGE = f"""👋 Всем привет! Я бот, в котором ты можешь собирать уникальные карточки и соревноваться с другими игроками\n
 Как начать?\n<blockquote>Напишите «{name_card}» для получения первой карточки</blockquote>
-\n\nУзнать все функции можно по команде /help'''
+\n\nУзнать все функции можно по команде /help"""
 
 HELP_MESSAGE = (
     "<b>Что это за бот?</b>\n"
-		f"<blockquote>Тут ты можешь собирать карточки <b>{name_card}</b> и соревноваться с другими игроками</blockquote>\n\n"
+    f"<blockquote>Тут ты можешь собирать карточки <b>{name_card}</b> и соревноваться с другими игроками</blockquote>\n\n"
     "<b>Команды:</b>\n"
     "<blockquote>👤 /profile — ваш профиль\n✨ /name [ник] — изменить никнейм\n🃏 /cards — собирать карточки\n🏆 /top — топ игроков\n🚀 /premium — приобрести премиум \n🛍️ /shop — игровой магазин \n🎲 /diceplay — играть в кости\n🏪 /market - биржа уникальных карточек</blockquote>"
     f"<b>Для получения карты отправьте любую из команд </b><blockquote>{name_card}\nКарта2 \nКарта3 \nКарта7 \nКарта5 \nКарта6</blockquote>"
 )
 
-PRIVACY_MESSAGE = '''Мы обрабатываем данные пользователей строго в целях улучшения функционала нашего бота. 
+PRIVACY_MESSAGE = """Мы обрабатываем данные пользователей строго в целях улучшения функционала нашего бота. 
 Гарантируем, что данные пользователя, включая идентификатор пользователя (user ID) и имя (first name), 
 не будут переданы третьим лицам или использованы вне контекста улучшения бота. Наш приоритет — обеспечение 
 безопасности и конфиденциальности информации, которую вы нам доверяете.\n\nДля повышения прозрачности нашей работы, 
 мы также обязуемся предоставлять пользователю доступ к информации о том, какие данные собраны и как они используются. 
 В случае изменения политики использования данных, мы своевременно информируем пользователей через обновления нашего 
 пользовательского соглашения. Мы прилагаем все усилия, чтобы наш сервис был максимально безопасным и удобным для 
-пользователя.'''
+пользователя."""
 
 responses = [
     "Уберите лапки от чужой кнопки.",
@@ -34,17 +34,17 @@
     "Ваши лапки попали не туда.",
     "Лапки в сторону!",
     "Ваши лапки слишком любопытны!",
-    "Лапки в сторону, эта кнопка охраняется."
+    "Лапки в сторону, эта кнопка охраняется.",
 ]
 
 PREMIUM_TEXT = (
-	              "<b>🚀 Premium</b>\n\n"
-                "<blockquote>⌛️ Возможность получать карточки каждые 3 часа вместо 4\n"
-                "🃏 Повышенная вероятность выпадения легендарных и мифических карт\n"
-                "🔥 Возможность использовать смайлики в никнейме\n"
-                "💎 Отображение алмаза в топе карточек\n"
-                "⚡️ Более быстрая обработка твоих сообщений\n"
-                "Срок действия • 30 дней</blockquote>\n\n"
+    "<b>🚀 Premium</b>\n\n"
+    "<blockquote>⌛️ Возможность получать карточки каждые 3 часа вместо 4\n"
+    "🃏 Повышенная вероятность выпадения легендарных и мифических карт\n"
+    "🔥 Возможность использовать смайлики в никнейме\n"
+    "💎 Отображение алмаза в топе карточек\n"
+    "⚡️ Более быстрая обработка твоих сообщений\n"
+    "Срок действия • 30 дней</blockquote>\n\n"
 )
 
 forbidden_symbols = [
@@ -53,21 +53,20 @@
     "\u534d",
     "1488",
     "heil",
-    "hitler"
+    "hitler",
 ]
 
 
-shop_text = ("🛍️ Магазин\n"
-"<blockquote> Выберите нужную категорию: </blockquote>")
+shop_text = "🛍️ Магазин\n" "<blockquote> Выберите нужную категорию: </blockquote>"
 
-booster_text = ("⚡️ Бустеры\n"
-                "<blockquote> Выберите нужный бустер </blockquote>")
+booster_text = "⚡️ Бустеры\n" "<blockquote> Выберите нужный бустер </blockquote>"
 
-coins_text = ("💰 Монеты"
-              "<blockquote> Приобрести монеты за звезды </blockquote>")
+coins_text = "💰 Монеты" "<blockquote> Приобрести монеты за звезды </blockquote>"
 
 
-confirmation_payment_text = "🌟Спасибо за покупку, на ваш баланс были начислены монеты. ({quantity})"
+confirmation_payment_text = (
+    "🌟Спасибо за покупку, на ваш баланс были начислены монеты. ({quantity})"
+)
 
 
 luck_message = (
@@ -96,16 +95,12 @@
 )
 
 dice_limit = (
-    "🎲 Игра «<b>Кости</b>»\n"
-    "<blockquote>Будет доступно через {} минут.</blockquote>"
+    "🎲 Игра «<b>Кости</b>»\n" "<blockquote>Будет доступно через {} минут.</blockquote>"
 )
 
 
 settings_chat = (
-    "⚙️ Настройки группы\n"
-    "Выберите один из параметров, который вы хотите изменить"
+    "⚙️ Настройки группы\n" "Выберите один из параметров, который вы хотите изменить"
 )
 
-post_msg = (
-    f"Напиши «<b>{name_card}</b>» и я подарю тебе карточку!"
-)
+post_msg = f"Напиши «<b>{name_card}</b>» и я подарю тебе карточку!"

database/__init__.py

@@ -3,6 +3,6 @@
 from utils.loader import engine
 
 
-async def setup_db():
+async def init_db():
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)

database/bonus_link.py

@@ -9,46 +9,53 @@
 
 
 def generate_random_string(length=14):
-    random_string = ''
+    random_string = ""
     random_str_seq = string.ascii_letters + string.digits
-    for i in range(0,length):
+    for i in range(0, length):
         if i % length == 0 and i != 0:
-            random_string += '-'
+            random_string += "-"
         random_string += str(random_str_seq[random.randint(0, len(random_str_seq) - 1)])
     return random_string
 
 
-async def create_bonus_link(for_user_id: int, ) -> BonusLink:
+async def create_bonus_link(
+    for_user_id: int,
+) -> BonusLink:
     ref_link: str = generate_random_string()
     async with AsyncSession(engine, expire_on_commit=False) as session:
-        ref_link = BonusLink(code=ref_link, for_user_id=for_user_id)
-        session.add(ref_link)
+        ref_link_model = BonusLink(code=ref_link, for_user_id=for_user_id)
+        session.add(ref_link_model)
         await session.commit()
-        return ref_link
+        return ref_link_model
 
 
 async def delete_bonus_link(bonus_link: str) -> None:
     async with AsyncSession(engine, expire_on_commit=False) as session:
-        bonus_link = (await session.execute(select(BonusLink).where(BonusLink.code == bonus_link))).scalar_one_or_none()
-        if bonus_link is None:
+        bonus_link_model = (
+            await session.execute(select(BonusLink).where(BonusLink.code == bonus_link))
+        ).scalar_one_or_none()
+        if bonus_link_model is None:
             return
-        await session.delete(bonus_link)
+        await session.delete(bonus_link_model)
         await session.commit()
 
 
-async def get_bonus_link(code: str) -> BonusLink:
+async def get_bonus_link(code: str) -> BonusLink | None:
     async with AsyncSession(engine, expire_on_commit=False) as session:
-        return (await session.execute(
-            select(BonusLink).where(BonusLink.code == code)
-        )).scalar_one_or_none()
+        return (
+            await session.execute(select(BonusLink).where(BonusLink.code == code))
+        ).scalar_one_or_none()
 
 
 async def deactivate_bonus_link(code: str):
     async with AsyncSession(engine, expire_on_commit=False) as session:
-        bonus = (await session.execute(
-            select(BonusLink).where(BonusLink.code == code)
-        )).scalar_one_or_none()
+        bonus_model = (
+            await session.execute(select(BonusLink).where(BonusLink.code == code))
+        ).scalar_one_or_none()
+
+        if bonus_model is None:
+            raise KeyError(f"Didn't found bonus link with code {code}")
 
-        if bonus:
-            bonus.is_active = False
-            await session.commit()
+        if bonus_model is not None:
+            bonus_model.is_active = False  # type: ignore
+            await session.commit()

database/cards.py

@@ -7,88 +7,111 @@
 from .models import Card, LimitedCards
 
 
-async def parse_cards(filename):
+async def parse_cards(filename) -> None:
+    with open(filename, "r", encoding="utf8") as f:
+        data = json.load(f)
+    data = data["cats"]
+
     async with AsyncSession(engine) as session:
-        with open(filename, 'r', encoding="utf8") as f:
-            data = json.load(f)
-            data = data['cats']
-            for card in data:
-                card_id = int(card['id'])
-
-                existing_card = await session.get(Card, card_id)
-                if existing_card:
-                    existing_card.name = card['name']
-                    existing_card.points = int(card['points'])
-                    existing_card.rarity = card['rarity']
-                    existing_card.photo = card['photo']
-                    existing_card.description = card.get('description', None)
-                else:
-                    db_card = Card(
-                        id=card_id, 
-                        name=card['name'],
-                        points=int(card['points']),
-                        rarity=card['rarity'],
-                        photo=card['photo'],
-                        description=card.get('description', None)
-                    )
-                    session.add(db_card)
+        for card in data:
+            card_id = int(card["id"])
+
+            existing_card = await session.get(Card, card_id)
+            if existing_card:
+                existing_card.name = card["name"]
+                existing_card.points = int(card["points"])
+                existing_card.rarity = card["rarity"]
+                existing_card.photo = card["photo"]
+                existing_card.description = card.get("description", None)
+                continue
+
+            db_card = Card(
+                id=card_id,
+                name=card["name"],
+                points=int(card["points"]),
+                rarity=card["rarity"],
+                photo=card["photo"],
+                description=card.get("description", None),
+            )
+            session.add(db_card)
         await session.commit()
 
 
-async def parse_limited_cards(filename):
+async def parse_limited_cards(filename) -> None:
+    with open(filename, "r", encoding="utf8") as f:
+        data = json.load(f)
+    data = data["cats"]
+
     async with AsyncSession(engine) as session:
-        with open(filename, 'r', encoding="utf8") as f:
-            data = json.load(f)
-            data = data['cats']
-            for card in data:
-                card_id = int(card['id'])
-
-                existing_card = await session.get(LimitedCards, card_id)
-                if existing_card:
-                    existing_card.name = card['name']
-                    existing_card.price = int(card['price'])
-                    existing_card.edition = int(card['edition'])
-                    existing_card.photo = card['photo']
-                    existing_card.description = card.get('description', None)
-                else:
-                    db_card = LimitedCards(
-                        id=card_id,
-                        name=card['name'],
-                        price=int(card['price']),
-                        edition=int(card['edition']),
-                        photo=card['photo'],
-                        description=card.get('description', None)
-                    )
-                    session.add(db_card)
+        for card in data:
+            card_id = int(card["id"])
+
+            existing_card = await session.get(LimitedCards, card_id)
+            if existing_card:
+                existing_card.name = card["name"]
+                existing_card.price = int(card["price"])
+                existing_card.edition = int(card["edition"])
+                existing_card.photo = card["photo"]
+                existing_card.description = card.get("description", None)
+                continue
+
+            db_card = LimitedCards(
+                id=card_id,
+                name=card["name"],
+                price=int(card["price"]),
+                edition=int(card["edition"]),
+                photo=card["photo"],
+                description=card.get("description", None),
+            )
+            session.add(db_card)
         await session.commit()
 
 
-async def get_card(card_id: int):
+async def get_card(card_id: int) -> Card | None:
     async with AsyncSession(engine) as session:
-        card: Card = (await session.execute(select(Card).where(Card.id == card_id))).scalar_one_or_none()
+        card: Card | None = (
+            await session.execute(select(Card).where(Card.id == card_id))
+        ).scalar_one_or_none()
         return card
 
 
-async def get_all_cards():
+async def get_all_cards() -> list[Card]:
     async with AsyncSession(engine) as session:
         cards = (await session.execute(select(Card))).scalars().all()
-        return cards
+        return list(cards)
+
+
+async def get_lcard(cat_id: int) -> LimitedCards | None:
+    async with AsyncSession(engine) as session:
+        cat: LimitedCards | None = (
+            await session.execute(select(LimitedCards).where(LimitedCards.id == cat_id))
+        ).scalar_one_or_none()
+        return cat
 
-async def get_lcard(cat_id: int):
-   async with AsyncSession(engine) as session:
-       cat: LimitedCards = (await session.execute(select(LimitedCards).where(LimitedCards.id == cat_id))).scalar_one_or_none()
-       return cat
 
+async def get_all_lcards() -> list[LimitedCards]:
+    async with AsyncSession(engine) as session:
+        cats = (await session.execute(select(LimitedCards))).scalars().all()
+        return list(cats)
 
-async def get_all_lcards():
-   async with AsyncSession(engine) as session:
-       cats = (await session.execute(select(LimitedCards))).scalars().all() 
-       return cats
-
 
 async def increment_buy_count(card_id: int):
-   async with AsyncSession(engine) as session:
-       card = (await session.execute(select(LimitedCards).where(LimitedCards.id == card_id))).scalar_one_or_none()
-       if card:
-           card.buy_count += 1
-           await session.commit()
+    """Increment card's buy count by it's ID
+
+    Args:
+        card_id (int): Card's id
+
+    Raises:
+        KeyError: If card didn't found
+    """    
+    async with AsyncSession(engine) as session:
+        card = (
+            await session.execute(
+                select(LimitedCards).where(LimitedCards.id == card_id)
+            )
+        ).scalar_one_or_none()
+        if card is None:
+            raise KeyError(f"Didn't found card with id {card_id}")
+
+        card.buy_count += 1
+        await session.commit()