Add TimePad tickets importing

fanfan/application/services/notification.py

@@ -15,7 +15,10 @@
 from fanfan.infrastructure.scheduler import (
     redis_async_result,
 )
-from fanfan.infrastructure.scheduler.tasks import delete_message, send_notification
+from fanfan.infrastructure.scheduler.tasks.notifications import (
+    delete_message,
+    send_notification,
+)
 
 logger = logging.getLogger(__name__)
 

fanfan/application/services/quest.py

@@ -15,7 +15,7 @@
     UserNotFound,
 )
 from fanfan.application.services.base import BaseService
-from fanfan.infrastructure.scheduler.tasks import send_notification
+from fanfan.infrastructure.scheduler.tasks.notifications import send_notification
 
 logger = logging.getLogger(__name__)
 

fanfan/application/services/ticket.py

@@ -12,7 +12,6 @@
 from fanfan.application.exceptions.users import (
     UserNotFound,
 )
-from fanfan.application.services.access import check_permission
 from fanfan.application.services.base import BaseService
 from fanfan.common.enums import UserRole
 from fanfan.infrastructure.db.models import Ticket
@@ -21,7 +20,6 @@
 
 
 class TicketService(BaseService):
-    @check_permission(allowed_roles=[UserRole.ORG])
     async def create_ticket(self, ticket_id: str, role: UserRole) -> TicketDTO:
         """Create a new ticket"""
         async with self.uow:
@@ -54,3 +52,13 @@ async def link_ticket(self, ticket_id: str, user_id: int) -> None:
             await self.uow.commit()
             logger.info(f"Ticket id={ticket.id} was linked to user id={user.id}")
             return
+
+    async def delete_ticket(self, ticket_id: str) -> None:
+        ticket = await self.uow.tickets.get_ticket(ticket_id)
+        if not ticket:
+            raise TicketNotFound
+        async with self.uow:
+            await self.uow.session.delete(ticket)
+            await self.uow.commit()
+            logger.info(f"Ticket id={ticket_id} was deleted")
+            return

fanfan/config.py

@@ -154,6 +154,13 @@ def build_qr_scanner_url(self) -> str:
         return url.unicode_string()
 
 
+class TimepadConfig(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="TIMEPAD_")
+
+    client_id: Optional[SecretStr] = None
+    event_id: Optional[int] = None
+
+
 class DebugConfig(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="DEBUG_")
 
@@ -177,6 +184,7 @@ class Configuration:
     redis: RedisConfig = RedisConfig()
     bot: BotConfig = BotConfig()
     web: WebConfig = WebConfig()
+    timepad: TimepadConfig = TimepadConfig()
     debug: DebugConfig = DebugConfig()
 
 

fanfan/infrastructure/di/__init__.py

@@ -6,6 +6,7 @@
 from fanfan.infrastructure.di.config import ConfigProvider
 from fanfan.infrastructure.di.db import DbProvider
 from fanfan.infrastructure.di.redis import RedisProvider
+from fanfan.infrastructure.di.timepad import TimepadProvider
 from fanfan.infrastructure.di.user_bot import UserProvider
 
 
@@ -16,6 +17,7 @@ def get_app_providers() -> List[Provider]:
         DpProvider(),
         BotProvider(),
         RedisProvider(),
+        TimepadProvider(),
     ]
 
 

fanfan/infrastructure/di/config.py

@@ -1,6 +1,13 @@
 from dishka import Provider, Scope, provide
 
-from fanfan.config import BotConfig, DatabaseConfig, DebugConfig, RedisConfig, WebConfig
+from fanfan.config import (
+    BotConfig,
+    DatabaseConfig,
+    DebugConfig,
+    RedisConfig,
+    TimepadConfig,
+    WebConfig,
+)
 
 
 class ConfigProvider(Provider):
@@ -22,6 +29,10 @@ def get_bot_config(self) -> BotConfig:
     def get_web_config(self) -> WebConfig:
         return WebConfig()
 
+    @provide
+    def get_timepad_config(self) -> TimepadConfig:
+        return TimepadConfig()
+
     @provide
     def get_debug_config(self) -> DebugConfig:
         return DebugConfig()

fanfan/infrastructure/di/timepad.py

