mautrix-telegram/mautrix_telegram/commands/telegram/misc.py

# mautrix-telegram - A Matrix-Telegram puppeting bridge
# Copyright (C) 2021 Tulir Asokan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations

from typing import cast
import base64
import codecs
import re
import shlex

from aiohttp import ClientSession, InvalidURL
from telethon.errors import (
    ChatIdInvalidError,
    EmoticonInvalidError,
    InviteHashExpiredError,
    InviteHashInvalidError,
    InviteRequestSentError,
    OptionsTooMuchError,
    UserAlreadyParticipantError,
)
from telethon.tl.functions.channels import JoinChannelRequest
from telethon.tl.functions.contacts import DeleteByPhonesRequest, ImportContactsRequest
from telethon.tl.functions.messages import (
    CheckChatInviteRequest,
    GetBotCallbackAnswerRequest,
    ImportChatInviteRequest,
    SendVoteRequest,
)
from telethon.tl.patched import Message
from telethon.tl.types import (
    InputMediaDice,
    InputPhoneContact,
    MessageMediaGame,
    MessageMediaPoll,
    TypeInputPeer,
    TypeUpdates,
    User as TLUser,
)
from telethon.tl.types.contacts import ImportedContacts
from telethon.tl.types.messages import BotCallbackAnswer

from mautrix.types import EventID, Format

from ... import portal as po, puppet as pu
from ...abstract_user import AbstractUser
from ...commands import (
    SECTION_CREATING_PORTALS,
    SECTION_MISC,
    SECTION_PORTAL_MANAGEMENT,
    CommandEvent,
    command_handler,
)
from ...db import Message as DBMessage
from ...types import TelegramID


@command_handler(
    needs_auth=False,
    needs_puppeting=False,
    help_section=SECTION_MISC,
    help_args="<_caption_>",
    help_text="Set a caption for the next image you send",
)
async def caption(evt: CommandEvent) -> EventID:
    if len(evt.args) == 0:
        return await evt.reply("**Usage:** `$cmdprefix+sp caption <caption>`")

    prefix = f"{evt.command_prefix} caption "
    if evt.content.format == Format.HTML:
        evt.content.formatted_body = evt.content.formatted_body.replace(prefix, "", 1)
    evt.content.body = evt.content.body.replace(prefix, "", 1)
    evt.sender.command_status = {"caption": evt.content, "action": "Caption"}
    return await evt.reply(
        "Your next image or file will be sent with that caption. "
        "Use `$cmdprefix+sp cancel` to cancel the caption."
    )


@command_handler(
    help_section=SECTION_MISC,
    help_args="[_-r|--remote_] <_query_>",
    help_text="Search your contacts or the Telegram servers for users.",
)
async def search(evt: CommandEvent) -> EventID:
    if len(evt.args) == 0:
        return await evt.reply("**Usage:** `$cmdprefix+sp search [-r|--remote] <query>`")

    force_remote = False
    if evt.args[0] in {"-r", "--remote"}:
        force_remote = True
        evt.args.pop(0)

    query = " ".join(evt.args)
    if force_remote and len(query) < 5:
        return await evt.reply("Minimum length of query for remote search is 5 characters.")

    results, remote = await evt.sender.search(query, force_remote)

    if not results:
        if len(query) < 5 and remote:
            return await evt.reply(
                "No local results. Minimum length of remote query is 5 characters."
            )
        return await evt.reply("No results 3:")

    reply: list[str] = []
    if remote:
        reply += ["**Results from Telegram server:**", ""]
    else:
        reply += ["**Results in contacts:**", ""]
    reply += [
        (
            f"* [{puppet.displayname}](https://matrix.to/#/{puppet.mxid}): "
            f"{puppet.id} ({similarity}% match)"
        )
        for puppet, similarity in results
    ]

    # TODO somehow show remote channel results when joining by alias is possible?

    return await evt.reply("\n".join(reply))


@command_handler(
    help_section=SECTION_CREATING_PORTALS,
    help_args="<_username_>",
    help_text=(
        "Open a private chat with the given Telegram user. You can also use a "
        "phone number instead of username, but you must have the number in "
        "your Telegram contacts for that to work."
    ),
)
async def pm(evt: CommandEvent) -> EventID:
    if len(evt.args) == 0:
        return await evt.reply("**Usage:** `$cmdprefix+sp pm <username>`")

    try:
        id = "".join(evt.args).translate({ord(c): None for c in "+()- "})
        user = await evt.sender.client.get_entity(id)
    except ValueError:
        return await evt.reply("Invalid user identifier or user not found.")

    if not user:
        return await evt.reply("User not found.")
    elif not isinstance(user, TLUser):
        return await evt.reply("That doesn't seem to be a user.")
    portal = await po.Portal.get_by_entity(user, tg_receiver=evt.sender.tgid)
    await portal.create_matrix_room(evt.sender, user, [evt.sender.mxid])
    displayname, _ = pu.Puppet.get_displayname(user, False)
    return await evt.reply(f"Created private chat room with {displayname}")


