TelegramBotAdmin/wbb/modules/admin.py

"""
MIT License

Copyright (c) 2024 TheHamkerCat

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""
import asyncio
import re
from contextlib import suppress
from time import time

from pyrogram import filters
from pyrogram.enums import ChatMembersFilter, ChatMemberStatus, ChatType
from pyrogram.errors import FloodWait
from pyrogram.types import (
    CallbackQuery,
    ChatMemberUpdated,
    ChatPermissions,
    ChatPrivileges,
    Message,
)

from wbb import BOT_ID, BOT_PERMISSIONS, SUDOERS, app, log
from wbb.core.decorators.errors import capture_err
from wbb.core.keyboard import ikb
from wbb.services.bot_permissions import has_permission
from wbb.services.interaction_settings import get_interaction_settings
from wbb.services.member_shortcuts import (
    member_action_shortcut_command,
)
from wbb.services.message_cleanup import schedule_group_cleanup
from wbb.utils.dbadmin import record_audit
from wbb.utils.dbfunctions import (
    add_warn,
    get_warn,
    int_to_alpha,
    remove_warns,
    save_filter,
)
from wbb.utils.functions import (
    extract_user,
    extract_user_and_reason,
    time_converter,
)

__MODULE__ = "成员管理"
__HELP__ = """/ban - 封禁成员。
/dban - 删除被回复消息并封禁发送者。
/tban - 临时封禁成员。
/unban - 解除封禁。
/warn - 警告成员。
/dwarn - 删除被回复消息并警告发送者。
/rmwarns - 清除成员的全部警告。
/warns - 查看成员警告次数。
/kick - 踢出成员。
/dkick - 删除被回复消息并踢出发送者。
/purge [数量] - 从被回复消息开始批量删除。
/del - 删除被回复消息。
/promote - 提升成员为管理员。
/fullpromote - 授予成员全部可用管理权限。
/demote - 取消成员的管理员身份。
/pin 或 /unpin - 置顶或取消置顶被回复消息。
/mute - 永久禁言成员。
/tmute - 临时禁言成员。
/unmute - 解除禁言。
/ban_ghosts - 封禁已注销账号。
/report、@admins 或 @admin - 向管理员举报消息。
/invite - 获取群邀请链接。
回复成员消息发送“封禁”“解封”“禁言”“禁言10分钟”“解除禁言”、
“警告”“删除消息”“删除”“删”“踢出”“踢”或“飞”，
可快捷执行对应管理操作。"""

MEMBER_ACTION_CLEANUP_SECONDS = 10


async def member_permissions(chat_id: int, user_id: int):
    perms = []
    member = (await app.get_chat_member(chat_id, user_id)).privileges
    if not member:
        return []
    if member.can_post_messages:
        perms.append("can_post_messages")
    if member.can_edit_messages:
        perms.append("can_edit_messages")
    if member.can_delete_messages:
        perms.append("can_delete_messages")
    if member.can_restrict_members:
        perms.append("can_restrict_members")
    if member.can_promote_members:
        perms.append("can_promote_members")
    if member.can_change_info:
        perms.append("can_change_info")
    if member.can_invite_users:
        perms.append("can_invite_users")
    if member.can_pin_messages:
        perms.append("can_pin_messages")
    if member.can_manage_video_chats:
        perms.append("can_manage_video_chats")
    return perms


from wbb.core.decorators.permissions import adminsOnly

admins_in_chat = {}


async def list_admins(chat_id: int):
    global admins_in_chat
    if chat_id in admins_in_chat:
        interval = time() - admins_in_chat[chat_id]["last_updated_at"]
        if interval < 3600:
            return admins_in_chat[chat_id]["data"]

    admins_in_chat[chat_id] = {
        "last_updated_at": time(),
        "data": [
            member.user.id
            async for member in app.get_chat_members(
                chat_id, filter=ChatMembersFilter.ADMINISTRATORS
            )
        ],
    }
    return admins_in_chat[chat_id]["data"]


# Admin cache reload


@app.on_chat_member_updated()
async def admin_cache_func(_, cmu: ChatMemberUpdated):
    if cmu.old_chat_member and cmu.old_chat_member.promoted_by:
        admins_in_chat[cmu.chat.id] = {
            "last_updated_at": time(),
            "data": [
                member.user.id
                async for member in app.get_chat_members(
                    cmu.chat.id, filter=ChatMembersFilter.ADMINISTRATORS
                )
            ],
        }
        log.info(f"Updated admin cache for {cmu.chat.id} [{cmu.chat.title}]")


# Purge Messages


@app.on_message(filters.command("purge") & ~filters.private)
@adminsOnly("can_delete_messages")
async def purgeFunc(_, message: Message):
    repliedmsg = message.reply_to_message
    await message.delete()

    if not repliedmsg:
        return await message.reply_text("请回复一条消息，机器人将从该消息开始删除。")

    cmd = message.command
    if len(cmd) > 1 and cmd[1].isdigit():
        purge_to = repliedmsg.id + int(cmd[1])
        if purge_to > message.id:
            purge_to = message.id
    else:
        purge_to = message.id

    chat_id = message.chat.id
    message_ids = []

    for message_id in range(
        repliedmsg.id,
        purge_to,
    ):
        message_ids.append(message_id)

        # Max message deletion limit is 100
        if len(message_ids) == 100:
            await app.delete_messages(
                chat_id=chat_id,
                message_ids=message_ids,
                revoke=True,  # For both sides
            )

            # To delete more than 100 messages, start again
            message_ids = []

    # Delete if any messages left
    if len(message_ids) > 0:
        await app.delete_messages(
            chat_id=chat_id,
            message_ids=message_ids,
            revoke=True,
        )


# Kick members


@app.on_message(filters.command(["kick", "dkick"]) & ~filters.private)
@adminsOnly("can_restrict_members")
async def kickFunc(_, message: Message):
    user_id, reason = await extract_user_and_reason(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")
    if user_id == BOT_ID:
        return await message.reply_text(
            "机器人不能将自己踢出群组。"
        )
    if user_id in SUDOERS:
        return await message.reply_text("不能踢出超级管理员。")
    if user_id in (await list_admins(message.chat.id)):
        return await message.reply_text(
            "不能踢出群管理员，请先取消其管理员身份。"
        )
    mention = (await app.get_users(user_id)).mention
    msg = f"""
