@@ -9,6 +9,8 @@ from aiogram.fsm.context import FSMContext
from aiogram . fsm . state import State , StatesGroup
from aiogram . fsm . storage . memory import MemoryStorage
from aiogram . types import (
BotCommand ,
BotCommandScopeChat ,
CallbackQuery ,
InlineKeyboardButton ,
InlineKeyboardMarkup ,
@@ -40,6 +42,18 @@ ADMIN_USERS: set[int] = {
bot = Bot ( token = BOT_TOKEN )
dp = Dispatcher ( storage = MemoryStorage ( ) )
USER_COMMANDS = [
BotCommand ( command = " start " , description = " Start the bot " ) ,
BotCommand ( command = " menu " , description = " Show available commands " ) ,
BotCommand ( command = " list " , description = " Browse saved memes " ) ,
BotCommand ( command = " cancel " , description = " Cancel current operation " ) ,
]
ADMIN_COMMANDS = [
* USER_COMMANDS ,
BotCommand ( command = " edit " , description = " Edit meme keywords " ) ,
BotCommand ( command = " delete " , description = " Remove a meme by ID " ) ,
]
def is_allowed ( user_id : int ) - > bool :
if user_id in ADMIN_USERS :
@@ -54,6 +68,20 @@ def is_admin(user_id: int) -> bool:
return not ALLOWED_USERS or user_id in ALLOWED_USERS
async def register_commands ( ) - > None :
""" Publish Telegram ' s command menu with role-appropriate suggestions. """
if not ADMIN_USERS :
await bot . set_my_commands ( ADMIN_COMMANDS )
return
await bot . set_my_commands ( USER_COMMANDS )
for admin_id in ADMIN_USERS :
await bot . set_my_commands (
ADMIN_COMMANDS ,
scope = BotCommandScopeChat ( chat_id = admin_id ) ,
)
class AddMedia ( StatesGroup ) :
waiting_for_keywords = State ( )
@@ -62,30 +90,43 @@ class EditMedia(StatesGroup):
waiting_for_new_keywords = State ( )
def menu_text ( user_id : int ) - > str :
if is_admin ( user_id ) :
return (
" Send me any <b>photo, GIF, video, voice, or sticker</b>. \n "
" I ' ll ask you for keywords next. \n \n "
" /list — show all saved memes \n "
" /edit [search] — edit keywords of a meme \n "
" /delete <id> — remove a meme by ID \n "
" /cancel — cancel current operation "
)
return (
" Use me inline: type <code>@botname keyword</code> in any chat. \n \n "
" /list — browse all saved memes "
)
# ── /start ───────────────────────────────────────────────────────────────────
@dp.message ( Command ( " start " ) )
async def cmd_start ( msg : Message ) :
if not is_allowed ( msg . from_user . id ) :
return
if is_admin ( msg . from_user . id ) :
await msg . answer (
" Send me any <b>photo, GIF, video, voice, or sticker</b>. \n "
" I ' ll ask you for keywords next. \n \n "
" /list — show all saved memes \n "
" /edit [search] — edit keywords of a meme \n "
" /delete <id> — remove a meme by ID \n "
" /cancel — cancel current operation " ,
parse_mode = " HTML " ,
)
else :
await msg . answer (
" Use me inline: type <code>@botname keyword</code> in any chat. \n \n "
" /list — browse all saved memes " ,
f " { menu_text ( msg . from_user . id ) } \n /menu — show this list again " ,
parse_mode = " HTML " ,
)
# ── /menu ────────────────────────────────────────────────────────────────────
@dp.message ( Command ( " menu " ) )
async def cmd_menu ( msg : Message ) :
if not is_allowed ( msg . from_user . id ) :
return
await msg . answer ( menu_text ( msg . from_user . id ) , parse_mode = " HTML " )
# ── /cancel ──────────────────────────────────────────────────────────────────
@dp.message ( Command ( " cancel " ) )
@@ -188,52 +229,21 @@ async def on_media_during_keywords(msg: Message):
# ── /list ────────────────────────────────────────────────────────────────────
LIST_PAGE_SIZE = 30
async def list_page ( page : int ) - > tuple [ str , InlineKeyboardMarkup | None ] :
total = await db . count_media ( )
if total == 0 :
return " No memes saved yet. " , None
total_pages = ( total + LIST_PAGE_SIZE - 1 ) / / LIST_PAGE_SIZE
page = max ( 0 , min ( page , total_pages - 1 ) )
items = await db . list_media ( page , LIST_PAGE_SIZE )
lines = [ f " <code> { item [ ' id ' ] } </code> [ { item [ ' media_type ' ] } ] — { item [ ' keywords ' ] } " for item in items ]
text = f " Page { page + 1 } of { total_pages } \n " + " \n " . join ( lines )
buttons = [ ]
if page > 0 :
buttons . append ( InlineKeyboardButton ( text = " ‹ " , callback_data = f " list: { page - 1 } " ) )
if page < total_pages - 1 :
buttons . append ( InlineKeyboardButton ( text = " Next › " , callback_data = f " list: { page + 1 } " ) )
keyboard = InlineKeyboardMarkup ( inline_keyboard = [ buttons ] ) if buttons else None
return text , keyboard
@dp.message ( Command ( " list " ) )
async def cmd_list ( msg : Message ) :
if not is_allowed ( msg . from_user . id ) :
return
text , keyboard = await list_page ( 0 )
await msg . answer ( text , parse_mode = " HTML " , reply_markup = keyboard )
@dp.callback_query ( F . data . startswith ( " list: " ) )
async def on_list_page ( callback : CallbackQuery ) :
if not is_allowed ( callback . from_user . id ) :
await callback . answer ( " Not allowed. " , show_alert = True )
items = await db . list_media ( )
if not items :
await msg . answer ( " No memes saved yet. " )
return
try :
page = int ( callback . data . split ( " : " , 1 ) [ 1 ] )
except ValueError :
await callback . answer ( " Invalid page. " , show_alert = True )
return
text , keyboard = await list_page ( page )
await callback . message . edit_text ( text , parse_mode = " HTML " , reply_markup = keyboard )
await callback . answer ( )
lines = [ ]
for item in items [ : 30 ] :
lines . append ( f " <code> { item [ ' id ' ] } </code> [ { item [ ' media_type ' ] } ] — { item [ ' keywords ' ] } " )
text = " \n " . join ( lines )
if len ( items ) > 30 :
text + = f " \n …and { len ( items ) - 30 } more. "
await msg . answer ( text , parse_mode = " HTML " )
# ── /delete ──────────────────────────────────────────────────────────────────
@@ -392,6 +402,7 @@ async def inline_handler(query: InlineQuery):
async def main ( ) :
await db . init_db ( )
await register_commands ( )
log . info ( " Bot starting… " )
await dp . start_polling ( bot )
snail