#!/usr/bin/env python3 from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes from uptime_kuma_api import MonitorStatus import api.kuma as kuma import api.torrent as torrent import api.ntfy as ntfy # --- Menu Definitions --- def main_menu_keyboard(): return InlineKeyboardMarkup( [ [InlineKeyboardButton("Torrents", callback_data="status_downloading")], [InlineKeyboardButton("Status", callback_data="menu_status")], ] ) def torrents_menu_keyboard(): return InlineKeyboardMarkup( [ # First row: Display the status counts for downloading, paused, seeding [InlineKeyboardButton(f"Downloading", callback_data="status_downloading")], [InlineKeyboardButton(f"Active", callback_data="status_active")], [InlineKeyboardButton(f"All", callback_data="status_all")], # Second row: Back button [InlineKeyboardButton("πŸ”™ Back", callback_data="menu_main")], ] ) def status_menu_keyboard(): return InlineKeyboardMarkup( [[InlineKeyboardButton("πŸ”™ Back", callback_data="menu_main")]] ) def format_torrents(torrents): if len(torrents) == 0: return "No torrents." text = "" i = 0 for torrent in torrents: if i > 5: text += "...\n" return text text += f"Name: {torrent['name']}\n" text += f"State: {torrent['state']}\n" text += f"Progress: {torrent['progress']:.2f}%\n" text += f"ETA: {torrent['eta']}\n" text += "-" * 20 + "\n" text += f"- {torrent['name']} - {torrent['progress']} ({torrent['eta']})\n" i += 1 return text # --- Callback Query Handler --- async def handle_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() match query.data: case "menu_main": await query.edit_message_text( "Choose an option:", reply_markup=main_menu_keyboard() ) case "status_downloading": t_api = context.bot_data.get("torrent_api", {}) torrents = t_api.get_filtered_torrents("downloading") if len(torrents) == 0: text = "No downloading torrents." else: text = format_torrents(torrents) await query.edit_message_text(text, reply_markup=torrents_menu_keyboard()) case "status_active": t_api = context.bot_data.get("torrent_api", {}) torrents = t_api.get_filtered_torrents("active") if len(torrents) == 0: text = "No active torrents." else: text = format_torrents(torrents) await query.edit_message_text(text, reply_markup=torrents_menu_keyboard()) case "status_all": t_api = context.bot_data.get("torrent_api", {}) torrents = t_api.get_filtered_torrents("all") if len(torrents) == 0: text = "No torrents." else: text = format_torrents(torrents) await query.edit_message_text(text, reply_markup=torrents_menu_keyboard()) case "menu_status": k_api = context.bot_data.get("kuma_api", {}) monitors = k_api.get_status() up_text, down_text, paused_text = "", "", "" for _, monitor in monitors.items(): status = monitor["status"] if status == MonitorStatus.UP: up_text += f" - {monitor['name']}\n" elif status == MonitorStatus.DOWN: down_text += f" - {monitor['name']}\n" else: paused_text += f" - {monitor['name']}\n" status_text = f"πŸ“‘ *Status:*\n\n 🟒 Up:\n{up_text}\nπŸ”΄ Down\n{down_text}\n⏸️ Paused\n{paused_text}" await query.edit_message_text( status_text, reply_markup=status_menu_keyboard() ) case _: await query.edit_message_text( "Unknown option selected.", reply_markup=main_menu_keyboard() )