86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
"""Telegram bot handlers module."""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
|
|
from telebot.async_telebot import AsyncTeleBot
|
|
from telebot import types
|
|
|
|
from config import States
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class BotState:
|
|
"""Bot conversation state manager."""
|
|
status_cache: dict[int, States] = field(default_factory=dict)
|
|
awaiting_torrents: dict[int, str] = field(default_factory=dict)
|
|
|
|
def get_state(self, chat_id: int) -> States:
|
|
"""Get the current state for a chat."""
|
|
return self.status_cache.get(chat_id, States.NONE)
|
|
|
|
def set_state(self, chat_id: int, state: States) -> None:
|
|
"""Set the state for a chat."""
|
|
self.status_cache[chat_id] = state
|
|
|
|
def clear_state(self, chat_id: int) -> None:
|
|
"""Clear the state for a chat."""
|
|
self.status_cache.pop(chat_id, None)
|
|
|
|
def set_awaiting_torrent(self, chat_id: int, magnet: str) -> None:
|
|
"""Store awaiting torrent magnet link."""
|
|
self.awaiting_torrents[chat_id] = magnet
|
|
|
|
def get_awaiting_torrent(self, chat_id: int) -> str | None:
|
|
"""Get awaiting torrent magnet link."""
|
|
return self.awaiting_torrents.pop(chat_id, None)
|
|
|
|
|
|
def create_main_keyboard() -> types.ReplyKeyboardMarkup:
|
|
"""Create the main menu keyboard."""
|
|
markup = types.ReplyKeyboardMarkup(
|
|
row_width=2,
|
|
resize_keyboard=True,
|
|
one_time_keyboard=True,
|
|
)
|
|
markup.row(types.KeyboardButton("qBittorrent"), types.KeyboardButton("Jellyfin"))
|
|
return markup
|
|
|
|
|
|
def create_qbittorrent_keyboard() -> types.ReplyKeyboardMarkup:
|
|
"""Create the qBittorrent menu keyboard."""
|
|
markup = types.ReplyKeyboardMarkup(
|
|
row_width=2,
|
|
resize_keyboard=True,
|
|
one_time_keyboard=True,
|
|
)
|
|
markup.row(types.KeyboardButton("Add torrent"), types.KeyboardButton("Delete torrent"))
|
|
markup.add(types.KeyboardButton("Get list of active torrents"))
|
|
markup.row(types.KeyboardButton("Get total stats"), types.KeyboardButton("Go back"))
|
|
return markup
|
|
|
|
|
|
def create_jellyfin_keyboard() -> types.ReplyKeyboardMarkup:
|
|
"""Create the Jellyfin menu keyboard."""
|
|
markup = types.ReplyKeyboardMarkup(
|
|
row_width=2,
|
|
resize_keyboard=True,
|
|
one_time_keyboard=True,
|
|
)
|
|
markup.add(types.KeyboardButton("Refresh Jellyfin"))
|
|
return markup
|
|
|
|
|
|
def create_category_keyboard(categories: list[str]) -> types.ReplyKeyboardMarkup:
|
|
"""Create a keyboard with torrent categories."""
|
|
markup = types.ReplyKeyboardMarkup(
|
|
row_width=3,
|
|
resize_keyboard=True,
|
|
one_time_keyboard=True,
|
|
)
|
|
for category in categories:
|
|
markup.add(category)
|
|
return markup
|