async def _handle_contact(source: AbstractUser, user: TLUser) -> str:
    puppet: pu.Puppet = await pu.Puppet.get_by_tgid(user.id)
    await puppet.update_info(source, user)

    params = []
    if user.username:
        params.append(f"[@{user.username}](https://t.me/{user.username})")
    if user.phone:
        params.append(f"+{user.phone}")
    params.append(f"ID `{user.id}`")
    params_str = " / ".join(params)
    return f"[{puppet.displayname}](https://matrix.to/#/{puppet.mxid}): {params_str}"


@command_handler(
    help_section=SECTION_CREATING_PORTALS,
    help_args="<_phone_> <_first name_> <_last name_>",
    help_text="Add a phone number to your contacts on Telegram",
)
async def add_contact(evt: CommandEvent) -> EventID:
    if len(evt.args) < 3:
        return await evt.reply(
            "**Usage:** `$cmdprefix+sp add-contact <phone> <first name> <last name>`"
        )
    try:
        names = shlex.split(" ".join(evt.args[1:]))
    except ValueError as e:
        return await evt.reply(
            f"Failed to parse names (use shell quoting for names with spaces): {e}"
        )
    if len(names) != 2:
        return await evt.reply(
            "Wrong number of names, must have first and last name "
            "(use shell quoting for names with spaces)"
        )
    res: ImportedContacts = await evt.sender.client(
        ImportContactsRequest(
            contacts=[
                InputPhoneContact(
                    client_id=1, phone=evt.args[0], first_name=names[0], last_name=names[1]
                )
            ]
        )
    )
    if res.retry_contacts:
        return await evt.reply("Failed to import contacts")
    elif not res.users:
        return await evt.reply("Contact imported, but user not found on Telegram")
    imported_str = "\n".join(
        [f"* {await _handle_contact(evt.sender, user)}" for user in res.users]
    )
    return await evt.reply(f"Imported contacts:\n\n{imported_str}")


@command_handler(
    help_section=SECTION_CREATING_PORTALS,
    help_args="<_phones..._>",
    help_text="Remove phone numbers from your contacts on Telegram.",
    aliases=["remove-contact", "delete-contacts", "remove-contacts"],
)
async def delete_contact(evt: CommandEvent) -> EventID:
    if len(evt.args) == 0:
        return await evt.reply("**Usage:** `$cmdprefix+sp delete-contact <phones...>`")
    ok = await evt.sender.client(DeleteByPhonesRequest(phones=evt.args))
    if ok:
        return await evt.reply("Contacts deleted")
    else:
        return await evt.reply("Contacts not deleted?")


async def _join(
    evt: CommandEvent, identifier: str, link_type: str
) -> tuple[TypeUpdates | None, EventID | None]:
    if link_type == "joinchat":
        try:
            await evt.sender.client(CheckChatInviteRequest(identifier))
        except InviteHashInvalidError:
            return None, await evt.reply("Invalid invite link.")
        except InviteHashExpiredError:
            return None, await evt.reply("Invite link expired.")
        try:
            return (await evt.sender.client(ImportChatInviteRequest(identifier))), None
        except UserAlreadyParticipantError:
            return None, await evt.reply("You are already in that chat.")
        except InviteRequestSentError:
            return None, await evt.reply("Invite request sent successfully.")
    else:
        channel = await evt.sender.client.get_entity(identifier)
        if not channel:
            return None, await evt.reply("Channel/supergroup not found.")
        return await evt.sender.client(JoinChannelRequest(channel)), None


