Merge pull request #99 from ilyarolf/develop

Develop

Author: ilyarolf

.env

@@ -13,4 +13,5 @@ BOT_LANGUAGE = ""
 MULTIBOT = ""
 ETHPLORER_API_KEY = ""
 CURRENCY = ""
-RUNTIME_ENVIRONMENT = "dev"
+RUNTIME_ENVIRONMENT = "DEV"
+WEBHOOK_SECRET_TOKEN = ""

bot.py

@@ -1,46 +1,79 @@
 import logging
+import traceback
+
 from aiogram.client.default import DefaultBotProperties
+from aiogram.fsm.storage.redis import RedisStorage
+from aiogram.types import BufferedInputFile
+from redis.asyncio import Redis
 import config
 from aiogram import Bot, Dispatcher
 from aiogram.enums import ParseMode
-from aiogram.fsm.storage.memory import MemoryStorage
-from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
-from aiohttp import web
-
-from config import TOKEN, WEBHOOK_URL, ADMIN_ID_LIST
+from fastapi import FastAPI, Request, status, HTTPException
+from config import TOKEN, WEBHOOK_URL, ADMIN_ID_LIST, WEBHOOK_SECRET_TOKEN
 from db import create_db_and_tables
+import uvicorn
+from fastapi.responses import JSONResponse
+from services.notification import NotificationService
 
