microservices are able to run (#5)

bot_microservice/core/bot.py

@@ -0,0 +1,77 @@
+import asyncio
+import os
+from asyncio import Queue, sleep
+from dataclasses import dataclass
+from functools import cached_property
+from http import HTTPStatus
+from typing import Any
+
+from fastapi import Request, Response
+from loguru import logger
+from settings.config import AppSettings
+from telegram import Update
+from telegram.ext import Application
+
+
+class BotApplication:
+    def __init__(
+        self,
+        settings: AppSettings,
+        handlers: list[Any],
+        application: Application | None = None,  # type: ignore[type-arg]
+    ) -> None:
+        self.application: Application = application or (  # type: ignore
+            Application.builder().token(token=settings.TELEGRAM_API_TOKEN).build()
+        )
+        self.handlers = handlers
+        self.settings = settings
+        self.start_with_webhook = settings.START_WITH_WEBHOOK
+        self._add_handlers()
+
+    async def set_webhook(self) -> None:
+        await self.application.initialize()
+        await self.application.bot.set_webhook(url=self.webhook_url)
+        logger.info('webhook is set')
+
+    async def delete_webhook(self) -> None:
+        await self.application.bot.delete_webhook()
+        logger.info('webhook has been deleted')
+
+    async def polling(self) -> None:
+        await self.application.initialize()
+        await self.application.start()
+        await self.application.updater.start_polling()  # type: ignore
+        logger.info("bot started in polling mode")
+
+    async def shutdown(self) -> None:
+        await self.application.updater.shutdown()  # type: ignore
+
+    @cached_property
+    def webhook_url(self) -> str:
+        return os.path.join(self.settings.DOMAIN.strip("/"), self.settings.bot_webhook_url.strip("/"))
+
+    def _add_handlers(self) -> None:
+        for handler in self.handlers:
+            self.application.add_handler(handler)
+
+
+@dataclass
+class BotQueue:
+    bot_app: BotApplication
+    queue: Queue = asyncio.Queue()  # type: ignore[type-arg]
+
+    async def put_updates_on_queue(self, request: Request) -> Response:
+        """
+        Listen /{URL_PREFIX}/{API_PREFIX}/{TELEGRAM_WEB_TOKEN} path and proxy post request to bot
+        """
+        data = await request.json()
+        tg_update = Update.de_json(data=data, bot=self.bot_app.application.bot)
+        self.queue.put_nowait(tg_update)
+
+        return Response(status_code=HTTPStatus.ACCEPTED)
+
+    async def get_updates_from_queue(self) -> None:
+        while True:
+            update = await self.queue.get()
+            await self.bot_app.application.process_update(update)
+            await sleep(0)

bot_microservice/core/commands.py

@@ -0,0 +1,74 @@
+import random
+import tempfile
+from uuid import uuid4
+
+import httpx
+from constants import CHAT_GPT_BASE_URL
+from core.utils import convert_file_to_wav
+from httpx import AsyncClient, AsyncHTTPTransport
+from loguru import logger
+from telegram import Update
+from telegram.ext import ContextTypes
+
+
+async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Send a message when the command /help is issued."""
+
+    if update.message:
+        await update.message.reply_text(
+            "Help!",
+            disable_notification=True,
+            api_kwargs={"text": "Hello World"},
+        )
+    return None
+
+
+async def ask_question(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    await update.message.reply_text(  # type: ignore[union-attr]
+        "Пожалуйста подождите, ответ в среднем занимает 10-15 секунд"
+    )
+
+    chat_gpt_request = {
+        "conversation_id": str(uuid4()),
+        "action": "_ask",
+        "model": "gpt-3.5-turbo",
+        "jailbreak": "default",
+        "meta": {
+            "id": random.randint(10**18, 10**19 - 1),  # noqa: S311
+            "content": {
+                "conversation": [],
+                "internet_access": False,
+                "content_type": "text",
+                "parts": [{"content": update.message.text, "role": "user"}],  # type: ignore[union-attr]
+            },
+        },
+    }
+
+    transport = AsyncHTTPTransport(retries=1)
+    async with AsyncClient(transport=transport) as client:
+        try:
+            response = await client.post(CHAT_GPT_BASE_URL, json=chat_gpt_request)
+            status = response.status_code
+            if status != httpx.codes.OK:
+                logger.info(f'got response status: {status} from chat api', data=chat_gpt_request)
+                await update.message.reply_text(  # type: ignore[union-attr]
+                    "Что-то пошло не так, попробуйте еще раз или обратитесь к администратору"
+                )
+                return
+
+            data = response.json()
+            await update.message.reply_text(data)  # type: ignore[union-attr]
+        except Exception as error:
+            logger.error("error get data from chat api", error=error)
+            await update.message.reply_text("Вообще всё сломалось :(")  # type: ignore[union-attr]
+
+
+async def voice_recognize(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    await update.message.reply_text(  # type: ignore[union-attr]
+        "Пожалуйста, ожидайте :)\nТрехминутная запись обрабатывается примерно 30 секунд"
+    )
+    sound_bytes = await update.message.voice.get_file()  # type: ignore[union-attr]
+    sound_bytes = await sound_bytes.download_as_bytearray()
+    with tempfile.NamedTemporaryFile(delete=False) as tmpfile:
+        tmpfile.write(sound_bytes)
+        convert_file_to_wav(tmpfile.name)

bot_microservice/core/handlers.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+from core.commands import ask_question, help_command, voice_recognize
+from telegram.ext import CommandHandler, MessageHandler, filters
+
+
+@dataclass
+class CommandHandlers:
+    handlers: list[Any] = field(default_factory=list[Any])
+
+    def add_handler(self, handler: Any) -> None:
+        self.handlers.append(handler)
+
+
+command_handlers = CommandHandlers()
+
+command_handlers.add_handler(CommandHandler("help", help_command))
+command_handlers.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, ask_question))
+command_handlers.add_handler(MessageHandler(filters.VOICE | filters.AUDIO, voice_recognize))

bot_microservice/core/logging.py

@@ -0,0 +1,102 @@
+import logging
+import sys
+from types import FrameType
+from typing import TYPE_CHECKING, Any, cast
+
+from constants import LogLevelEnum
+from loguru import logger
+from sentry_sdk.integrations.logging import EventHandler
+
+if TYPE_CHECKING:
+    from loguru import Record
+else:
+    Record = dict[str, Any]
+
+
+class InterceptHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord) -> None:
+        # Get corresponding Loguru level if it exists
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = str(record.levelno)
+
+        # Find caller from where originated the logged message
+        frame, depth = logging.currentframe(), 2
+        while frame.f_code.co_filename == logging.__file__:
+            frame = cast(FrameType, frame.f_back)
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(
+            level,
+            record.getMessage(),
+        )
+
+
+def configure_logging(*, level: LogLevelEnum, enable_json_logs: bool, enable_sentry_logs: bool) -> None:
+    logging_level = level.name
+
+    intercept_handler = InterceptHandler()
+
+    logging.basicConfig(handlers=[intercept_handler], level=logging_level)
+
+    formatter = _json_formatter if enable_json_logs else _text_formatter
+    logger.configure(
+        handlers=[
+            {
+                "sink": sys.stdout,
+                "level": logging_level,
+                "serialize": enable_json_logs,
+                "format": formatter,
+                "colorize": True,
+            }
+        ],
+    )
+
+    # sentry sdk не умеет из коробки работать с loguru, нужно добавлять хандлер
+    # https://github.com/getsentry/sentry-python/issues/653#issuecomment-788854865
+    # https://forum.sentry.io/t/changing-issue-title-when-logging-with-traceback/446
+    if enable_sentry_logs:
+        handler = EventHandler(level=logging.WARNING)
+        logger.add(handler, diagnose=True, level=logging.WARNING, format=_sentry_formatter)
+
+
+def _json_formatter(record: Record) -> str:
+    # Обрезаем `\n` в конце логов, т.к. в json формате переносы не нужны
+    return record.get("message", "").strip()
+
+
+def _sentry_formatter(record: Record) -> str:
+    return "{name}:{function} {message}"
+
+
+def _text_formatter(record: Record) -> str:
+    # WARNING !!!
+    # Функция должна возвращать строку, которая содержит только шаблоны для форматирования.
+    # Если в строку прокидывать значения из record (или еще откуда-либо),
+    # то loguru может принять их за f-строки и попытается обработать, что приведет к ошибке.
+    # Например, если нужно достать какое-то значение из поля extra, вместо того чтобы прокидывать его в строку формата,
+    # нужно прокидывать подстроку вида {extra[тут_ключ]}
+
+    # Стандартный формат loguru. Задается через env LOGURU_FORMAT
+    format_ = (
+        "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
+        "<level>{level: <8}</level> | "
+        "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
+    )
+
+    # Добавляем мета параметры по типу user_id, art_id, которые передаются через logger.bind(...)
+    extra = record["extra"]
+    if extra:
+        formatted = ", ".join(f"{key}" + "={extra[" + str(key) + "]}" for key, value in extra.items())
+        format_ += f" - <cyan>{formatted}</cyan>"
+
+    format_ += "\n"
+
+    if record["exception"] is not None:
+        format_ += "{exception}\n"
+
+    return format_
+
+
+configure_logging(level=LogLevelEnum.DEBUG, enable_json_logs=True, enable_sentry_logs=True)

bot_microservice/core/utils.py

@@ -0,0 +1,39 @@
+import subprocess  # noqa
+from datetime import datetime, timedelta
+from functools import lru_cache, wraps
+from typing import Any
+
+from loguru import logger
+
+
+def timed_cache(**timedelta_kwargs: Any) -> Any:
+    def _wrapper(func: Any) -> Any:
+        update_delta = timedelta(**timedelta_kwargs)
+        next_update = datetime.utcnow() + update_delta
+        # Apply @lru_cache to f with no cache size limit
+        cached_func = lru_cache(None)(func)
+
+        @wraps(func)
+        def _wrapped(*args: Any, **kwargs: Any) -> Any:
+            nonlocal next_update
+            now = datetime.utcnow()
+            if now >= next_update:
+                cached_func.cache_clear()
+                next_update = now + update_delta
+            return cached_func(*args, **kwargs)
+
+        return _wrapped
+
+    return _wrapper
+
+
+def convert_file_to_wav(filename: str) -> str:
+    new_filename = filename + '.wav'
+
+    cmd = ['ffmpeg', '-loglevel', 'quiet', '-i', filename, '-vn', new_filename]
+
+    try:
+        subprocess.run(args=cmd)  # noqa: S603
+    except Exception as error:
+        logger.error("cant convert voice: reason", error=error)
+    return new_filename