Merge branch 'production' into codex/handle-404-errors-in-instagram-parser-s6wmrc

Author: ohld

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/storage/parsers/ig.py

@@ -37,20 +37,31 @@ async def _fetch_hikerapi(  # pragma: no cover - thin wrapper around httpx
         return response.json()
 
 
-async def _get_user_info(username: str) -> dict | None:
-    return await _fetch_hikerapi(
-        "/user/by/username",
-        params={"username": username},
-        not_found_message=f"Instagram user '{username}' not found. Skipping.",
-    )
+async def _get_user_medias(
+    user_id: int,
+) -> dict | None:
+    async with httpx.AsyncClient(timeout=20.0) as client:
+        try:
+            response = await client.get(
+                "https://api.hikerapi.com/v2/user/medias",
+                params={"user_id": user_id},
+                headers={
+                    "accept": "application/json",
+                    "x-access-key": settings.HIKERAPI_TOKEN,
+                },
+            )
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 404:
+                logging.warning(
+                    "Instagram user with id '%s' not found. Skipping.",
+                    user_id,
+                )
+                return None
 
+            raise
 
-async def _get_user_medias(user_id: int) -> dict | None:
-    return await _fetch_hikerapi(
-        "/user/medias",
-        params={"user_id": user_id},
-        not_found_message=f"Instagram user with id '{user_id}' not found. Skipping.",
-    )
+        return response.json()
 
 
 async def get_user_info(instagram_username: str):
@@ -74,21 +85,10 @@ async def get_user_info(instagram_username: str):
 async def get_user_medias(user_id: int) -> list[IgPostParsingResult]:
     user_medias_response = await _get_user_medias(user_id)
     if not user_medias_response:
-        return []
-
-    response_payload = user_medias_response.get("response") or {}
-    if response_payload.get("status") != "ok":
-        logging.warning("Failed to get %s medias: %s", user_id, user_medias_response)
-        return []
-
-    medias = response_payload.get("items") or []
-    if not isinstance(medias, list):
-        logging.warning(
-            "Unexpected medias payload for %s: %s",
-            user_id,
-            user_medias_response,
-        )
-        return []
+        return None
+    if user_medias_response["response"]["status"] != "ok":
+        logging.warning(f"Failed to get {user_id} medias: {user_medias_response}")
+        return None
 
     logging.info("Received %s medias for %s", len(medias), user_id)
 

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,
@@ -62,6 +63,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 +229,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(
@@ -241,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(

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 <telegram_id|@username>")
+        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.")