**已踢出：** {mention}
**操作者：** {message.from_user.mention if message.from_user else '匿名管理员'}
**原因：** {reason or '未填写原因'}"""
    if message.command[0][0] == "d":
        await message.reply_to_message.delete()
    await message.chat.ban_member(user_id)
    replied_message = message.reply_to_message
    if replied_message:
        message = replied_message
    await message.reply_text(msg)
    await asyncio.sleep(1)
    await message.chat.unban_member(user_id)


# Ban members


@app.on_message(filters.command(["ban", "dban", "tban"]) & ~filters.private)
@adminsOnly("can_restrict_members")
async def banFunc(_, message: Message):
    user_id, reason = await extract_user_and_reason(message, sender_chat=True)

    if not user_id:
        return await message.reply_text("未找到该成员。")
    if user_id == BOT_ID:
        return await message.reply_text(
            "机器人不能封禁自己。"
        )
    if user_id in SUDOERS:
        return await message.reply_text(
            "不能封禁超级管理员。"
        )
    if user_id in (await list_admins(message.chat.id)):
        return await message.reply_text(
            "不能封禁群管理员，请先取消其管理员身份。"
        )

    try:
        mention = (await app.get_users(user_id)).mention
    except IndexError:
        mention = (
            message.reply_to_message.sender_chat.title
            if message.reply_to_message
            else "匿名成员"
        )

    msg = (
        f"**已封禁：** {mention}\n"
        f"**操作者：** "
        f"{message.from_user.mention if message.from_user else '匿名管理员'}\n"
    )
    if message.command[0][0] == "d":
        await message.reply_to_message.delete()
    if message.command[0] == "tban":
        split = reason.split(None, 1)
        time_value = split[0]
        temp_reason = split[1] if len(split) > 1 else ""
        temp_ban = await time_converter(message, time_value)
        msg += f"**封禁时长：** {time_value}\n"
        if temp_reason:
            msg += f"**原因：** {temp_reason}"
        with suppress(AttributeError):
            if len(time_value[:-1]) < 3:
                await message.chat.ban_member(user_id, until_date=temp_ban)
                replied_message = message.reply_to_message
                if replied_message:
                    message = replied_message
                await message.reply_text(msg)
            else:
                await message.reply_text("时间数值不能超过两位数。")
        return
    if reason:
        msg += f"**原因：** {reason}"
    await message.chat.ban_member(user_id)
    replied_message = message.reply_to_message
    if replied_message:
        message = replied_message
    await message.reply_text(msg)


# Unban members


@app.on_message(filters.command("unban") & ~filters.private)
@adminsOnly("can_restrict_members")
async def unban_func(_, message: Message):
    # we don't need reasons for unban, also, we
    # don't need to get "text_mention" entity, because
    # normal users won't get text_mention if the user
    # they want to unban is not in the group.
    reply = message.reply_to_message

    if reply and reply.sender_chat and reply.sender_chat != message.chat.id:
        return await message.reply_text("不能对频道身份执行解除封禁。")

    if len(message.command) == 2:
        user = message.text.split(None, 1)[1]
    elif len(message.command) == 1 and reply:
        user = message.reply_to_message.from_user.id
    else:
        return await message.reply_text(
            "请提供用户名，或回复该成员的消息。"
        )
    await message.chat.unban_member(user)
    umention = (await app.get_users(user)).mention
    replied_message = message.reply_to_message
    if replied_message:
        message = replied_message
    await message.reply_text(f"已解除封禁：{umention}")


# Ban users listed in a message


@app.on_message(SUDOERS & filters.command("listban") & ~filters.private)
async def list_ban_(c, message: Message):
    userid, msglink_reason = await extract_user_and_reason(message)
    if not userid or not msglink_reason:
        return await message.reply_text(
            "请提供用户 ID 或用户名、群列表消息链接和封禁原因。"
        )
    if (
        len(msglink_reason.split(" ")) == 1
    ):  # message link included with the reason
        return await message.reply_text(
            "必须填写批量封禁原因。"
        )
    # seperate messge link from reason
    lreason = msglink_reason.split()
    messagelink, reason = lreason[0], " ".join(lreason[1:])

    if not re.search(
        r"(https?://)?t(elegram)?\.me/\w+/\d+", messagelink
    ):  # validate link
        return await message.reply_text("消息链接无效。")

    if userid == BOT_ID:
        return await message.reply_text("机器人不能封禁自己。")
    if userid in SUDOERS:
        return await message.reply_text(
            "不能封禁超级管理员。"
        )
    splitted = messagelink.split("/")
    uname, mid = splitted[-2], int(splitted[-1])
    m = await message.reply_text(
        "`正在从多个群组封禁该成员，请稍候。`"
    )
    try:
        msgtext = (await app.get_messages(uname, mid)).text
        gusernames = re.findall(r"@\\w+", msgtext)
    except:
        return await m.edit_text("无法从消息中读取群组用户名。")
    count = 0
    for username in gusernames:
        try:
            await app.ban_chat_member(username.strip("@"), userid)
            await asyncio.sleep(1)
        except FloodWait as e:
            await asyncio.sleep(e.x)
        except:
            continue
        count += 1
    mention = (await app.get_users(userid)).mention

    msg = f"""
