Simplify moderator invite milestone checks

src/recommendations/meme_queue.py

@@ -1,3 +1,5 @@
+import logging
+from math import ceil
 from typing import Any, Optional
 
 from src import redis
@@ -9,6 +11,11 @@
     get_selected_sources,
 )
 from src.storage.schemas import MemeData
+from sqlalchemy import text
+
+from src.database import fetch_all
+from src.recommendations.utils import exclude_meme_ids_sql_filter
+from src.tgbot.constants import UserType
 from src.tgbot.user_info import get_user_info
 
 
@@ -84,8 +91,8 @@ async def generate_recommendations(
     Will be refactored
     """
 
+    user_info = await get_user_info(user_id)
     if nmemes_sent is None:
-        user_info = await get_user_info(user_id)
         nmemes_sent = user_info["nmemes_sent"]
 
     queue_key = redis.get_meme_queue_key(user_id)
@@ -97,6 +104,38 @@ async def generate_recommendations(
     if retriever is None:
         retriever = CandidatesRetriever()
 
+    async def get_low_sent_candidates(
+        user_id: int, limit: int, exclude_ids: list[int]
+    ) -> list[dict[str, Any]]:
+        if limit <= 0:
+            return []
+
+        query = f"""
+            SELECT
+                M.id,
+                M.type,
+                M.telegram_file_id,
+                M.caption,
+                'low_sent_pool' AS recommended_by
+            FROM meme M
+            LEFT JOIN meme_stats MS
+                ON MS.meme_id = M.id
+            LEFT JOIN user_meme_reaction R
+                ON R.user_id = {user_id}
+                AND R.meme_id = M.id
+            INNER JOIN user_language UL
+                ON UL.user_id = {user_id}
+                AND UL.language_code = M.language_code
+            WHERE 1=1
+                AND M.status = 'ok'
+                AND R.meme_id IS NULL
+                {exclude_meme_ids_sql_filter(exclude_ids)}
+            ORDER BY COALESCE(MS.nmemes_sent, 0), M.id
+            LIMIT {limit}
+        """
+
+        return await fetch_all(text(query))
+
     async def get_candidates(user_id, limit):
         """A helper function to avoid copy-paste"""
 
@@ -158,7 +197,37 @@ async def get_candidates(user_id, limit):
 
         return candidates
 
-    candidates = await get_candidates(user_id, limit)
+    user_type_value = user_info.get("type")
+    user_type = None
+    if user_type_value:
+        try:
+            user_type = UserType(str(user_type_value))
+        except ValueError:
+            logging.warning(
+                "Unknown user type '%s' for user_id=%s during queue generation",
+                user_type_value,
+                user_id,
+            )
+
+    candidates: list[dict[str, Any]] = []
+
+    if user_type in (UserType.MODERATOR, UserType.ADMIN):
+        low_sent_quota = ceil(limit * 0.75)
+        low_sent_candidates = await get_low_sent_candidates(
+            user_id,
+            low_sent_quota,
+            meme_ids_in_queue,
+        )
+        candidates.extend(low_sent_candidates)
+        meme_ids_in_queue.extend(candidate["id"] for candidate in low_sent_candidates)
+
+        remaining_limit = max(0, limit - len(candidates))
+        if remaining_limit > 0:
+            extra_candidates = await get_candidates(user_id, remaining_limit)
+            candidates.extend(extra_candidates)
+    else:
+        candidates = await get_candidates(user_id, limit)
+
     if len(candidates) > 0:
         await redis.add_memes_to_queue_by_key(queue_key, candidates)
 

src/redis.py

@@ -100,3 +100,4 @@ async def get_user_wrapped(user_id: str) -> dict | None:
 
 async def set_user_wrapped(user_id: str, data: dict) -> None:
     await redis_client.set(get_user_wrapped_key(user_id), orjson.dumps(data))
+

src/tgbot/app.py

@@ -62,6 +62,10 @@
     send_tokens_to_reply,
 )
 from src.tgbot.handlers.moderator import get_meme, meme_source
+from src.tgbot.handlers.moderator.invite import (
+    MODERATOR_INVITE_CALLBACK_DATA,
+    handle_moderator_invite_callback,
+)
 from src.tgbot.handlers.payments.purchase import (
     PURCHASE_TOKEN_CALLBACK_DATA_REGEXP,
     handle_new_token_purchase_request_callback,
@@ -224,6 +228,13 @@ def add_handlers(application: Application) -> None:
         )
     )
 
+    application.add_handler(
+        CallbackQueryHandler(
+            handle_moderator_invite_callback,
+            pattern=fr"^{MODERATOR_INVITE_CALLBACK_DATA}$",
+        )
+    )
+
     ############## popup reaction
     application.add_handler(
         CallbackQueryHandler(

src/tgbot/handlers/moderator/invite.py

@@ -0,0 +1,124 @@
+"""Utilities and handlers for promoting power users to moderators."""
+
+import logging
+from collections.abc import Mapping
+
+from telegram import (
+    Bot,
+    InlineKeyboardButton,
+    InlineKeyboardMarkup,
+    Update,
+)
+from telegram.error import TelegramError
+from telegram.ext import ContextTypes
+
+from src.tgbot.constants import TELEGRAM_MODERATOR_CHAT_ID, UserType
+from src.tgbot.senders.next_message import next_message
+from src.tgbot.service import (
+    add_user_tg_chat_membership,
+    get_user_languages,
+    update_user,
+)
+from src.tgbot.user_info import update_user_info_cache
+
+
+INVITE_MESSAGE_TEMPLATE = (
+    "🥳 Ты посмотрел {nmemes_sent} мемов! Нам очень нужна помощь модераторов "
+    "в русской ленте Fast Food & Memes. Нажми кнопку ниже, чтобы получить "
+    "одноразовую ссылку и присоединиться к нашему модераторскому чату."
+)
+INVITE_BUTTON_TEXT = "Хочу модерировать 🇷🇺"
+MODERATOR_INVITE_CALLBACK_DATA = "moderator_invite:join"
+
+
+async def maybe_send_moderator_invite(
+    bot: Bot, user_id: int, user_info: Mapping[str, object]
+) -> None:
+    """Send a moderator invite when the user hits a milestone."""
+
+    nmemes_sent = int(user_info.get("nmemes_sent") or 0)
+    if nmemes_sent == 0 or nmemes_sent % 500 != 0:
+        return
+
+    raw_type = user_info.get("type")
+    user_type = None
+    if raw_type:
+        try:
+            user_type = UserType(str(raw_type))
+        except ValueError:
+            logging.warning(
+                "Unknown user type '%s' for user %s when evaluating moderator invite",
+                raw_type,
+                user_id,
+            )
+
+    if user_type and user_type in (UserType.MODERATOR, UserType.ADMIN):
+        return
+
+    user_languages = await get_user_languages(user_id)
+    if "ru" not in user_languages:
+        return
+
+    keyboard = InlineKeyboardMarkup(
+        [
+            [
+                InlineKeyboardButton(
+                    text=INVITE_BUTTON_TEXT,
+                    callback_data=MODERATOR_INVITE_CALLBACK_DATA,
+                )
+            ]
+        ]
+    )
+
+    await bot.send_message(
+        chat_id=user_id,
+        text=INVITE_MESSAGE_TEMPLATE.format(nmemes_sent=nmemes_sent),
+        reply_markup=keyboard,
+        disable_web_page_preview=True,
+    )
+    logging.info(
+        "Sent moderator invitation to user_id=%s for milestone=%s",
+        user_id,
+        nmemes_sent,
+    )
+
+
+async def handle_moderator_invite_callback(
+    update: Update, context: ContextTypes.DEFAULT_TYPE
+) -> None:
+    query = update.callback_query
+    if query is None or query.data != MODERATOR_INVITE_CALLBACK_DATA:
+        return
+
+    user_id = query.from_user.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", user_id)
+        await query.answer(text="Не получилось выдать ссылку, попробуй позже", show_alert=True)
+        return
+
+    await query.edit_message_reply_markup(reply_markup=None)
+
+    await context.bot.send_message(
+        chat_id=user_id,
+        text=(
+            "Добро пожаловать в модераторскую команду! "
+            f"Вот твоя одноразовая ссылка: {invite_link.invite_link}"
+        ),
+        disable_web_page_preview=True,
+    )
+
+    await add_user_tg_chat_membership(user_id, TELEGRAM_MODERATOR_CHAT_ID)
+    await update_user(user_id, type=UserType.MODERATOR.value)
+    await update_user_info_cache(user_id)
+
+    logging.info("Promoted user_id=%s to moderator", user_id)
+
+    await query.answer()
+    await next_message(context.bot, user_id, prev_update=update)

src/tgbot/handlers/reaction.py

@@ -16,6 +16,7 @@
     update_user_meme_reaction,
 )
 from src.tgbot.senders.next_message import next_message
+from src.tgbot.handlers.moderator.invite import maybe_send_moderator_invite
 from src.tgbot.user_info import update_user_info_counters
 
 
@@ -27,7 +28,8 @@ async def handle_reaction(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
     )
 
     # do that in sync since we'll use counters in next_message
-    await update_user_info_counters(user_id)
+    user_info = await update_user_info_counters(user_id)
+    await maybe_send_moderator_invite(context.bot, user_id, user_info)
     asyncio.create_task(update_user_last_active_at(user_id))
     asyncio.create_task(reward_user_for_daily_activity(user_id))
 

src/tgbot/user_info.py

@@ -40,8 +40,9 @@ async def get_user_info(user_id: int) -> defaultdict:
     return defaultdict(lambda: None, **user_info)
 
 
-async def update_user_info_counters(user_id: int):
+async def update_user_info_counters(user_id: int) -> defaultdict:
     user_info = await get_user_info(user_id)
     user_info["nmemes_sent"] += 1
     user_info["memes_watched_today"] += 1
     await cache_user_info(user_id, user_info)
+    return user_info