@@ -0,0 +1,26 @@
+from typing import AsyncIterable, NewType
+
+from aiohttp import ClientSession
+from dishka import Provider, Scope, provide
+
+from fanfan.config import TimepadConfig
+from fanfan.infrastructure.timepad.client import TimepadClient
+
+TimepadSession = NewType("TimepadSession", ClientSession)
+
+
+class TimepadProvider(Provider):
+    scope = Scope.APP
+
+    @provide(scope=Scope.REQUEST)
+    async def get_timepad_session(
+        self, config: TimepadConfig
+    ) -> AsyncIterable[TimepadSession]:
+        async with ClientSession(
+            headers={"Authorization": f"Bearer {config.client_id.get_secret_value()}"}
+        ) as session:
+            yield session
+
+    @provide(scope=Scope.REQUEST)
+    async def get_timepad_client(self, session: TimepadSession) -> TimepadClient:
+        return TimepadClient(session)

fanfan/infrastructure/scheduler/__init__.py

@@ -1,6 +1,8 @@
 import uuid
 from datetime import timedelta
 
+from taskiq import SimpleRetryMiddleware, TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
 from taskiq.serializers import ORJSONSerializer
 from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
 
@@ -17,5 +19,11 @@
     )
     .with_result_backend(redis_async_result)
     .with_serializer(ORJSONSerializer())
+    .with_middlewares(SimpleRetryMiddleware(default_retry_count=0))
     .with_id_generator(lambda: f"task:{uuid.uuid4().hex}")
 )
+
+scheduler = TaskiqScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker)],
+)

fanfan/infrastructure/scheduler/tasks/__init__.py

@@ -0,0 +1 @@
+

fanfan/infrastructure/scheduler/tasks.py → fanfan/infrastructure/scheduler/tasks/notifications.py

@@ -16,7 +16,7 @@
 logger = logging.getLogger("__name__")
 
 
