feat: add notify system to tg

src/application/contracts/__init__.py

@@ -1 +1,2 @@
-from src.application.contracts.i_logger import ILogger
+from src.application.contracts.i_logger import ILogger
+from src.application.contracts.i_sender_bot import ISenderBot

src/application/contracts/i_sender_bot.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+
+
+class ISenderBot(ABC):
+
+    @abstractmethod
+    async def send(
+        self,
+        text: str,
+    ) -> None:
+        raise NotImplementedError

src/application/domain/enums/__init__.py

@@ -1,2 +1,3 @@
 from src.application.domain.enums.log_level import LogLevel
-from src.application.domain.enums.log_format import LogFormat
+from src.application.domain.enums.log_format import LogFormat
+from src.application.domain.enums.notify_type import NotifyType

src/application/domain/enums/notify_type.py

@@ -0,0 +1,50 @@
+from enum import Enum
+
+
+class NotifyType(Enum):
+    DOCS = 10
+    PURCHASE = 20
+
+    def __str__(self) -> str:
+        return self.name
+
+    def __repr__(self) -> str:
+        return f"[{self.value}, '{self.name}']"
+
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, NotifyType):
+            return self.value == other.value
+        if isinstance(other, int):
+            return self.value == other
+        return NotImplemented
+
+    def __ne__(self, other: object) -> bool:
+        return not self.__eq__(other)
+
+    def __lt__(self, other: object) -> bool:
+        if isinstance(other, NotifyType):
+            return self.value < other.value
+        if isinstance(other, int):
+            return self.value < other
+        return NotImplemented
+
+    def __le__(self, other: object) -> bool:
+        if isinstance(other, NotifyType):
+            return self.value <= other.value
+        if isinstance(other, int):
+            return self.value <= other
+        return NotImplemented
+
+    def __gt__(self, other: object) -> bool:
+        if isinstance(other, NotifyType):
+            return self.value > other.value
+        if isinstance(other, int):
+            return self.value > other
+        return NotImplemented
+
+    def __ge__(self, other: object) -> bool:
+        if isinstance(other, NotifyType):
+            return self.value >= other.value
+        if isinstance(other, int):
+            return self.value >= other
+        return NotImplemented

src/infrastructure/config/settings.py

@@ -48,6 +48,13 @@ class Settings(BaseSettings):
     RABBIT_PUBLISH_PERSIST: bool = True
     RABBIT_CONNECT_TIMEOUT: int = 5
     RABBIT_EMAIL_CODE_QUEUE: str = "email.verification_code"
+    RABBIT_TELEGRAM_NOTIFY_QUEUE: str = "telegram.notify"
+
+    TELEGRAM_PROXY_URL: str = ""
+
+    TELEGRAM_BOT_TOKEN: str = ""
+    TELEGRAM_CHAT_ID: int = 0
+    TELEGRAM_MESSAGE_THREAD_ID: int | None = None
 
     LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
     LOG_FORMAT: Literal["JSON", "TEXT"] = "TEXT"
@@ -68,6 +75,7 @@ class Settings(BaseSettings):
         auth_mount = data.get('VAULT_AUTH_MOUNT') or os.getenv('VAULT_AUTH_MOUNT') or 'approle'
         mount = data.get('VAULT_MOUNT_POINT') or os.getenv('VAULT_MOUNT_POINT') or 'secrets'
 
+
         if not role_id or not secret_id:
             raise RuntimeError('VAULT_ROLE_ID and VAULT_SECRET_ID are required')
 
@@ -87,6 +95,7 @@ class Settings(BaseSettings):
 
         rabbitmq = safe_read("rabbitmq")
         email = safe_read("email")
+        telegram = safe_read("telegram")
 
 
         if rabbitmq:
@@ -101,8 +110,13 @@ class Settings(BaseSettings):
             data['SMTP_HOST'] = _vault_key_get(email, 'HOST', data.get('SMTP_HOST'))
             data['SMTP_PASSWORD'] = _vault_key_get(email, 'PASSWORD', data.get('SMTP_PASSWORD'))
             data['SMTP_PORT'] = _vault_key_get(email, 'PORT', data.get('SMTP_PORT'))
-            data['SMTP_USER'] = _vault_key_get(email, 'USER', data.get('SMTP_USER'))
+            data['SMTP_USER'] = _vault_key_get(email, 'USER', data.get('SMTP_USER'))        
 
+        if telegram:
+            data['TELEGRAM_BOT_TOKEN'] = _vault_key_get(telegram, 'BOT_TOKEN', data.get('TELEGRAM_BOT_TOKEN'))
+            data['TELEGRAM_CHAT_ID'] = _vault_key_get(telegram, 'CHAT_ID', data.get('TELEGRAM_CHAT_ID'))
+            data['TELEGRAM_MESSAGE_THREAD_ID'] = _vault_key_get(telegram, 'MESSAGE_THREAD_ID', data.get('TELEGRAM_MESSAGE_THREAD_ID'))
+            data['TELEGRAM_PROXY_URL'] = _vault_key_get(telegram, 'PROXY_URL', data.get('TELEGRAM_PROXY_URL'))
         return data
 
     @property

src/infrastructure/rabbit/broker.py

@@ -8,4 +8,9 @@ broker = RabbitBroker(settings.RABBIT_URL)
 email_code_queue = RabbitQueue(
     name=settings.RABBIT_EMAIL_CODE_QUEUE,
     durable=True,
+)
+
+telegram_notify_queue = RabbitQueue(
+    name=settings.RABBIT_TELEGRAM_NOTIFY_QUEUE,
+    durable=True,
 )