**已批量封禁：** {mention}
**用户 ID：** `{userid}`
**操作者：** {message.from_user.mention}
**影响群数：** `{count}`
**原因：** {reason}
"""
    await m.edit_text(msg)


# Unban users listed in a message


@app.on_message(SUDOERS & filters.command("listunban") & ~filters.private)
async def list_unban_(c, message: Message):
    userid, msglink = await extract_user_and_reason(message)
    if not userid or not msglink:
        return await message.reply_text(
            "请提供用户 ID 或用户名，以及群列表消息链接。"
        )

    if not re.search(
        r"(https?://)?t(elegram)?\.me/\w+/\d+", msglink
    ):  # validate link
        return await message.reply_text("消息链接无效。")

    splitted = msglink.split("/")
    uname, mid = splitted[-2], int(splitted[-1])
    m = await message.reply_text(
        "`正在从多个群组解除封禁，请稍候。`"
    )
    try:
        msgtext = (await app.get_messages(uname, mid)).text
        gusernames = re.findall(r"@\\w+", msgtext)
    except:
        return await m.edit_text("无法从消息中读取群组用户名。")
    count = 0
    for username in gusernames:
        try:
            await app.unban_chat_member(username.strip("@"), userid)
            await asyncio.sleep(1)
        except FloodWait as e:
            await asyncio.sleep(e.x)
        except:
            continue
        count += 1
    mention = (await app.get_users(userid)).mention
    msg = f"""
