danbooru-bot/inline_query.py

from uuid import uuid4

from telegram import (
    Update,
    InlineKeyboardButton,
    InlineQueryResultArticle,
    InputTextMessageContent,
    InlineKeyboardMarkup,
    LinkPreviewOptions,
)
from telegram.constants import ParseMode
from telegram.ext import ContextTypes

import html_parser
from config import *
from extensions import humanize_tags_from_json, format_rating, get_json


async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.inline_query.query

    if not query:
        return

    if not query.isdigit():
        return

    data = get_json(f"posts/{query}")
    if data is None:
        await invalid_query(update, query)
        return

    await answer_query(update, query, data)


async def answer_query(update: Update, query: str, data) -> None:
    characters = humanize_tags_from_json(data["tag_string_character"], "no characters")
    copyrights = humanize_tags_from_json(
        data["tag_string_copyright"], "unknown copyright"
    )
    artists = humanize_tags_from_json(data["tag_string_artist"], "unknown artist")
    rating = format_rating(data["rating"])

    if data["is_banned"]:
        results = [
            InlineQueryResultArticle(
                id=str(uuid4()),
                title=f"#{query} {rating}",
                description=f"{characters} ({copyrights}) drawn by {artists}",
                input_message_content=InputTextMessageContent(
                    " ".join(
                        [
                            html_parser.code(f"#{query}"),
                            html_parser.strikethrough(
                                " ".join(
                                    [
                                        html_parser.bold(
                                            f"{characters} ({copyrights})"
                                        ),
                                        "drawn by",
                                        html_parser.bold(artists),
                                    ]
                                )
                            ),
                            rating,
                            "\nThis post has been removed because of a takedown request or rule violation.",
                        ]
                    ),
                    parse_mode=ParseMode.HTML,
                ),
            )
        ]

        await update.inline_query.answer(results)
        return

    keyboard = [
        [
            InlineKeyboardButton(
                f"Open in {app.name}",
                url=f"{app.protocol}://{app.hostname}/posts/{query}",
            ),
            InlineKeyboardButton("View original", url=data["file_url"]),
        ]
    ]

    results = [
        InlineQueryResultArticle(
            id=str(uuid4()),
            title=f"#{query} {rating}",
            description=f"{characters} ({copyrights}) drawn by {artists}",
            thumbnail_url=data["preview_file_url"],
            input_message_content=InputTextMessageContent(
                " ".join(
                    [
                        html_parser.code(f"#{query}"),
                        html_parser.bold(f"{characters} ({copyrights})"),
                        "drawn by",
                        html_parser.bold(artists),
                        rating,
                    ]
                ),
                parse_mode=ParseMode.HTML,
                link_preview_options=LinkPreviewOptions(url=data["large_file_url"]),
            ),
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
    ]

    await update.inline_query.answer(results)


async def invalid_query(update: Update, query: str) -> None:
    results = [
        InlineQueryResultArticle(
            id=str(uuid4()),
            title=f"#{query}",
            description="Error",
            input_message_content=InputTextMessageContent(
                " ".join([html_parser.code(f"#{query}"), "That record was not found."]),
                parse_mode=ParseMode.HTML,
            ),
        )
    ]

    await update.inline_query.answer(results)