src/infrastructure/telegram/__init__.py

@@ -0,0 +1,14 @@
+from anyio.functools import lru_cache
+from src.application.contracts import ISenderBot
+from src.infrastructure.config import settings
+from src.infrastructure.telegram.render import TemplateTelegramRenderer
+from src.infrastructure.telegram.sender import NotifySender
+
+@lru_cache(maxsize=1)
+def get_renderer() -> TemplateTelegramRenderer:
+    return TemplateTelegramRenderer()
+
+
+@lru_cache(maxsize=1)
+def get_notify_sender() -> ISenderBot:
+    return NotifySender()

src/infrastructure/telegram/render.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+from jinja2 import Environment, FileSystemLoader, select_autoescape
+
+
+class TemplateTelegramRenderer:
+    def __init__(self, templates_dir: Path | str | None = None):
+        if templates_dir is None:
+            templates_dir = Path(__file__).parent / "templates"
+        self._env = Environment(
+            loader=FileSystemLoader(templates_dir),
+            autoescape=select_autoescape(["html"]),
+            trim_blocks=True,
+            lstrip_blocks=True,
+        )
+
+    def render(self, template_name: str, **kwargs: object) -> str:
+        return self._env.get_template(template_name).render(**kwargs)

src/infrastructure/telegram/sender.py

@@ -0,0 +1,17 @@
+from aiohttp import ClientSession
+from src.application.contracts import ISenderBot
+from src.infrastructure.config import settings
+from fastapi import HTTPException
+
+
+class NotifySender(ISenderBot):
+
+    async def send(self, client: ClientSession, text: str) -> None:
+        url = f'https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage'
+        payload = {
+            'chat_id': settings.TELEGRAM_CHAT_ID,
+            'text': text,
+        }
+        await client.post(url, proxy=settings.TELEGRAM_PROXY_URL, json=payload)
+
+

src/infrastructure/telegram/templates/docs_notify.txt

@@ -0,0 +1,3 @@
+#docs
+Клиент с ID: {{id_client}}
+Отправил документы на проверку

src/infrastructure/telegram/templates/purchase_notify.txt

@@ -0,0 +1,3 @@
+#purchase
+Клиент с ID: {{id_client}}
+Сделал запрос на обмен на сумму {{amount}} USDT

src/main.py

@@ -12,6 +12,7 @@ from src.infrastructure.logger import logger
 from src.infrastructure.config import settings
 from src.presentation.handlers import application_exception_handler, unhandled_exception_handler
 from src.presentation.messaging.code import code_broker
+from src.presentation.messaging.consumers import notify_broker
 from src.presentation.middleware import TraceIDMiddleware, SecurityHeadersMiddleware
 
 
@@ -59,6 +60,7 @@ app.add_exception_handler(Exception, unhandled_exception_handler)
 v1_router = APIRouter(prefix='/v1')
 app.include_router(v1_router)
 app.include_router(code_broker)
+app.include_router(notify_broker)
 
 
 # Added middleware

src/presentation/messaging/consumers.py

@@ -0,0 +1,69 @@
+from fastapi import Depends
+from decimal import Decimal
+from aiohttp import ClientSession
+from faststream.rabbit.fastapi import RabbitRouter, RabbitMessage
+from src.infrastructure.config import settings
+from src.infrastructure.logger import logger
+from src.infrastructure.telegram import TemplateTelegramRenderer, get_notify_sender, get_renderer
+from src.infrastructure.telegram import NotifySender
+from datetime import datetime
+from typing import Literal, Optional
+from pydantic import BaseModel
+from src.infrastructure.rabbit.broker import telegram_notify_queue
+
+
+notify_broker = RabbitRouter(settings.RABBIT_URL)
+
+
+class Metadata(BaseModel):
+    trace_id: Optional[str] = None
+    timestamp: datetime
+    message_id: str
+
+
+class Payload(BaseModel):
+    id_client: str 
+    usdt_amount: Optional[Decimal] = None
+
+
+class NotifyCreated(BaseModel):
+    event: Literal['docs_upload', 'purchase_created']
+    payload: Payload
+    metadata: Metadata
+
+
+@notify_broker.subscriber(telegram_notify_queue)
+async def consume_notify(
+    msg_body: NotifyCreated,
+    message: RabbitMessage,
+    sender: NotifySender = Depends(get_notify_sender),
+    renderer: TemplateTelegramRenderer = Depends(get_renderer),
+):
+    trace_id = (
+        (message.headers or {}).get("trace_id")
+        or message.correlation_id
+        or msg_body.metadata.trace_id
+    )
+
+    logger.info(
+        f"received event={msg_body.event} "
+        f"id_client={msg_body.payload.id_client} "
+        f"trace_id={trace_id}"
+    )
+
+    if msg_body.event == 'purchase_created':
+        text = renderer.render(
+            'purchase_notify.txt',
+            id_client=msg_body.payload.id_client,
+            amount=msg_body.payload.usdt_amount,
+            trace_id=trace_id,
+        )
+    elif msg_body.event == 'docs_upload':
+        text = renderer.render(
+            'docs_notify.txt',
+            id_client=msg_body.payload.id_client,
+            trace_id=trace_id
+        )
+
+    async with ClientSession() as session:
+        await sender.send(client=session, text=text)