+redis = Redis()
 bot = Bot(TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
-dp = Dispatcher(storage=MemoryStorage())
+dp = Dispatcher(storage=RedisStorage(redis))
+app = FastAPI()
+
+
+@app.post(config.WEBHOOK_PATH)
+async def webhook(request: Request):
+    secret_token = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
+    if secret_token != WEBHOOK_SECRET_TOKEN:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
 
+    try:
+        update_data = await request.json()
+        await dp.feed_webhook_update(bot, update_data)
+        return {"status": "ok"}
+    except Exception as e:
+        logging.error(f"Error processing webhook: {e}")
+        return {"status": "error"}, status.HTTP_500_INTERNAL_SERVER_ERROR
 
-async def on_startup(bot: Bot):
+
+@app.on_event("startup")
+async def on_startup():
     await create_db_and_tables()
-    await bot.set_webhook(WEBHOOK_URL)
+    await bot.set_webhook(
+        url=WEBHOOK_URL,
+        secret_token=WEBHOOK_SECRET_TOKEN
+    )
     for admin in ADMIN_ID_LIST:
         try:
             await bot.send_message(admin, 'Bot is working')
         except Exception as e:
             logging.warning(e)
 
 
-async def on_shutdown(dp):
+@app.on_event("shutdown")
+async def on_shutdown():
     logging.warning('Shutting down..')
-
     await bot.delete_webhook()
     await dp.storage.close()
-    await dp.storage.wait_closed()
-
     logging.warning('Bye!')
 
 
-def main() -> None:
-    dp.startup.register(on_startup)
-    app = web.Application()
-    webhook_requests_handler = SimpleRequestHandler(
-        dispatcher=dp,
-        bot=bot
+@app.exception_handler(Exception)
+async def exception_handler(request: Request, exc: Exception):
+    traceback_str = traceback.format_exc()
+    admin_notification = (
+        f"Critical error caused by {exc}\n\n"
+        f"Stack trace:\n{traceback_str}"
+    )
+    if len(admin_notification) > 4096:
+        byte_array = bytearray(admin_notification, 'utf-8')
+        admin_notification = BufferedInputFile(byte_array, "exception.txt")
+    await NotificationService.send_to_admins(admin_notification, None)
+    return JSONResponse(
+        status_code=500,
+        content={"message": f"An error occurred: {str(exc)}"},
     )
-    webhook_requests_handler.register(app, path=config.WEBHOOK_PATH)
-    setup_application(app, dp, bot=bot)
-    web.run_app(app, host=config.WEBAPP_HOST, port=config.WEBAPP_PORT)
+
+
+def main() -> None:
+    uvicorn.run(app, host=config.WEBAPP_HOST, port=config.WEBAPP_PORT)

config.py

@@ -3,12 +3,13 @@
 from dotenv import load_dotenv
 
 from enums.currency import Currency
+from enums.runtime_environment import RuntimeEnvironment
 from external_ip import get_sslipio_external_url
 from ngrok_executor import start_ngrok
 
 load_dotenv(".env")
-
-if os.environ.get("RUNTIME_ENVIRONMENT") == "dev":
+RUNTIME_ENVIRONMENT = RuntimeEnvironment(os.environ.get("RUNTIME_ENVIRONMENT"))
+if RUNTIME_ENVIRONMENT == RuntimeEnvironment.DEV:
     WEBHOOK_HOST = start_ngrok()
 else:
     WEBHOOK_HOST = get_sslipio_external_url()
@@ -28,3 +29,4 @@
 MULTIBOT = os.environ.get("MULTIBOT", False) == 'true'
 ETHPLORER_API_KEY = os.environ.get("ETHPLORER_API_KEY")
 CURRENCY = Currency(os.environ.get("CURRENCY"))
+WEBHOOK_SECRET_TOKEN = os.environ.get("WEBHOOK_SECRET_TOKEN")

crypto_api/CryptoApiManager.py

@@ -1,5 +1,8 @@
 from datetime import datetime, timedelta
 import aiohttp
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+
 import config
 from enums.cryptocurrency import Cryptocurrency
 from models.deposit import DepositDTO
@@ -19,7 +22,7 @@ async def fetch_api_request(url: str, params: dict | None = None) -> dict:
                     return data
 
     @staticmethod
-    async def get_new_btc_deposits(user_dto: UserDTO, deposits) -> float:
+    async def get_new_btc_deposits(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session) -> float:
         url = f'https://mempool.space/api/address/{user_dto.btc_address}/utxo'
         data = await CryptoApiManager.fetch_api_request(url)
         deposits = [deposit.tx_id for deposit in deposits if deposit.network == "BTC"]
@@ -33,12 +36,12 @@ async def get_new_btc_deposits(user_dto: UserDTO, deposits) -> float:
                     amount=deposit['value'],
                     vout=deposit['vout']
                 )
-                await DepositService.create(deposit_dto)
+                await DepositService.create(deposit_dto, session)
                 deposit_sum += float(deposit["value"]) / 100_000_000
         return deposit_sum
 
     @staticmethod
-    async def get_new_ltc_deposits(user_dto: UserDTO, deposits) -> float:
+    async def get_new_ltc_deposits(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session) -> float:
         url = f"https://api.blockcypher.com/v1/ltc/main/addrs/{user_dto.ltc_address}"
         params = {"unspentOnly": "true"}
         data = await CryptoApiManager.fetch_api_request(url, params=params)
@@ -54,12 +57,12 @@ async def get_new_ltc_deposits(user_dto: UserDTO, deposits) -> float:
                         amount=deposit['value'],
                         vout=deposit['tx_output_n']
                     )
-                    await DepositService.create(deposit_dto)
+                    await DepositService.create(deposit_dto, session)
                     deposits_sum += float(deposit['value']) / 100_000_000
         return deposits_sum
 
     @staticmethod
-    async def get_sol_balance(user_dto: UserDTO, deposits) -> float:
+    async def get_sol_balance(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session) -> float:
         url = f"https://api.solana.fm/v0/accounts/{user_dto.sol_address}/transfers"
         data = await CryptoApiManager.fetch_api_request(url)
         deposits = [deposit.tx_id for deposit in deposits if deposit.network == "SOL"]
@@ -78,12 +81,12 @@ async def get_sol_balance(user_dto: UserDTO, deposits) -> float:
                                 amount=transfer['amount'],
                                 vout=transfer['instructionIndex']
                             )
-                            await DepositService.create(deposit_dto)
+                            await DepositService.create(deposit_dto, session)
                             deposits_sum += float(transfer['amount'] / 1_000_000_000)
         return deposits_sum
 
     @staticmethod