@command_handler(
    help_section=SECTION_CREATING_PORTALS,
    help_args="<_link_>",
    help_text="Join a chat with an invite link.",
)
async def join(evt: CommandEvent) -> EventID | None:
    if len(evt.args) == 0:
        return await evt.reply("**Usage:** `$cmdprefix+sp join <invite link>`")

    url = evt.args[0]
    if evt.config["bridge.invite_link_resolve"]:
        try:
            async with ClientSession() as sess, sess.get(url) as resp:
                url = str(resp.url)
        except InvalidURL:
            return await evt.reply("That doesn't look like a Telegram invite link.")

    regex = re.compile(
        r"(?:https?://)?t(?:elegram)?\.(?:dog|me)(?:/(?P<type>joinchat|s))?/(?P<id>[^/]+)/?",
        flags=re.IGNORECASE,
    )
    arg = regex.match(url)
    if not arg:
        return await evt.reply("That doesn't look like a Telegram invite link.")

    data = arg.groupdict()
    identifier = data["id"]
    link_type = data["type"]
    if link_type:
        link_type = link_type.lower()
    elif identifier.startswith("+"):
        link_type = "joinchat"
        identifier = identifier[1:]
    updates, _ = await _join(evt, identifier, link_type)
    if not updates:
        return None

    for chat in updates.chats:
        portal = await po.Portal.get_by_entity(chat)
        if portal.mxid:
            await portal.invite_to_matrix([evt.sender.mxid])
            return await evt.reply(f"Invited you to portal of {portal.title}")
        else:
            await evt.reply(f"Creating room for {chat.title}... This might take a while.")
            try:
                await portal.create_matrix_room(evt.sender, chat, [evt.sender.mxid])
            except ChatIdInvalidError as e:
                evt.log.trace(
                    "ChatIdInvalidError while creating portal from !tg join command: %s",
                    updates.stringify(),
                )
                raise e
            if portal.mxid:
                return await evt.reply(f"Created room for {portal.title}")
            else:
                return await evt.reply(f"Couldn't create room for {portal.title}")
    return None


@command_handler(
    help_section=SECTION_MISC,
    help_args="[`chats`|`contacts`|`me`]",
    help_text="Synchronize your chat portals, contacts and/or own info.",
)
async def sync(evt: CommandEvent) -> EventID:
    if len(evt.args) > 0:
        sync_only = evt.args[0]
        if sync_only not in ("chats", "contacts", "me"):
            return await evt.reply("**Usage:** `$cmdprefix+sp sync [chats|contacts|me]`")
    else:
        sync_only = None

    if not sync_only or sync_only == "chats":
        await evt.reply("Synchronizing chats...")
        await evt.sender.sync_dialogs()
    if not sync_only or sync_only == "contacts":
        await evt.reply("Synchronizing contacts...")
        await evt.sender.sync_contacts()
    if not sync_only or sync_only == "me":
        await evt.sender.update_info()
    return await evt.reply("Synchronization complete.")


PEER_TYPE_CHAT = b"g"


class MessageIDError(ValueError):
    def __init__(self, message: str) -> None:
        super().__init__(message)
        self.message = message


async def _parse_encoded_msgid(
    user: AbstractUser, enc_id: str, type_name: str
) -> tuple[TypeInputPeer, Message]:
    try:
        enc_id += (4 - len(enc_id) % 4) * "="
        enc_id = base64.b64decode(enc_id)
        peer_type, enc_id = bytes([enc_id[0]]), enc_id[1:]
        tgid = TelegramID(int(codecs.encode(enc_id[0:5], "hex_codec"), 16))
        msg_id = TelegramID(int(codecs.encode(enc_id[5:10], "hex_codec"), 16))
        space = None
        if peer_type == PEER_TYPE_CHAT:
            space = TelegramID(int(codecs.encode(enc_id[10:15], "hex_codec"), 16))
    except ValueError as e:
        raise MessageIDError(f"Invalid {type_name} ID (format)") from e

    if peer_type == PEER_TYPE_CHAT:
        orig_msg = await DBMessage.get_one_by_tgid(msg_id, space)
        if not orig_msg:
            raise MessageIDError(f"Invalid {type_name} ID (original message not found in db)")
        new_msg = await DBMessage.get_by_mxid(orig_msg.mxid, orig_msg.mx_room, user.tgid)
        if not new_msg:
            raise MessageIDError(f"Invalid {type_name} ID (your copy of message not found in db)")
        msg_id = new_msg.tgid
    try:
        peer = await user.client.get_input_entity(tgid)
    except ValueError as e:
        raise MessageIDError(f"Invalid {type_name} ID (chat not found)") from e

    msg = await user.client.get_messages(entity=peer, ids=msg_id)
    if not msg:
        raise MessageIDError(f"Invalid {type_name} ID (message not found)")
    return peer, cast(Message, msg)


