[WebApp] Add logic of web_app form to telegram

apps/web_app/.env.dev

@@ -1,6 +1,5 @@
 DERISK_API_URL=#
 
-ENV=dev
 # PostgreSQL
 DB_USER=postgres
 DB_PASSWORD=password

apps/web_app/database/models.py

@@ -35,7 +35,7 @@ class NotificationData(Base):
     email = Column(String, index=True, nullable=True)
     wallet_id = Column(String, nullable=False)
     telegram_id = Column(String, unique=False, nullable=False)
-    ip_address = Column(IPAddressType, nullable=False)
+    ip_address = Column(IPAddressType, nullable=True)
     health_ratio_level = Column(Float, nullable=False)
     protocol_id = Column(ChoiceType(ProtocolIDs, impl=String()), nullable=False)
 

apps/web_app/migrations/versions/b3179b2fff8b_ip_address_nullable_true.py

@@ -0,0 +1,42 @@
+"""ip address nullable true
+
+Revision ID: b3179b2fff8b
+Revises: f4baaac5103f
+Create Date: 2024-11-29 18:44:44.613470
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlalchemy_utils
+
+
+# revision identifiers, used by Alembic.
+revision: str = "b3179b2fff8b"
+down_revision: Union[str, None] = "f4baaac5103f"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column(
+        "notification",
+        "ip_address",
+        existing_type=sa.VARCHAR(length=50),
+        nullable=True,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column(
+        "notification",
+        "ip_address",
+        existing_type=sa.VARCHAR(length=50),
+        nullable=False,
+    )
+    # ### end Alembic commands ###

apps/web_app/telegram/__main__.py

@@ -25,6 +25,6 @@ async def bot_start_polling():
 
 
 if __name__ == "__main__":
-    if os.getenv("ENV") == "DEV":
+    if bot is not None:
         loop = asyncio.get_event_loop()
         loop.run_until_complete(bot_start_polling())

apps/web_app/telegram/crud.py

@@ -108,3 +108,14 @@ async def get_objects_by_filter(
             if limit == 1:
                 return await db.scalar(stmp)
             return await db.scalars(stmp).all()
+
+    async def write_to_db(self, obj: ModelType) -> None:
+        """
+        Write an object to the database.
+
+        Args:
+            obj (ModelType): The object to be added to the database.
+        """
+        async with self.Session() as db:
+            db.add(obj)
+            await db.commit()

apps/web_app/telegram/handlers/__init__.py

@@ -2,8 +2,10 @@
 
 from .command import cmd_router
 from .menu import menu_router
+from .create_notification import create_notification_router
 
 # Create the main router to simplify imports
 index_router = Router()
 index_router.include_router(cmd_router)
 index_router.include_router(menu_router)
+index_router.include_router(create_notification_router)

apps/web_app/telegram/handlers/create_notification.py

@@ -0,0 +1,99 @@
+from aiogram import F, Router, types
+from aiogram.fsm.context import FSMContext
+from aiogram.fsm.state import State, StatesGroup
+from database.models import NotificationData
+from telegram.crud import TelegramCrud
+from .utils import kb
+
+create_notification_router = Router()
+
+
+class NotificationFormStates(StatesGroup):
+    """States for the notification form process."""
+
+    wallet_id = State()
+    health_ratio_level = State()
+    protocol_id = State()
+
+
+# Define constants for health ratio limits
+HEALTH_RATIO_MIN = 0
+HEALTH_RATIO_MAX = 10
+
+
+@create_notification_router.callback_query(F.data == "create_subscription")
+async def start_form(callback: types.CallbackQuery, state: FSMContext):
+    """Initiates the notification creation form by asking for the wallet ID."""
+    await state.set_state(NotificationFormStates.wallet_id)
+    return callback.message.edit_text(
+        "Please enter your wallet ID:",
+        reply_markup=kb.cancel_form(),
+    )
+
+
+@create_notification_router.message(NotificationFormStates.wallet_id)
+async def process_wallet_id(message: types.Message, state: FSMContext):
+    """Processes the wallet ID input from the user."""
+    await state.update_data(wallet_id=message.text)
+    await state.set_state(NotificationFormStates.health_ratio_level)
+    return message.answer(
+        "Please enter your health ratio level (between 0 and 10):",
+        reply_markup=kb.cancel_form(),
+    )
+
+
+@create_notification_router.message(NotificationFormStates.health_ratio_level)
+async def process_health_ratio(message: types.Message, state: FSMContext):
+    """Processes the health ratio level input from the user."""
+    try:
+        health_ratio = float(message.text)
+        if not (HEALTH_RATIO_MIN <= health_ratio <= HEALTH_RATIO_MAX):
+            raise ValueError
+    except ValueError:
+        return message.answer(
+            "Please enter a valid number between 0 and 10.",
+            reply_markup=kb.cancel_form(),
+        )
+
+    await state.update_data(health_ratio_level=health_ratio)
+    await state.set_state(NotificationFormStates.protocol_id)
+
+    return message.answer(
+        "Please select your protocol:",
+        reply_markup=kb.protocols(),
+    )
+
+
+@create_notification_router.callback_query(F.data.startswith("protocol_"))
+async def process_protocol(
+    callback: types.CallbackQuery, state: FSMContext, crud: TelegramCrud
+):
+    """Processes the selected protocol and saves the subscription data."""
+    protocol_id = callback.data.replace("protocol_", "")
+    data = await state.get_data()
+
+    subscription = NotificationData(
+        wallet_id=data["wallet_id"],
+        health_ratio_level=data["health_ratio_level"],
+        protocol_id=protocol_id,
+        telegram_id=str(callback.from_user.id),
+        email=None,
+        ip_address=None,
+    )
+    await crud.write_to_db(subscription)
+
+    await state.clear()
+    return callback.message.edit_text(
+        "Subscription created successfully!",
+        reply_markup=kb.go_menu(),
+    )
+
+
+@create_notification_router.callback_query(F.data == "cancel_form")
+async def cancel_form(callback: types.CallbackQuery, state: FSMContext):
+    """Cancels the form and clears the state."""
+    await state.clear()
+    return callback.message.edit_text(
+        "Form cancelled.",
+        reply_markup=kb.go_menu(),
+    )

apps/web_app/telegram/handlers/utils/kb.py

@@ -2,6 +2,7 @@
 
 from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
 from aiogram.utils.keyboard import InlineKeyboardBuilder
+from utils.values import ProtocolIDs
 
 
 def go_menu():
@@ -69,6 +70,11 @@ def menu():
                     text="Shows notifications", callback_data="show_notifications"
                 ),
             ],
+            [
+                InlineKeyboardButton(
+                    text="Create subscription", callback_data="create_subscription"
+                )
+            ],
             [
                 InlineKeyboardButton(
                     text="Unsubscribe all", callback_data="all_unsubscribe"
@@ -98,3 +104,21 @@ def pagination_notifications(curent_uuid: UUID, page: int):
     if page == 0:
         markup.adjust(2, 1, 1)
     return markup.as_markup()
+
+def cancel_form():
+    """
+    Returns an InlineKeyboardMarkup with a single button labeled "Cancel" with the callback data "cancel_form".
+    """
+    return InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Cancel", callback_data="cancel_form")]])
+
+
+def protocols():
+    """
+    Returns an InlineKeyboardMarkup with buttons for each protocol.
+    """
+    # Create protocol selection buttons
+    markup = InlineKeyboardBuilder()
+    for protocol in ProtocolIDs:
+        markup.button(text=protocol.name, callback_data=f"protocol_{protocol.value}")
+    markup.button(text="Cancel", callback_data="cancel_form")
+    return markup.as_markup()