**已批量解除封禁：** {mention}
**用户 ID：** `{userid}`
**操作者：** {message.from_user.mention}
**影响群数：** `{count}`
"""
    await m.edit_text(msg)


# Delete messages


@app.on_message(filters.command("del") & ~filters.private)
@adminsOnly("can_delete_messages")
async def deleteFunc(_, message: Message):
    if not message.reply_to_message:
        return await message.reply_text("请回复要删除的消息。")
    await message.reply_to_message.delete()
    await message.delete()


# Promote Members


@app.on_message(filters.command(["promote", "fullpromote"]) & ~filters.private)
@adminsOnly("can_promote_members")
async def promoteFunc(_, message: Message):
    user_id = await extract_user(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")

    bot = (await app.get_chat_member(message.chat.id, BOT_ID)).privileges
    if user_id == BOT_ID:
        return await message.reply_text("机器人不能提升自己。")
    if not bot:
        return await message.reply_text("机器人不是本群管理员。")
    if not bot.can_promote_members:
        return await message.reply_text("机器人缺少任命管理员权限。")

    umention = (await app.get_users(user_id)).mention

    if message.command[0][0] == "f":
        await message.chat.promote_member(
            user_id=user_id,
            privileges=ChatPrivileges(
                can_change_info=bot.can_change_info,
                can_invite_users=bot.can_invite_users,
                can_delete_messages=bot.can_delete_messages,
                can_restrict_members=bot.can_restrict_members,
                can_pin_messages=bot.can_pin_messages,
                can_promote_members=bot.can_promote_members,
                can_manage_chat=bot.can_manage_chat,
                can_manage_video_chats=bot.can_manage_video_chats,
            ),
        )
        return await message.reply_text(f"已授予完整管理权限：{umention}")

    await message.chat.promote_member(
        user_id=user_id,
        privileges=ChatPrivileges(
            can_change_info=False,
            can_invite_users=bot.can_invite_users,
            can_delete_messages=bot.can_delete_messages,
            can_restrict_members=False,
            can_pin_messages=False,
            can_promote_members=False,
            can_manage_chat=bot.can_manage_chat,
            can_manage_video_chats=bot.can_manage_video_chats,
        ),
    )
    await message.reply_text(f"已提升为管理员：{umention}")


# Demote Member


@app.on_message(filters.command("demote") & ~filters.private)
@adminsOnly("can_promote_members")
async def demote(_, message: Message):
    user_id = await extract_user(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")
    if user_id == BOT_ID:
        return await message.reply_text("机器人不能取消自己的管理员身份。")
    if user_id in SUDOERS:
        return await message.reply_text(
            "不能取消超级管理员。"
        )
    try:
        member = await app.get_chat_member(message.chat.id, user_id)
        if member.status == ChatMemberStatus.ADMINISTRATOR:
            await message.chat.promote_member(
                user_id=user_id,
                privileges=ChatPrivileges(
                    can_change_info=False,
                    can_invite_users=False,
                    can_delete_messages=False,
                    can_restrict_members=False,
                    can_pin_messages=False,
                    can_promote_members=False,
                    can_manage_chat=False,
                    can_manage_video_chats=False,
                ),
            )
            umention = (await app.get_users(user_id)).mention
            await message.reply_text(f"已取消管理员身份：{umention}")
        else:
            await message.reply_text(
                "该成员不是管理员。"
            )
    except Exception:
        await message.reply_text("取消管理员身份失败，请检查机器人权限。")


# Pin Messages


@app.on_message(filters.command(["pin", "unpin"]) & ~filters.private)
@adminsOnly("can_pin_messages")
async def pin(_, message: Message):
    if not message.reply_to_message:
        return await message.reply_text("请回复要置顶或取消置顶的消息。")
    r = message.reply_to_message
    if message.command[0][0] == "u":
        await r.unpin()
        return await message.reply_text(
            f"**已取消置顶[该消息]({r.link})。**",
            disable_web_page_preview=True,
        )
    await r.pin(disable_notification=True)
    await message.reply(
        f"**已置顶[该消息]({r.link})。**",
        disable_web_page_preview=True,
    )
    msg = "请查看置顶消息：~ " + f"[查看, {r.link}]"
    filter_ = dict(type="text", data=msg)
    await save_filter(message.chat.id, "~pinned", filter_)


# Mute members


@app.on_message(filters.command(["mute", "tmute"]) & ~filters.private)
@adminsOnly("can_restrict_members")
async def mute(_, message: Message):
    user_id, reason = await extract_user_and_reason(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")
    if user_id == BOT_ID:
        return await message.reply_text("机器人不能禁言自己。")
    if user_id in SUDOERS:
        return await message.reply_text(
            "不能禁言超级管理员。"
        )
    if user_id in (await list_admins(message.chat.id)):
        return await message.reply_text(
            "不能禁言群管理员，请先取消其管理员身份。"
        )
    mention = (await app.get_users(user_id)).mention
    keyboard = ikb({"解除禁言": f"unmute_{user_id}"})
    msg = (
        f"**已禁言：** {mention}\n"
        f"**操作者：** "
        f"{message.from_user.mention if message.from_user else '匿名管理员'}\n"
    )
    if message.command[0] == "tmute":
        split = reason.split(None, 1)
        time_value = split[0]
        temp_reason = split[1] if len(split) > 1 else ""
        temp_mute = await time_converter(message, time_value)
        msg += f"**禁言时长：** {time_value}\n"
        if temp_reason:
            msg += f"**原因：** {temp_reason}"
        try:
            if len(time_value[:-1]) < 3:
                await message.chat.restrict_member(
                    user_id,
                    permissions=ChatPermissions(),
                    until_date=temp_mute,
                )
                replied_message = message.reply_to_message
                if replied_message:
                    message = replied_message
                await message.reply_text(msg, reply_markup=keyboard)
            else:
                await message.reply_text("时间数值不能超过两位数。")
        except AttributeError:
            pass
        return
    if reason:
        msg += f"**原因：** {reason}"
    await message.chat.restrict_member(user_id, permissions=ChatPermissions())
    replied_message = message.reply_to_message
    if replied_message:
        message = replied_message
    await message.reply_text(msg, reply_markup=keyboard)


async def _member_action_shortcut_filter(_, __, message: Message) -> bool:
    return member_action_shortcut_command(message.text or "") is not None


member_action_shortcut_filter = filters.create(
    _member_action_shortcut_filter,
    "MemberActionShortcutFilter",
)


@adminsOnly("can_delete_messages")
async def delete_shortcut(_, message: Message):
    await message.reply_text("已接收删除请求。")


@app.on_message(
    filters.group & filters.reply & member_action_shortcut_filter,
    group=-999,
)
async def member_action_shortcut(client, message: Message):
    settings = await get_interaction_settings(message.chat.id)
    if not settings["commands"]["shortcuts"]:
        return None
    if not has_permission("chat.members", BOT_PERMISSIONS):
        return await message.reply_text(
            "当前机器人角色未启用成员管理权限。"
        )
    command = member_action_shortcut_command(message.text or "")
    if not command or not message.reply_to_message:
        return None
    original_text = message.text
    original_command = message.command
    replied_message = message.reply_to_message
    original_message_reply = message.reply_text
    original_target_reply = replied_message.reply_text

    def cleanup_reply(original_reply):
        async def wrapped_reply(text, *args, **kwargs):
            response = await original_reply(text, *args, **kwargs)
            schedule_group_cleanup(
                message,
                replied_message,
                response,
                delay_seconds=settings["cleanup_seconds"],
            )
            return response

        return wrapped_reply

    message.reply_text = cleanup_reply(original_message_reply)
    replied_message.reply_text = cleanup_reply(original_target_reply)
    message.text, message.command = command
    log.info(
        f"成员快捷操作：chat={message.chat.id} "
        f"actor={getattr(message.from_user, 'id', 0)} "
        f"action={message.command[0]}"
    )
    try:
        result = None
        if message.command[0] == "ban":
            result = await banFunc(client, message)
        elif message.command[0] in {"mute", "tmute"}:
            result = await mute(client, message)
        elif message.command[0] == "warn":
            result = await warn_user(client, message)
        elif message.command[0] == "unban":
            result = await unban_func(client, message)
        elif message.command[0] == "unmute":
            result = await unmute(client, message)
        elif message.command[0] == "del":
            result = await delete_shortcut(client, message)
        else:
            result = await kickFunc(client, message)
        with suppress(Exception):
            await record_audit(
                source="telegram", actor_id=getattr(message.from_user, "id", None),
                actor_name=getattr(message.from_user, "first_name", ""),
                action=f"chat.shortcut.{message.command[0]}", chat_id=message.chat.id,
                target_id=getattr(getattr(replied_message, "from_user", None), "id", None),
                summary=original_text,
            )
        return result
    finally:
        message.text = original_text
        message.command = original_command
        message.reply_text = original_message_reply
        replied_message.reply_text = original_target_reply


# Unmute members


@app.on_message(filters.command("unmute") & ~filters.private)
@adminsOnly("can_restrict_members")
async def unmute(_, message: Message):
    user_id = await extract_user(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")
    await message.chat.unban_member(user_id)
    umention = (await app.get_users(user_id)).mention
    replied_message = message.reply_to_message
    if replied_message:
        message = replied_message
    await message.reply_text(f"已解除禁言：{umention}")


# Ban deleted accounts


@app.on_message(filters.command("ban_ghosts") & ~filters.private)
@adminsOnly("can_restrict_members")
async def ban_deleted_accounts(_, message: Message):
    chat_id = message.chat.id
    deleted_users = []
    banned_users = 0
    m = await message.reply("正在查找已注销账号...")

    async for i in app.get_chat_members(chat_id):
        if i.user.is_deleted:
            deleted_users.append(i.user.id)
    if len(deleted_users) > 0:
        for deleted_user in deleted_users:
            try:
                await message.chat.ban_member(deleted_user)
            except Exception:
                pass
            banned_users += 1
        await m.edit(f"已封禁 {banned_users} 个已注销账号。")
    else:
        await m.edit("本群没有已注销账号。")


@app.on_message(filters.command(["warn", "dwarn"]) & ~filters.private)
@adminsOnly("can_restrict_members")
async def warn_user(_, message: Message):
    user_id, reason = await extract_user_and_reason(message)
    chat_id = message.chat.id
    if not user_id:
        return await message.reply_text("未找到该成员。")
    if user_id == BOT_ID:
        return await message.reply_text(
            "机器人不能警告自己。"
        )
    if user_id in SUDOERS:
        return await message.reply_text(
            "不能警告超级管理员。"
        )
    if user_id in (await list_admins(chat_id)):
        return await message.reply_text(
            "不能警告群管理员。"
        )
    user, warns = await asyncio.gather(
        app.get_users(user_id),
        get_warn(chat_id, await int_to_alpha(user_id)),
    )
    mention = user.mention
    keyboard = ikb({"撤销一次警告": f"unwarn_{user_id}"})
    if warns:
        warns = warns["warns"]
    else:
        warns = 0
    if message.command[0][0] == "d":
        await message.reply_to_message.delete()
    if warns >= 2:
        await message.chat.ban_member(user_id)
        await message.reply_text(
            f"{mention} 已累计 3 次警告，现已封禁。"
        )
        await remove_warns(chat_id, await int_to_alpha(user_id))
    else:
        warn = {"warns": warns + 1}
        msg = f"""