@command_handler(
    help_section=SECTION_MISC, help_args="<_play ID_>", help_text="Play a Telegram game."
)
async def play(evt: CommandEvent) -> EventID:
    if len(evt.args) < 1:
        return await evt.reply("**Usage:** `$cmdprefix+sp play <play ID>`")
    elif not await evt.sender.is_logged_in():
        return await evt.reply("You must be logged in with a real account to play games.")
    elif evt.sender.is_bot:
        return await evt.reply("Bots can't play games :(")

    try:
        peer, msg = await _parse_encoded_msgid(evt.sender, evt.args[0], type_name="play")
    except MessageIDError as e:
        return await evt.reply(e.message)

    if not isinstance(msg.media, MessageMediaGame):
        return await evt.reply("Invalid play ID (message doesn't look like a game)")

    game = await evt.sender.client(
        GetBotCallbackAnswerRequest(peer=peer, msg_id=msg.id, game=True)
    )
    if not isinstance(game, BotCallbackAnswer):
        return await evt.reply("Game request response invalid")

    return await evt.reply(
        f"Click [here]({game.url}) to play {msg.media.game.title}:\n\n"
        f"{msg.media.game.description}"
    )


@command_handler(
    help_section=SECTION_MISC,
    help_args="<_poll ID_> <_choice number_>",
    help_text="Vote in a Telegram poll.",
)
async def vote(evt: CommandEvent) -> EventID | None:
    if len(evt.args) < 1:
        return await evt.reply("**Usage:** `$cmdprefix+sp vote <poll ID> <choice number>`")
    elif not await evt.sender.is_logged_in():
        return await evt.reply("You must be logged in with a real account to vote in polls.")
    elif evt.sender.is_bot:
        return await evt.reply("Bots can't vote in polls :(")

    try:
        peer, msg = await _parse_encoded_msgid(evt.sender, evt.args[0], type_name="poll")
    except MessageIDError as e:
        return await evt.reply(e.message)

    if not isinstance(msg.media, MessageMediaPoll):
        return await evt.reply("Invalid poll ID (message doesn't look like a poll)")

    options = []
    for option in evt.args[1:]:
        try:
            if len(option) > 10:
                raise ValueError("option index too long")
            option_index = int(option) - 1
        except ValueError:
            option_index = None
        if option_index is None:
            return await evt.reply(
                f'Invalid option number "{option}"', render_markdown=False, allow_html=False
            )
        elif option_index < 0:
            return await evt.reply(
                f"Invalid option number {option}. Option numbers must be positive."
            )
        elif option_index >= len(msg.media.poll.answers):
            return await evt.reply(
                f"Invalid option number {option}. "
                f"The poll only has {len(msg.media.poll.answers)} options."
            )
        options.append(msg.media.poll.answers[option_index].option)
    options = [msg.media.poll.answers[int(option) - 1].option for option in evt.args[1:]]
    try:
        await evt.sender.client(SendVoteRequest(peer=peer, msg_id=msg.id, options=options))
    except OptionsTooMuchError:
        return await evt.reply("You passed too many options.")
    # TODO use response
    return await evt.mark_read()


@command_handler(
    help_section=SECTION_MISC,
    help_args="<_emoji_>",
    help_text="Roll a dice (\U0001F3B2), kick a football (\u26BD\uFE0F) or throw a "
    "dart (\U0001F3AF) or basketball (\U0001F3C0) on the Telegram servers.",
)
async def random(evt: CommandEvent) -> EventID:
    if not evt.is_portal:
        return await evt.reply("You can only randomize values in portal rooms")
    portal = await po.Portal.get_by_mxid(evt.room_id)
    arg = evt.args[0] if len(evt.args) > 0 else "dice"
    emoticon = {
        "dart": "\U0001F3AF",
        "dice": "\U0001F3B2",
        "ball": "\U0001F3C0",
        "basketball": "\U0001F3C0",
        "football": "\u26BD",
        "soccer": "\u26BD",
    }.get(arg, arg)
    try:
        await evt.sender.client.send_media(
            await portal.get_input_entity(evt.sender), InputMediaDice(emoticon)
        )
    except EmoticonInvalidError:
        return await evt.reply("Invalid emoji for randomization")


@command_handler(
    help_section=SECTION_PORTAL_MANAGEMENT,
    help_args="[_limit_]",
    help_text="Backfill messages from Telegram history.",
)
async def backfill(evt: CommandEvent) -> None:
    if not evt.is_portal:
        await evt.reply("You can only use backfill in portal rooms")
        return
    elif not evt.config["bridge.backfill.enable"]:
        await evt.reply("Backfilling is disabled in the bridge config")
        return
    try:
        limit = int(evt.args[0])
    except (ValueError, IndexError):
        limit = -1
    portal = await po.Portal.get_by_mxid(evt.room_id)
    if not evt.config["bridge.backfill.normal_groups"] and portal.peer_type == "chat":
        await evt.reply("Backfilling normal groups is disabled in the bridge config")
        return
    output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
    await evt.reply(output)