-@broker.task
+@broker.task()
 @inject
 async def send_notification(
     notification: UserNotification,
@@ -53,7 +53,7 @@ async def send_notification(
         logger.info(f"Failed to send message to {notification.user_id}, skip")
 
 
-@broker.task
+@broker.task()
 @inject
 async def delete_message(
     message: Message,

fanfan/infrastructure/scheduler/tasks/tickets.py

@@ -0,0 +1,81 @@
+import asyncio
+import logging
+import math
+
+from dishka import FromDishka
+from dishka.integrations.taskiq import inject
+
+from fanfan.application.exceptions.ticket import TicketAlreadyExist, TicketNotFound
+from fanfan.application.services import TicketService
+from fanfan.common.enums import UserRole
+from fanfan.config import TimepadConfig
+from fanfan.infrastructure.db import UnitOfWork
+from fanfan.infrastructure.scheduler import broker
+from fanfan.infrastructure.timepad.client import TimepadClient
+from fanfan.infrastructure.timepad.models import OrderStatus
+
+ORDERS_PER_REQUEST = 100
+PARTICIPANT_NOMINATIONS = [
+    "Участник сценической программы",
+    "Участник не сценических конкурсов",
+]
+
+logger = logging.getLogger("__name__")
+
+
+@broker.task(schedule=[{"cron": "0 * * * *"}], retry_on_error=True, max_retries=3)
+@inject
+async def update_tickets(
+    client: FromDishka[TimepadClient],
+    config: FromDishka[TimepadConfig],
+    uow: FromDishka[UnitOfWork],
+) -> None:
+    if not (config.client_id or config.event_id):
+        logger.info(
+            "TimePad client id or event id was not provided, skipping importing"
+        )
+        return None
+    added_tickets, deleted_tickets = 0, 0
+    step = 0
+    service = TicketService(uow)
+    init = await client.get_orders(config.event_id)
+    logger.info(f"Tickets import started, about to process {init.total} orders")
+    while step != math.ceil(init.total / ORDERS_PER_REQUEST):
+        orders = await client.get_orders(
+            config.event_id, limit=ORDERS_PER_REQUEST, skip=step * ORDERS_PER_REQUEST
+        )
+        for order in orders.values:
+            if order.status.name in [
+                OrderStatus.PAID,
+                OrderStatus.OK,
+                OrderStatus.PAID_OFFLINE,
+                OrderStatus.PAID_UR,
+            ]:
+                for ticket in order.tickets:
+                    try:
+                        await service.create_ticket(
+                            ticket_id=ticket.number,
+                            role=UserRole.PARTICIPANT
+                            if ticket.ticket_type.name in PARTICIPANT_NOMINATIONS
+                            else UserRole.VISITOR,
+                        )
+                        added_tickets += 1
+                    except TicketAlreadyExist:
+                        pass
+            else:
+                for ticket in order.tickets:
+                    try:
+                        await service.delete_ticket(ticket_id=ticket.number)
+                        deleted_tickets += 1
+                    except TicketNotFound:
+                        pass
+        logger.info(
+            f"Ongoing import: {added_tickets} tickets added, {deleted_tickets} "
+            f"tickets deleted"
+        )
+        await asyncio.sleep(3)
+        step += 1
+    logger.info(
+        f"Import done: {added_tickets} tickets added, {deleted_tickets} tickets deleted"
+    )
+    return None

fanfan/infrastructure/timepad/client.py

@@ -0,0 +1,21 @@
+from aiohttp import ClientSession
+from dataclass_rest import get
+from dataclass_rest.http.aiohttp import AiohttpClient
+
+from fanfan.infrastructure.timepad.models import RegistrationOrdersResponse
+
+TIMEPAD_API_BASE_URL = "https://api.timepad.ru/"
+
+
+class TimepadClient(AiohttpClient):
+    def __init__(self, session: ClientSession):
+        super().__init__(base_url=TIMEPAD_API_BASE_URL, session=session)
+
+    @get("v1/events/{event_id}/orders")
+    async def get_orders(
+        self,
+        event_id: int,
+        limit: int = 10,
+        skip: int = 0,
+    ) -> RegistrationOrdersResponse:
+        pass

fanfan/infrastructure/timepad/models.py

@@ -0,0 +1,53 @@
+import enum
+from dataclasses import dataclass
+from typing import List
+
+
+class OrderStatus(enum.StrEnum):
+    NOT_PAID = "notpaid"
+    OK = "ok"
+    PAID = "paid"
+    INACTIVE = "inactive"
+    DELETED = "deleted"
+    BLOCKED = "blocked"
+    RETURNED = "returned"
+    BOOKED = "booked"
+    PENDING = "pending"
+    REJECTED = "rejected"
+    BOOKED_OFFLINE = "booked_offline"
+    PAID_OFFLINE = "paid_offline"
+    PAID_UR = "paid_ur"
+    TRANSFER_PAYMENT = "transfer_payment"
+    RETURN_PAYMENT_REQUEST = "return_payment_request"
+    RETURN_PAYMENT_REJECT = "return_payment_reject"
+    RETURN_ORG = "return_org"
+    RETURN_TP = "return_tp"
+
+
+@dataclass
+class TicketTypeResponse:
+    name: str
+
+
+@dataclass
+class TicketResponse:
+    number: str
+    ticket_type: TicketTypeResponse
+
+
+@dataclass
+class OrderStatusResponse:
+    name: OrderStatus
+    title: str
+
+
+@dataclass
+class RegistrationOrderResponse:
+    status: OrderStatusResponse
+    tickets: List[TicketResponse]
+
+
+@dataclass
+class RegistrationOrdersResponse:
+    total: int
+    values: List[RegistrationOrderResponse]

fanfan/presentation/__main__.py

@@ -15,13 +15,13 @@
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from starlette.middleware.cors import CORSMiddleware
 from starlette.middleware.sessions import SessionMiddleware
-from taskiq.api import run_receiver_task
+from taskiq.api import run_receiver_task, run_scheduler_task
 
 from fanfan.application.services import SettingsService
 from fanfan.common.enums import BotMode
 from fanfan.config import get_config
 from fanfan.infrastructure.db import UnitOfWork
-from fanfan.infrastructure.scheduler import broker
+from fanfan.infrastructure.scheduler import broker, scheduler
 from fanfan.presentation.admin import setup_admin
 from fanfan.presentation.tgbot.web.webapp import webapp_router
 from fanfan.presentation.tgbot.web.webhook import webhook_router
@@ -50,7 +50,8 @@ async def lifespan(app: FastAPI):
 
     # Run scheduler
     setup_dishka_taskiq(app_container, broker)
-    worker_task = asyncio.create_task(run_receiver_task(broker, run_startup=True))
+    worker_task = asyncio.create_task(run_receiver_task(broker))
+    scheduler_task = asyncio.create_task(run_scheduler_task(scheduler))
 
     async with app_container() as request_container:
         uow = await request_container.get(UnitOfWork)
@@ -78,7 +79,8 @@ async def lifespan(app: FastAPI):
             bot_task = asyncio.create_task(dp.start_polling(bot))
             logger.info("Running in polling mode")
     yield
-    logger.info("Stopping scheduler worker...")
+    logger.info("Stopping scheduler...")
+    scheduler_task.cancel()
     worker_task.cancel()
     await broker.shutdown()
     logger.info("Stopping bot...")
@@ -126,6 +128,7 @@ def create_app() -> FastAPI:
             host=config.web.host,
             port=config.web.port,
             app=create_app(),
+            log_level=logging.ERROR,
         )
     except KeyboardInterrupt:
         pass