**已警告：** {mention}
**操作者：** {message.from_user.mention if message.from_user else '匿名管理员'}
**原因：** {reason or '未填写原因'}
**警告次数：** {warns + 1}/3"""
        replied_message = message.reply_to_message
        if replied_message:
            message = replied_message
        await message.reply_text(msg, reply_markup=keyboard)
        await add_warn(chat_id, await int_to_alpha(user_id), warn)


@app.on_callback_query(filters.regex("unwarn_"))
async def remove_warning(_, cq: CallbackQuery):
    from_user = cq.from_user
    chat_id = cq.message.chat.id
    permissions = await member_permissions(chat_id, from_user.id)
    permission = "can_restrict_members"
    if permission not in permissions:
        return await cq.answer(
            "你没有执行此操作所需的限制成员权限。",
            show_alert=True,
        )
    user_id = cq.data.split("_")[1]
    warns = await get_warn(chat_id, await int_to_alpha(user_id))
    if warns:
        warns = warns["warns"]
    if not warns or warns == 0:
        return await cq.answer("该成员没有警告记录。")
    warn = {"warns": warns - 1}
    await add_warn(chat_id, await int_to_alpha(user_id), warn)
    text = cq.message.text.markdown
    text = f"~~{text}~~\n\n"
    text += f"__{from_user.mention} 已撤销一次警告。__"
    await cq.message.edit(text)


# Rmwarns


@app.on_message(filters.command("rmwarns") & ~filters.private)
@adminsOnly("can_restrict_members")
async def remove_warnings(_, message: Message):
    if not message.reply_to_message:
        return await message.reply_text(
            "请回复该成员的消息以清除警告。"
        )
    user_id = message.reply_to_message.from_user.id
    mention = message.reply_to_message.from_user.mention
    chat_id = message.chat.id
    warns = await get_warn(chat_id, await int_to_alpha(user_id))
    if warns:
        warns = warns["warns"]
    if warns == 0 or not warns:
        await message.reply_text(f"{mention} 没有警告记录。")
    else:
        await remove_warns(chat_id, await int_to_alpha(user_id))
        await message.reply_text(f"已清除 {mention} 的全部警告。")


# Warns


@app.on_message(filters.command("warns") & ~filters.private)
@capture_err
async def check_warns(_, message: Message):
    user_id = await extract_user(message)
    if not user_id:
        return await message.reply_text("未找到该成员。")
    warns = await get_warn(message.chat.id, await int_to_alpha(user_id))
    mention = (await app.get_users(user_id)).mention
    if warns:
        warns = warns["warns"]
    else:
        return await message.reply_text(f"{mention} 没有警告记录。")
    return await message.reply_text(f"{mention} 当前有 {warns}/3 次警告。")


# Report


@app.on_message(
    (
        filters.command("report")
        | filters.command(["admins", "admin"], prefixes="@")
    )
    & ~filters.private
)
@capture_err
async def report_user(_, message):
    if len(message.text.split()) <= 1 and not message.reply_to_message:
        return await message.reply_text(
            "请回复要举报的消息。"
        )

    reply = message.reply_to_message if message.reply_to_message else message
    reply_id = reply.from_user.id if reply.from_user else reply.sender_chat.id
    user_id = (
        message.from_user.id if message.from_user else message.sender_chat.id
    )

    list_of_admins = await list_admins(message.chat.id)
    linked_chat = (await app.get_chat(message.chat.id)).linked_chat
    if linked_chat is not None:
        if (
            reply_id in list_of_admins
            or reply_id == message.chat.id
            or reply_id == linked_chat.id
        ):
            return await message.reply_text(
                "不能举报群管理员或关联频道。"
            )
    else:
        if reply_id in list_of_admins or reply_id == message.chat.id:
            return await message.reply_text(
                "不能举报群管理员。"
            )

    user_mention = (
        reply.from_user.mention if reply.from_user else reply.sender_chat.title
    )
    text = f"已向管理员举报 {user_mention}。"
    admin_data = [
        i
        async for i in app.get_chat_members(
            chat_id=message.chat.id, filter=ChatMembersFilter.ADMINISTRATORS
        )
    ]  # will it give floods ???
    for admin in admin_data:
        if admin.user.is_bot or admin.user.is_deleted:
            # return bots or deleted admins
            continue
        text += f"[\u2063](tg://user?id={admin.user.id})"

    await reply.reply_text(text)


@app.on_message(filters.command("invite"))
@adminsOnly("can_invite_users")
async def invite(_, message):
    if message.chat.type in [ChatType.GROUP, ChatType.SUPERGROUP]:
        link = (await app.get_chat(message.chat.id)).invite_link
        if not link:
            link = await app.export_chat_invite_link(message.chat.id)
        text = f"本群邀请链接：\n\n{link}"
        if message.reply_to_message:
            await message.reply_to_message.reply_text(
                text, disable_web_page_preview=True
            )
        else:
            await message.reply_text(text, disable_web_page_preview=True)