-    async def get_usdt_trc20_balance(user_dto: UserDTO, deposits) -> float:
+    async def get_usdt_trc20_balance(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session) -> float:
         url = f"https://api.trongrid.io/v1/accounts/{user_dto.trx_address}/transactions/trc20"
         params = {"only_confirmed": "true",
                   "min_timestamp": CryptoApiManager.min_timestamp,
@@ -102,12 +105,12 @@ async def get_usdt_trc20_balance(user_dto: UserDTO, deposits) -> float:
                     token_name='USDT_TRC20',
                     amount=deposit['value'],
                 )
-                await DepositService.create(deposit_dto)
+                await DepositService.create(deposit_dto, session)
                 deposits_sum += float(deposit['value']) / pow(10, deposit['token_info']['decimals'])
         return deposits_sum
 
     @staticmethod
-    async def get_usdt_erc20_balance(user_dto: UserDTO, deposits) -> float:
+    async def get_usdt_erc20_balance(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session) -> float:
         # TODO(Combine the function to obtain erc20 tokens.)
         url = f'https://api.ethplorer.io/getAddressHistory/{user_dto.eth_address}'
         params = {
@@ -129,12 +132,12 @@ async def get_usdt_erc20_balance(user_dto: UserDTO, deposits) -> float:
                     token_name='USDT_ERC20',
                     amount=deposit['value']
                 )
-                await DepositService.create(deposit_dto)
+                await DepositService.create(deposit_dto, session)
                 deposits_sum += float(deposit['value']) / pow(10, 6)
         return deposits_sum
 
     @staticmethod
-    async def get_usdc_erc20_balance(user_dto: UserDTO, deposits):
+    async def get_usdc_erc20_balance(user_dto: UserDTO, deposits: list[DepositDTO], session: AsyncSession | Session):
         # TODO(Combine the function to obtain erc20 tokens.)
         url = f'https://api.ethplorer.io/getAddressHistory/{user_dto.eth_address}'
         params = {
@@ -156,7 +159,7 @@ async def get_usdc_erc20_balance(user_dto: UserDTO, deposits):
                     token_name='USDC_ERC20',
                     amount=deposit['value']
                 )
-                await DepositService.create(deposit_dto)
+                await DepositService.create(deposit_dto, session)
                 deposits_sum += float(deposit['value']) / pow(10, 6)
         return deposits_sum
 
@@ -175,18 +178,19 @@ async def get_crypto_prices(cryptocurrency: Cryptocurrency) -> float:
         return float(next(iter(response_json['result'].values()))['c'][0])
 
     @staticmethod
-    async def get_new_deposits_amount(user_dto: UserDTO, cryptocurrency: Cryptocurrency):
-        deposits = await DepositService.get_by_user_dto(user_dto)
+    async def get_new_deposits_amount(user_dto: UserDTO, cryptocurrency: Cryptocurrency,
+                                      session: AsyncSession | Session):
+        deposits = await DepositService.get_by_user_dto(user_dto, session)
         match cryptocurrency:
             case Cryptocurrency.BTC:
-                return await CryptoApiManager.get_new_btc_deposits(user_dto, deposits)
+                return await CryptoApiManager.get_new_btc_deposits(user_dto, deposits, session)
             case Cryptocurrency.LTC:
-                return await CryptoApiManager.get_new_ltc_deposits(user_dto, deposits)
+                return await CryptoApiManager.get_new_ltc_deposits(user_dto, deposits, session)
             case Cryptocurrency.SOL:
-                return await CryptoApiManager.get_sol_balance(user_dto, deposits)
+                return await CryptoApiManager.get_sol_balance(user_dto, deposits, session)
             case Cryptocurrency.USDT_TRC20:
-                return await CryptoApiManager.get_usdt_trc20_balance(user_dto, deposits)
+                return await CryptoApiManager.get_usdt_trc20_balance(user_dto, deposits, session)
             case Cryptocurrency.USDT_ERC20:
-                return await CryptoApiManager.get_usdt_erc20_balance(user_dto, deposits)
+                return await CryptoApiManager.get_usdt_erc20_balance(user_dto, deposits, session)
             case Cryptocurrency.USDC_ERC20:
-                return await CryptoApiManager.get_usdc_erc20_balance(user_dto, deposits)
+                return await CryptoApiManager.get_usdc_erc20_balance(user_dto, deposits, session)

db.py

@@ -73,11 +73,11 @@ async def session_execute(stmt, session: AsyncSession | Session) -> Result[Any]
         return query_result
 
 
-async def session_refresh(session: AsyncSession | Session, instance: object) -> None:
+async def session_flush(session: AsyncSession | Session) -> None:
     if isinstance(session, AsyncSession):
-        await session.refresh(instance)
+        await session.flush()
     else:
-        session.refresh(instance)
+        session.flush()
 
 
 async def session_commit(session: AsyncSession | Session) -> None:
@@ -91,6 +91,8 @@ async def session_commit(session: AsyncSession | Session) -> None:
 def set_sqlite_pragma(dbapi_connection, connection_record):
     cursor = dbapi_connection.cursor()
     cursor.execute("PRAGMA foreign_keys=ON")
+    if config.DB_ENCRYPTION:
+        cursor.execute("PRAGMA journal_mode=WAL")
     cursor.close()
 
 

docker-compose.yml

@@ -35,7 +35,8 @@ services:
       MULTIBOT: "false" # Allows the use of a multibot
       ETHPLORER_API_KEY: "" # API key from Ethplorer
       CURRENCY: "USD" # fiat currency
-      RUNTIME_ENVIRONMENT: "prod"
+      RUNTIME_ENVIRONMENT: "PROD"
+      WEBHOOK_SECRET_TOKEN: "1234567890" # Any string you want
     labels:
       caddy: YOUR-DOMAIN-GOES-HERE
       caddy.reverse_proxy: "bot:5000"
@@ -51,11 +52,21 @@ services:
       - 5000 # ${WEBAPP_PORT}
     volumes:
       - ./AiogramShopBot:/bot/data # endswith your database name
-    command: ["python", "-u", "run.py"]
+    command: [ "python", "-u", "run.py" ]
+
+  redis:
+    image: redis:latest
+    container_name: redis
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+    restart: always
 
 volumes:
   AiogramShopBot:
   caddy_data:
+  redis_data:
 
 networks:
   caddy:

enums/cryptocurrency.py

@@ -1,7 +1,7 @@
 from enum import Enum
 
 
-class Cryptocurrency(Enum):
+class Cryptocurrency(str, Enum):
     BTC = "BTC"
     LTC = "LTC"
     SOL = "SOL"

enums/runtime_environment.py

@@ -0,0 +1,6 @@
+from enum import Enum
+
+
+class RuntimeEnvironment(str, Enum):
+    DEV = "DEV"
+    PROD = "PROD"

handlers/admin/admin.py

@@ -24,10 +24,11 @@
 
 @admin_router.message(F.text == Localizator.get_text(BotEntity.ADMIN, "menu"), AdminIdFilter())
 async def admin_command_handler(message: types.message):
-    await admin(message)
+    await admin(message=message)
 
 
-async def admin(message: Message | CallbackQuery):
+async def admin(**kwargs):
+    message = kwargs.get("message") or kwargs.get("callback")
     admin_menu_builder = InlineKeyboardBuilder()
     admin_menu_builder.button(text=Localizator.get_text(BotEntity.ADMIN, "announcements"),
                               callback_data=AdminAnnouncementCallback.create(level=0))
@@ -58,7 +59,10 @@ async def admin_menu_navigation(callback: CallbackQuery, state: FSMContext, call
     }
 
     current_level_function = levels[current_level]
-    if inspect.getfullargspec(current_level_function).annotations.get("state") == FSMContext:
-        await current_level_function(callback, state)
-    else:
-        await current_level_function(callback)
+
+    kwargs = {
+        "callback": callback,
+        "state": state,
+    }
+
+    await current_level_function(**kwargs)