diff --git a/src/tgbot/app.py b/src/tgbot/app.py index 8d37ff2..dd8ce26 100644 --- a/src/tgbot/app.py +++ b/src/tgbot/app.py @@ -42,6 +42,7 @@ handle_broadcast_text_ru_trigger, ) from src.tgbot.handlers.admin.forward_channel import handle_forwarded_from_tgchannelru +from src.tgbot.handlers.admin.promote_moderator import handle_promote_moderator from src.tgbot.handlers.admin.user_info import ( DELETE_USER_DATA_CONFIRMATION_CALLBACK, delete_user_data, @@ -252,6 +253,14 @@ def add_handlers(application: Application) -> None: ) ) + application.add_handler( + CommandHandler( + "promotemod", + handle_promote_moderator, + filters=filters.ChatType.PRIVATE & filters.UpdateType.MESSAGE, + ) + ) + ###################### # broadcast texts application.add_handlers( diff --git a/src/tgbot/handlers/admin/promote_moderator.py b/src/tgbot/handlers/admin/promote_moderator.py new file mode 100644 index 0000000..bc5d14c --- /dev/null +++ b/src/tgbot/handlers/admin/promote_moderator.py @@ -0,0 +1,104 @@ +from collections.abc import Mapping + +import logging + +from telegram import Update +from telegram.error import TelegramError +from telegram.ext import ContextTypes + +from src.tgbot.constants import TELEGRAM_MODERATOR_CHAT_ID, UserType +from src.tgbot.handlers.admin.service import get_user_by_tg_username +from src.tgbot.service import ( + add_user_tg_chat_membership, + get_user_by_id, + update_user, +) +from src.tgbot.user_info import update_user_info_cache + + +async def _is_admin(user_info: Mapping[str, object] | None) -> bool: + if not user_info: + return False + + raw_type = user_info.get("type") + try: + return UserType(str(raw_type)) == UserType.ADMIN + except ValueError: + logging.warning("Unknown user type '%s' encountered during admin check", raw_type) + return False + + +async def handle_promote_moderator( + update: Update, context: ContextTypes.DEFAULT_TYPE +) -> None: + message = update.effective_message + if message is None or message.text is None: + return + + requester = await get_user_by_id(message.from_user.id) + if not await _is_admin(requester): + await message.reply_text("🚫 Only admins can promote moderators.") + return + + parts = message.text.strip().split(maxsplit=1) + if len(parts) < 2: + await message.reply_text("Usage: /promotemod ") + return + + identifier = parts[1].strip() + target = None + + if identifier.startswith("@"): # username lookup + username = identifier[1:] + if not username: + await message.reply_text("🚫 Please provide a valid username after @.") + return + target = await get_user_by_tg_username(username) + else: + try: + telegram_id = int(identifier) + except ValueError: + await message.reply_text( + "🚫 Identifier must be a Telegram ID or @username." + ) + return + target = await get_user_by_id(telegram_id) + + if target is None: + await message.reply_text("🚫 Could not find the specified user.") + return + + target_id = int(target["id"]) + + try: + invite_link = await context.bot.create_chat_invite_link( + chat_id=TELEGRAM_MODERATOR_CHAT_ID, + creates_join_request=False, + member_limit=1, + ) + except TelegramError: + logging.exception("Failed to generate moderator invite link for user_id=%s", target_id) + await message.reply_text("❌ Failed to generate invite link. Try again later.") + return + + await update_user(target_id, type=UserType.MODERATOR.value) + await add_user_tg_chat_membership(target_id, TELEGRAM_MODERATOR_CHAT_ID) + await update_user_info_cache(target_id) + + try: + await context.bot.send_message( + chat_id=target_id, + text=( + "🎉 You have been promoted to moderator! " + f"Here is your invite link: {invite_link.invite_link}" + ), + disable_web_page_preview=True, + ) + except TelegramError: + logging.exception("Failed to send moderator invite to user_id=%s", target_id) + await message.reply_text( + "⚠️ Promotion updated, but sending the invite link failed." + ) + return + + await message.reply_text("✅ User promoted to moderator and invite link sent.")