PagerMaid_Plugins_Pyro/pmcaptcha/main.py

# -*- coding: utf-8 -*-
"""PMCaptcha v2 - An plugin a day keeps the ads away
v1 by xtaodata and cloudreflection
v2 by Sam
"""

import asyncio
import gc
import html
import inspect
import json
import re
import time
import traceback
from base64 import b64decode, b64encode
from dataclasses import dataclass, field
from io import BytesIO
from random import randint
from typing import Optional, Callable, Union, List, Any, Dict, Coroutine

from pyrogram.enums.chat_type import ChatType
from pyrogram.enums.parse_mode import ParseMode
from pyrogram.errors import (
    FloodWait,
    AutoarchiveNotAvailable,
    ChannelsAdminPublicTooMuch,
    BotResponseTimeout,
    PeerIdInvalid,
)
from pyrogram.raw.functions import messages
from pyrogram.raw.functions.account import (
    SetGlobalPrivacySettings,
    GetGlobalPrivacySettings,
)
from pyrogram.raw.functions.channels import UpdateUsername
from pyrogram.raw.types import GlobalPrivacySettings
from pyrogram.types import User, Sticker

from pagermaid import bot, logs
from pagermaid.config import Config
from pagermaid.listener import listener
from pagermaid.single_utils import sqlite
from pagermaid.sub_utils import Sub
from pagermaid.utils import Message, alias_command

cmd_name = "pmcaptcha"

lang_version = "2.25"

log_collect_bot = img_captcha_bot = "PagerMaid_Sam_Bot"

# Get alias for user command
user_cmd_name = alias_command(cmd_name)


def _sort_line_number(m):
    try:
        func = getattr(m[1], "__func__", m[1])
        return func.__code__.co_firstlineno
    except AttributeError:
        return -1


async def log(message: str, remove_prefix: bool = False):
    console.info(message.replace("`", '"'))
    Config.LOG and logging.send_log(message, remove_prefix)


def get_version():
    from pagermaid import working_dir
    from os import sep
    from json import load

    with open(
        f"{working_dir}{sep}plugins{sep}version.json", "r", encoding="utf-8"
    ) as f:
        version_json = load(f)
    return version_json.get(cmd_name, "unknown")


# region Text Formatting


def code(text: str) -> str:
    return f"<code>{text}</code>"


def italic(text: str) -> str:
    return f"<i>{text}</i>"


def bold(text: str) -> str:
    return f"<b>{text}</b>"


def gen_link(text: str, url: str) -> str:
    return f'<a href="{url}">{text}</a>'


def str_timestamp(unix_ts: int) -> str:
    import datetime

    date_time = datetime.datetime.fromtimestamp(
        unix_ts, datetime.timezone(datetime.timedelta(hours=8))
    )
    return date_time.strftime("%Y-%m-%dT%XZ%z")


# endregion


def get_lang_list():  # Yes, blocking
    from httpx import Client

    endpoint = f"https://raw.githubusercontent.com/TeamPGM/PMCaptcha-i18n/main/v{lang_version}.py"
    for _ in range(3):
        try:
            with Client() as client:
                response = client.get(endpoint)
                if response.status_code == 200:
                    return eval(f"lambda: {response.text}")()
        except Exception as e:
            console.error(f"Failed to get language file: {e}\n{traceback.format_exc()}")
    exit(1)


# Language file
lang_dict = get_lang_list()


def lang(lang_id: str, lang_code: str = Config.LANGUAGE or "en") -> str:
    lang_code = lang_code or "en"
    return lang_dict.get(
        lang_id, [f"Get lang failed[{code(lang_id)}]", f"获取语言失败[{code(lang_id)}]"]
    )[1 if lang_code.startswith("zh") else 0]


def lang_full(lang_id: str, *format_args):
    return "\n".join(
        lang_str % format_args
        for lang_str in lang_dict.get(
            lang_id, [f"Get lang failed[{code(lang_id)}]", f"获取语言失败[{code(lang_id)}]"]
        )
    )


async def exec_api(coro_func: Coroutine):
    for _ in range(3):
        try:
            return await coro_func
        except FloodWait as e:
            console.debug(f"API Flood triggered, waiting for {e.value} seconds")
            await asyncio.sleep(e.value)
        except Exception as e:
            console.error(f"Function Call Failed: {e}\n{traceback.format_exc()}")
            return None


@dataclass
class Log:
    task: Optional[asyncio.Task] = None
    queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    last_send_time: int = 0

    async def worker(self):
        while True:
            text = None
            try:
                if int(time.time()) - self.last_send_time < 5:
                    await asyncio.sleep(5 - (int(time.time()) - self.last_send_time))
                    continue
                (text,) = await self.queue.get()
                await bot.send_message(Config.LOG_ID, text, ParseMode.HTML)
                self.last_send_time = int(time.time())
            except asyncio.CancelledError:
                break
            except FloodWait as e:
                console.debug(f"Flood triggered when sending log, wait for {e.value}s")
                await asyncio.sleep(e.value)
            except Exception as e:
                console.error(f"Error when sending log: {e}\n{traceback.format_exc()}")
            finally:
                text and self.queue.task_done()

    def send_log(self, message: str, remove_prefix: bool):
        if not self.task or self.task.done():
            self.task = asyncio.create_task(self.worker())
        message = message if remove_prefix else " ".join(("[PMCaptcha]", message))
        self.queue.put_nowait((message,))


@dataclass
class Setting:
    key_name: str
    whitelist: Sub = field(init=False)
    pending_ban_list: Sub = field(init=False)
    pending_challenge_list: Sub = field(init=False)

    def __post_init__(self):
        self.whitelist = Sub("pmcaptcha.success")
        self.pending_ban_list = Sub("pmcaptcha.pending_ban")
        self.pending_challenge_list = Sub("pmcaptcha.pending_challenge")

    def get(self, key: str, default=None):
        if sqlite.get(self.key_name) is None:
            return default
        return sqlite[self.key_name].get(key, default)

    def set(self, key: str, value: Any):
        """Set the value of a key in the database, return value"""
        if (data := sqlite.get(self.key_name)) is None:
            sqlite[self.key_name] = data = {}
        data.update({key: value})
        sqlite[self.key_name] = data
        return value

    def delete(self, key: str):
        """Delete a key in the database, if key exists"""
        if self.get(key) is not None:
            data = sqlite[self.key_name]
            del data[key]
            sqlite[self.key_name] = data
        return self

    def is_verified(self, user_id: int) -> bool:
        return self.whitelist.check_id(user_id) or user_id == bot.me.id

    # region Captcha Challenge

    def get_challenge_state(self, user_id: int) -> dict:
        return sqlite.get(f"{self.key_name}.challenge.{user_id}", {})

    def set_challenge_state(self, user_id: int, state: dict):
        sqlite[f"{self.key_name}.challenge.{user_id}"] = state
        return state

    def del_challenge_state(self, user_id: int):
        key = f"{self.key_name}.challenge.{user_id}"
        if sqlite.get(key):
            del sqlite[key]

    # endregion

    # region Flood State

    def get_flood_state(self) -> dict:
        return sqlite.get(f"{self.key_name}.flood_state", {})

    def set_flood_state(self, state: dict) -> dict:
        sqlite[f"{self.key_name}.flood_state"] = state
        return state

    def del_flood_state(self):
        key = f"{self.key_name}.flood_state"
        if sqlite.get(key):
            del sqlite[key]

    # endregion


@dataclass
class Command:
    user: User
    msg: Message

    # Regex
    alias_rgx = r":alias: (.+)"
    param_rgx = r":param (opt)?\s?(\w+):\s?(.+)"

    # region Helpers (Formatting, User ID)

    @classmethod
    def _generate_markdown(cls):
        self = cls(None, None)  # noqa
        result = [f"版本: v{get_version()}", ""]
        members = inspect.getmembers(self, inspect.iscoroutinefunction)
        members.sort(key=_sort_line_number)
        for name, func in members:
            if name.startswith("_"):
                continue
            result.append(self._extract_docs(func.__name__, func.__doc__ or "", True))
        return "\n".join(result)

    async def _run_command(self):
        command = len(self.msg.parameter) > 0 and self.msg.parameter[0] or cmd_name
        if not (func := self[command]):
            return False, "NOT_FOUND", command
        full_arg_spec = inspect.getfullargspec(func)
        args_len = None if full_arg_spec.varargs else len(full_arg_spec.args)
        cmd_args = self.msg.parameter[1:args_len]
        func_args = []
        for index, arg_type in enumerate(
            tuple(full_arg_spec.annotations.values())
        ):  # Check arg type
            if args_len is None:
                func_args = cmd_args
                break
            try:
                if getattr(arg_type, "__origin__", None) == Union:
                    NoneType = type(None)
                    if (
                        len(arg_type.__args__) != 2
                        or arg_type.__args__[1] is not NoneType
                    ):
                        continue
                    if (
                        len(cmd_args) - 1 > index
                        and not cmd_args[index]
                        or len(cmd_args) - 1 < index
                    ):
                        func_args.append(None)
                        continue
                    arg_type = arg_type.__args__[0]
                func_args.append(arg_type(cmd_args[index]))
            except ValueError:
                return (
                    False,
                    "INVALID_PARAM",
                    tuple(full_arg_spec.annotations.keys())[index],
                )
            except IndexError:  # No more args
                await self.help(command)
                return True, None, None
        try:
            await func(*func_args)
        except Exception as e:
            console.error(
                f"Error when running command {command}: {e}\n{traceback.format_exc()}"
            )
            await self.msg.edit(
                lang("cmd_err_run")
                % (self._get_user_cmd_input(), str(e), traceback.format_exc())
            )
        return True, None, None

    def _get_user_cmd_input(self) -> str:
        return f",{user_cmd_name} {' '.join(self.msg.parameter)}"

    def _extract_docs(self, subcmd_name: str, text: str, markdown: bool = False) -> str:
        extras = []
        if result := re.search(self.param_rgx, text):
            is_optional = (
                f"({italic(lang('cmd_param_optional'))} ) " if result[1] else ""
            )
            extras.extend(
                (
                    f"{lang('cmd_param')}:",
                    f"{is_optional}{code(result[2].lstrip('_'))} - {result[3]}",
                )
            )

            text = re.sub(self.param_rgx, "", text)
        if result := re.search(self.alias_rgx, text):
            alias = result[1].replace(" ", "").split(",")
            alia_text = ", ".join(code(a) for a in alias)
            extras.append(f"{lang('cmd_alias')}: {alia_text}")
            text = re.sub(self.alias_rgx, "", text)
        len(extras) and extras.insert(0, "")
        cmd_display = code(
            f",{cmd_name} {self._get_cmd_with_param(subcmd_name)}".strip()
        )
        if markdown:
            result = [
                "<details>",
                f"<summary>{self._get_cmd_with_param(subcmd_name, markdown) or code(cmd_name)} · {re.search(r'(.+)', self[subcmd_name].__doc__ or '')[1].strip()}</summary>",
                "\n>\n",
                f"用法：{cmd_display}",
                "",
                re.sub(r" {4,}", "", text)
                .replace("{cmd_name}", cmd_name)
                .strip()
                .replace("\n", "\n\n"),
                "\n\n".join(extras),
                "",
                "---",
                "</details>",
            ]
            return "\n".join(result)
        return "\n".join(
            [
                cmd_display,
                re.sub(r" {4,}", "", text).replace("{cmd_name}", cmd_name).strip(),
            ]
            + extras
        )

    def _get_cmd_with_param(self, subcmd_name: str, markdown: bool = False) -> str:
        if subcmd_name == cmd_name:
            return ""
        msg = subcmd_name
        if result := re.search(self.param_rgx, getattr(self, msg).__doc__ or ""):
            if markdown:
                msg = f"<code>{msg}</code>"
            param = result[2].lstrip("_")
            msg += f" [{param}]" if result[1] else html.escape(f" <{param}>")
        elif markdown:
            msg = f"<code>{msg}</code>"
        return msg

    def _get_mapped_alias(self, alias_name: str, ret_type: str):
        # Get alias function
        for name, func in inspect.getmembers(self, inspect.iscoroutinefunction):
            if name.startswith("_"):
                continue
            if (
                result := re.search(self.alias_rgx, func.__doc__ or "")
            ) and alias_name in result[1].replace(" ", "").split(","):
                return func if ret_type == "func" else name

    async def _display_value(
        self,
        *,
        key: Optional[str] = None,
        display_text: str,
        sub_cmd: str,
        value_type: str,
    ):
        text = [
            display_text,
            "",
            lang("tip_edit")
            % html.escape(f",{user_cmd_name} {sub_cmd} <{lang(value_type)}>"),
        ]
        key and text.insert(0, f'{lang(f"{key}_curr_rule")}:')
        return await self._edit("\n".join(text))

    # Set On / Off Boolean
    async def _set_toggle(self, key: str, toggle: str, *, reverse: bool = False):
        if (toggle := toggle.lower()[0]) not in ("y", "n", "t", "f", "1", "0") and (
            toggle := toggle.lower()
        ) not in ("on", "off"):
            return await self.help(key)
        toggle = toggle in ("y", "t", "1", "on")
        not reverse and (toggle and setting.set(key, True) or setting.delete(key))
        reverse and (toggle and setting.delete(key) or setting.set(key, False))
        await self._edit(lang(f"{key}_set") % lang("enabled" if toggle else "disabled"))

    async def _get_user_id(self, user_id: Union[str, int]) -> Optional[int]:
        if (
            not user_id
            and not self.msg.reply_to_message_id
            and self.msg.chat.type != ChatType.PRIVATE
        ):
            return
        try:
            if int(user_id) < 0:
                return
        except (ValueError, TypeError):
            pass
        user = None
        user_id = user_id or (
            self.msg.reply_to_message
            and self.msg.reply_to_message.from_user.id
            or self.msg.chat.id
        )
        try:
            if (
                not user_id
                or not (user := await bot.get_users(user_id))
                or (user.is_bot or user.is_verified or user.is_deleted)
            ):
                return
        except (ValueError, PeerIdInvalid):
            return
        return user.id

    # Set Black / White List
    async def _set_list(self, _type: str, array: str):
        if not array:
            return await self._display_value(
                key=_type,
                display_text=code(setting.get(_type, lang("none"))),
                sub_cmd=f"{_type[0]}l",
                value_type="vocab_array",
            )
        if array.startswith("-c"):
            setting.delete(_type)
            return await self._edit(lang(f"{_type}_reset"))
        setting.set(_type, array.replace(" ", "").split(","))
        await self._edit(lang(f"{_type}_set"))

    # endregion

    async def _edit(self, msg: str):
        text = "\n\n".join((f">>> {code(self._get_user_cmd_input())}", msg))
        return await self.msg.edit_text(
            text, parse_mode=ParseMode.HTML, disable_web_page_preview=True
        )

    def __getitem__(self, cmd: str) -> Optional[Callable]:
        # Get subcommand function
        if func := getattr(self, cmd, None):
            return func
        # Check for alias
        if func := self._get_mapped_alias(cmd, "func"):
            return func
        return  # Not found

    def get(self, cmd: str, default=None):
        return self[cmd] or default

    async def pmcaptcha(self):
        """查询当前用户的验证状态"""
        if not (user_id := await self._get_user_id(self.msg.chat.id)):
            return await self._edit(lang("invalid_user_id"))
        await self._edit(
            lang(f'verify_{"" if setting.is_verified(user_id) else "un"}verified')
        )
        await asyncio.sleep(5)
        await self.msg.safe_delete()

    async def version(self):
        """查看 <code>PMCaptcha</code> 当前版本

        :alias: v, ver
        """
        await self._edit(f"{lang('curr_version') % get_version()}")

    async def help(self, command: Optional[str], search_str: Optional[str] = None):
        """显示指令帮助信息，使用 <code>,{cmd_name} search [搜索内容]</code> 进行文档、指令(和别名)搜索

        :param opt command: 命令名称
        :param opt search_str: 搜索的文字，只有 <code>command</code> 为 <code>search</code> 时有效
        :alias: h
        """
        help_msg = [f"{code('PMCaptcha')} {lang('cmd_list')}:", ""]
        footer = [
            italic(lang("cmd_detail")),
            "",
            f"{lang('priority')}:\n{' > '.join(Rule._get_rules_priority())}",
            "",
            f"遇到任何问题请先 {code(',apt update')} 、 {code(',restart')} 后复现再反馈",
            (
                f"👉 {gen_link('捐赠网址', 'https://afdian.net/@xtaodada')} "
                f"{gen_link('捐赠说明', 'https://t.me/PagerMaid_Modify/121')} "
                f"(v{get_version()})"
            ),
        ]
        if command == "search":  # Search for commands or docs
            if not search_str:
                return await self.help("h")
            search_str = search_str.lower()
            search_results = [lang("cmd_search_result") % search_str]
            have_cmd = False
            have_doc = False
            for name, func in inspect.getmembers(self, inspect.iscoroutinefunction):
                if name.startswith("_"):
                    continue
                # Search for docs
                docs = func.__doc__ or ""
                if docs.lower().find(search_str) != -1:
                    not have_doc and search_results.append(
                        f"{lang('cmd_search_docs')}:"
                    )
                    have_doc = True
                    search_results.append(self._extract_docs(func.__name__, docs))
                # Search for commands
                if name.find(search_str) != -1:
                    not have_cmd and search_results.append(
                        f"{lang('cmd_search_cmds')}:"
                    )
                    have_cmd = True
                    search_results.append(
                        f"""{code(f"- {code(self._get_cmd_with_param(name))}".strip())}\n· {re.search('(.+)', docs)[1].strip()}\n"""
                    )
                elif result := re.search(self.alias_rgx, docs):
                    if search_str not in result[1].replace(" ", "").split(","):
                        continue
                    not have_cmd and search_results.append(
                        f"{lang('cmd_search_cmds')}:"
                    )
                    have_cmd = True
                    search_results.append(
                        f"""{f"* {code(search_str)} -> {code(self._get_cmd_with_param(func.__name__))}".strip()}\n· {re.search('(.+)', docs)[1].strip()}\n"""
                    )
            len(search_results) == 1 and search_results.append(
                italic(lang("cmd_search_none"))
            )
            return await self._edit("\n\n".join(search_results))
        elif command:  # Single command help
            func = getattr(self, command, self._get_mapped_alias(command, "func"))
            return await (
                self._edit(self._extract_docs(func.__name__, func.__doc__ or ""))
                if func
                else self._edit(f"{lang('cmd_not_found')}: {code(command)}")
            )
        members = inspect.getmembers(self, inspect.iscoroutinefunction)
        members.sort(key=_sort_line_number)
        for name, func in members:
            if name.startswith("_"):
                continue
            help_msg.append(
                (
                    code(f",{user_cmd_name} {self._get_cmd_with_param(name)}".strip())
                    + f"\n· {re.search(r'(.+)', func.__doc__ or '')[1].strip()}\n"
                )
            )
        await self._edit("\n".join(help_msg + footer))

    # region Checking User / Manual update

    async def check(self, _id: Optional[str]):
        """查询指定用户验证状态，对该信息回复或者输入用户 ID，如未指定为当前私聊用户 ID

        :param opt _id: 用户 ID
        """
        if not (user_id := await self._get_user_id(_id)):
            return await self._edit(lang("invalid_user_id"))
        await self._edit(
            lang(f"user_{'' if setting.is_verified(user_id) else 'un'}verified")
            % user_id
        )

    async def add(self, _id: Optional[str]):
        """将 ID 加入已验证，对该信息回复或者输入用户 ID，如未指定为当前私聊用户 ID

        :param opt _id: 用户 ID
        """
        if not (user_id := await self._get_user_id(_id)):
            return await self._edit(lang("invalid_user_id"))
        if captcha := curr_captcha.get(
            user_id
        ):  # This user is currently in challenge state
            await captcha.action(True)
            if curr_captcha.get(user_id):
                del curr_captcha[user_id]
            result = True
        else:
            result = setting.whitelist.add_id(user_id)
            await bot.unarchive_chats(chat_ids=user_id)
        await self._edit(
            lang(f"add_whitelist_{'success' if result else 'failed'}") % user_id
        )

    async def delete(self, _id: Optional[str]):
        """移除 ID 验证记录，对该信息回复或者输入用户 ID，如未指定为当前私聊用户 ID

        :param opt _id: 用户 ID
        :alias: del
        """
        if not (user_id := await self._get_user_id(_id)):
            return await self._edit(lang("invalid_user_id"))
        text = lang(
            f"remove_verify_log_{'success' if setting.whitelist.del_id(user_id) else 'not_found'}"
        )
        await self._edit(text % user_id)

    # endregion

    async def unstuck(self, _id: Optional[str]):
        """解除一个用户的验证状态，通常用于解除卡死的验证状态
        使用：对该信息回复或者输入用户 ID，如未指定为当前私聊用户 ID

        :param opt _id: 用户 ID
        """
        if not (user_id := await self._get_user_id(_id)):
            return await self._edit(lang("invalid_user_id"))
        captcha = None
        if (state := setting.get_challenge_state(user_id)) or (
            captcha := curr_captcha.get(user_id)
        ):
            await CaptchaTask.archive(user_id, un_archive=True)
            try:
                (
                    captcha and captcha.type or state.get("type", "math")
                ) == "img" and await bot.unblock_user(user_id)
            except Exception as e:
                console.error(
                    f"Error when unblocking user {user_id}: {e}\n{traceback.format_exc()}"
                )
            if captcha := curr_captcha.get(user_id):
                captcha.timer_task and captcha.timer_task.cancel()
                del curr_captcha[user_id]
            state and setting.del_challenge_state(user_id)
            return await self._edit(lang("unstuck_success") % user_id)
        await self._edit(lang("not_stuck") % user_id)

    async def welcome(self, *message: Optional[str]):
        """查看或设置验证通过时发送的消息
        使用 <code>,{cmd_name} welcome -c</code> 可恢复默认规则

        :param opt message: 消息内容
        :alias: wel
        """
        if not message:
            return await self._display_value(
                key="welcome",
                display_text=code(setting.get("welcome", lang("none"))),
                sub_cmd="wel",
                value_type="vocab_msg",
            )
        message = " ".join(message)
        if message.startswith("-c"):
            setting.delete("welcome")
            return await self._edit(lang("welcome_reset"))
        setting.set("welcome", message)
        await self._edit(lang("welcome_set"))

    async def whitelist(self, array: Optional[str]):
        """查看或设置关键词白名单列表（英文逗号分隔）
        使用 <code>,{cmd_name} whitelist -c</code> 可清空列表

        :param opt array: 白名单列表 (英文逗号分隔)
        :alias: wl, whl
        """
        return await self._set_list("whitelist", array)

    async def blacklist(self, array: Optional[str]):
        """查看或设置关键词黑名单列表 (英文逗号分隔)
        使用 <code>,{cmd_name} blacklist -c</code> 可清空列表

        :param opt array: 黑名单列表 (英文逗号分隔)
        :alias: bl
        """
        return await self._set_list("blacklist", array)

    async def timeout(self, seconds: Optional[int], _type: Optional[str]):
        """查看或设置超时时间，默认为 <code>30</code> 秒、图像模式为 <code>5</code> 分钟
        使用 <code>,{cmd_name} wait off</code> 可关闭验证时间限制

        有关验证超时的默认选项：
        - <code>math</code> | <code>30</code> 秒
        - <code>img</code> | <code>5</code> 分钟
        - <code>sticker</code> | <code>30</code> 秒

        在图像模式中，此超时时间会于用户最后活跃而重置，
        建议数值设置大一点让机器人有一个时间可以处理后端操作

        :param opt seconds: 超时时间，单位秒
        :param opt _type: 验证类型，默认为当前类型
        :alias: wait
        """
        if _type and _type not in ("math", "img", "sticker"):
            return await self.help("wait")
        captcha_type: str = _type or setting.get("type", "math")
        key_name: str = {
            "sticker": "sticker_timeout",
            "img": "img_timeout",
            "math": "timeout",
        }.get(captcha_type)
        default_timeout_time: int = {"sticker": 30, "img": 300, "math": 30}.get(
            captcha_type
        )
        if seconds is None:
            return await self._display_value(
                display_text=lang("timeout_curr_rule")
                % int(setting.get(key_name, default_timeout_time)),
                sub_cmd="wait",
                value_type="vocab_int",
            )
        elif seconds == "off":
            setting.delete(key_name)
            return await self._edit(lang("timeout_off"))
        if seconds < 0:
            return await self._edit(lang("invalid_param"))
        setting.set(key_name, seconds)
        await self._edit(lang("timeout_set") % seconds)

    async def disable_pm(self, toggle: Optional[str]):
        """启用 / 禁止陌生人私聊，默认为 <code>关闭</code> （允许私聊）
        此功能会放行联系人和白名单(<i>已通过验证</i>)用户
        您可以使用 <code>,{cmd_name} add</code> 将用户加入白名单

        :param opt toggle: 开关 (y / n)
        :alias: disablepm, disable
        """
        if not toggle:
            return await self._display_value(
                display_text=lang("disable_pm_curr_rule")
                % lang("enabled" if setting.get("disable") else "disabled"),
                sub_cmd="disable_pm",
                value_type="vocab_bool",
            )
        await self._set_toggle("disable", toggle)

    async def stats(self, arg: Optional[str]):
        """查看验证统计
        使用 <code>,{cmd_name} stats -c</code> 重置数据

        :param opt arg: 参数 (reset)
        """
        if not arg:
            data = (
                setting.get("pass", 0) + setting.get("banned", 0),
                setting.get("pass", 0),
                setting.get("banned", 0),
                len(curr_captcha) + len(setting.pending_challenge_list.get_subs()),
                len(setting.pending_ban_list.get_subs()),
                setting.get("flooded", 0),
            )
            text = f"{code('PMCaptcha')} {lang('stats_display') % data}"
            if the_world_eye.triggered:
                text += f"\n\n{lang('stats_flooding') % len(the_world_eye.user_ids)}"
            await self.msg.edit_text(text, parse_mode=ParseMode.HTML)
            return
        if arg.startswith("-c"):
            setting.delete("pass").delete("banned").delete("flooded")
            return await self.msg.edit(lang("stats_reset"), parse_mode=ParseMode.HTML)

    async def action(self, action: Optional[str]):
        """选择验证失败的处理方式，默认为 <code>封禁</code>
        处理方式如下：
        - <code>ban</code> | 封禁
        - <code>delete</code> | 封禁并删除对话
        - <code>none</code> | 不执行任何操作

        :param opt action: 处理方式
        :alias: act
        """
        if not action:
            action = setting.get("action", "ban")
            return await self._display_value(
                key="action",
                display_text=lang(
                    f"action_{action == 'none' and 'set_none' or action}"
                ),
                sub_cmd="act",
                value_type="vocab_action",
            )
        if action not in ("ban", "delete", "none"):
            return await self.help("act")
        action == "ban" and setting.delete("action") or setting.set("action", action)
        if action == "none":
            return await self._edit(lang("action_set_none"))
        return await self._edit(lang("action_set") % lang(f"action_{action}"))

    async def report(self, toggle: Optional[str]):
        """选择验证失败后是否举报该用户，默认为 <code>开启</code>

        :param opt toggle: 开关 (y / n)
        """
        if not toggle:
            return await self._display_value(
                display_text=lang("report_curr_rule")
                % lang("enabled" if setting.get("report", True) else "disabled"),
                sub_cmd="report",
                value_type="vocab_bool",
            )
        await self._set_toggle("report", toggle, reverse=True)

    async def premium(self, action: Optional[str]):
        """选择对 <b>Premium</b> 用户的操作，默认为 <code>不执行任何操作</code>
        处理方式如下：
        - <code>allow</code> | 白名单
        - <code>ban</code> | 封禁
        - <code>only</code> | 只允许
        - <code>none</code> | 不执行任何操作

        :param opt action: 处理方式
        :alias: vip, prem
        """
        if not action:
            return await self._display_value(
                key="premium",
                display_text=lang(f'premium_set_{setting.get("premium", "none")}'),
                sub_cmd="vip",
                value_type="vocab_action",
            )
        if action not in ("allow", "ban", "only", "none"):
            return await self.help("vip")
        action == "none" and setting.delete("action") or setting.set("action", action)
        await self._edit(lang(f"premium_set_{action}"))

    async def groups_in_common(self, count: Optional[int]):
        """设置是否对拥有一定数量的共同群的用户添加白名单
        使用 <code>,{cmd_name} groups -1</code> 重置设置

        :param opt count: 共同群数量
        :alias: group, groups, common
        """
        if not count:
            groups = setting.get("groups_in_common")
            text = lang(
                f"groups_in_common_{'set' if groups is not None else 'disabled'}"
            )
            if groups is not None:
                text = text % groups
            return await self._display_value(
                display_text=text, sub_cmd="groups", value_type="vocab_int"
            )
        if count == -1:
            setting.delete("groups_in_common")
            return await self._edit(lang("groups_in_common_disable"))
        elif count < 0:
            return await self.help("groups_in_common")
        setting.set("groups_in_common", count)
        await self._edit(lang("groups_in_common_set") % count)

    async def chat_history(self, count: Optional[int]):
        """设置对拥有一定数量的聊天记录的用户添加白名单（触发验证的信息不计算在内）
        使用 <code>,{cmd_name} his -1</code> 重置设置

        <b>请注意，由于 <code>Telegram</code> 内部限制，数值过大会导致程序缓慢，请不要设置过大的数值</b>

        :param opt count: 聊天记录数量
        :alias: his, history
        """
        if not count:
            history_count = setting.get("history_count")
            text = lang(
                "chat_history_curr_rule"
                if history_count is not None
                else "chat_history_disabled"
            )
            if history_count is not None:
                text = text % history_count
            return await self._display_value(
                display_text=text, sub_cmd="his", value_type="vocab_bool"
            )
        count < 0 and setting.delete("history_count") or setting.set(
            "history_count", count
        )
        await self._edit(lang("chat_history_curr_rule") % count)

    async def initiative(self, toggle: Optional[str]):
        """设置对主动进行对话的用户添加白名单，默认为 <code>开启</code>

        :param opt toggle: 开关 (y / n)
        """
        if not toggle:
            return await self._display_value(
                display_text=lang("initiative_curr_rule")
                % lang("enabled" if setting.get("initiative", True) else "disabled"),
                sub_cmd="initiative",
                value_type="vocab_bool",
            )
        await self._set_toggle("initiative", toggle, reverse=True)

    async def silent(self, toggle: Optional[str]):
        """减少信息发送，默认为 <code>关闭</code>
        开启后，封禁、验证成功提示（包括欢迎信息）信息将不会发送
        （并不会影响到 <code>log</code> 发送）

        :param opt toggle: 开关 (y / n)
        :alias: quiet
        """
        if not toggle:
            return await self._display_value(
                display_text=lang("silent_curr_rule")
                % lang("enabled" if setting.get("silent") else "disabled"),
                sub_cmd="quiet",
                value_type="vocab_bool",
            )
        await self._set_toggle("silent", toggle)

    async def flood(self, limit: Optional[int]):
        """设置一分钟内超过 <code>n</code> 人开启轰炸检测机制，默认为 <code>5</code> 人
        此机制会在用户被轰炸时启用，持续 <code>5</code> 分钟，假如有用户继续进行私聊计时将会重置

        当轰炸开始时，<code>PMCaptcha</code> 将会启动以下一系列机制
        - 强制开启自动归档（无论是否 <code>Telegram Premium</code> 用户都会尝试开启）
        - 不向用户发送 <code>CAPTCHA</code> 挑战
        - 继上面的机制，记录未发送 <code>CAPTCHA</code> 的用户 ID
        - （用户可选）创建临时频道，并把用户名转移到创建的频道上 【默认关闭】

        轰炸结束后，如果用户名已转移到频道上，将恢复用户名，并删除频道
        并对记录收集机器人发送轰炸的<code>用户数量</code>、<code>轰炸开始时间</code>、<code>轰炸结束时间</code>、<code>轰炸时长</code>
        <i>（由于不存在隐私问题，此操作为强制性）</i>

        请参阅 <code>,{cmd_name} h flood_username</code> 了解更多有关创建临时频道的机制
        请参阅 <code>,{cmd_name} h flood_act</code> 查看轰炸结束后的处理方式

        :param opt limit: 人数限制
        :alias: boom
        """
        if not limit:
            return await self._display_value(
                display_text=lang("flood_curr_rule") % setting.get("flood_limit", 50),
                sub_cmd="flood",
                value_type="vocab_int",
            )
        setting.set("flood_limit", limit)
        await self._edit(lang("flood_curr_rule") % limit)

    async def flood_username(self, toggle: Optional[str]):
        """设置是否在轰炸时启用“转移用户名到临时频道”机制（如有用户名）
        将此机制分开出来的原因是此功能有可能会被抢注用户名<i>(虽然经测试并不会出现此问题)</i>
        但为了以防万一依然分开出来作为一个选项了

        启用后，在轰炸机制开启时，会进行以下操作
        - 创建临时频道
        - （如创建成功）清空用户名，设置用户名为临时频道，并在频道简介设置正在受到轰炸提示
        - （如设置失败）恢复用户名，删除频道

        注意：<b>请预留足够的公开群用户名设置额度，否则将不会设置成功，但同时用户名也不会被清空</b>
        <i>（操作失败虽然会有 log 提醒，但请不要过度依赖 log）</i>

        :param opt toggle: 开关 (y / n)
        :alias: boom_username
        """
        global user_want_set_flood_username
        if not toggle:
            return await self._display_value(
                display_text=lang("flood_username_curr_rule")
                % lang("enabled" if setting.get("flood_username") else "disabled"),
                sub_cmd="flood_username",
                value_type="vocab_bool",
            )
        if toggle in ("y", "t", "1", "on") and not user_want_set_flood_username:
            user_want_set_flood_username = True
            return await self._edit(lang("flood_username_set_confirm"))
        user_want_set_flood_username = None
        await self._set_toggle("flood_username", toggle)

    async def flood_act(self, action: Optional[str]):
        """设置轰炸结束后进行的处理方式，默认为 <code>删除并举报所有轰炸的用户</code>
        可用的处理方式如下：
        - <code>asis</code> | 与验证失败的处理方式一致，但不会进行验证失败通知以及发送<code>log</code>记录
        - <code>delete</code> | 删除并举报所有轰炸的用户（速度最快）
        - <code>captcha</code> | 对每个用户进行 <code>CAPTCHA</code> 挑战
        - <code>none</code> | 不进行任何操作

        :param opt action: 处理方式
        :alias: boom_act
        """
        if not action:
            return await self._display_value(
                display_text=lang("flood_act_curr_rule")
                % lang(f"flood_act_set_{setting.get('flood_act', 'delete')}"),
                sub_cmd="flood_act",
                value_type="vocab_action",
            )
        if action not in ("asis", "captcha", "none", "delete"):
            return await self.help("flood_act")
        action == "none" and setting.delete("flood_act") or setting.set(
            "flood_act", action
        )
        await self._edit(lang(f"flood_act_set_{action}"))

    async def custom_rule(self, *rule: Optional[str]):
        """用户自定义过滤规则，规则返回<code>True</code>为白名单，否则继续执行下面的规则
        使用 <code>,{cmd_name} custom_rule -c</code> 可删除规则

        注意事项：
        - 返回<code>True</code>并不代表添加到白名单，只是停止继续执行规则
        - 规则发送错误默认返回<code>False</code>（继续执行规则），并透过<code>log</code>输出错误信息
        - <b>由于此指令能够直接操作账号，因此请在输入他人给与的规则前先亲自确认是否安全</b>

        可用参数：
        - <code>msg</code> | 触发验证的信息
        - <code>text</code> | 触发验证的信息的文本，永远不为<code>None</code>
        - <code>user</code> | 用户
        - <code>me</code> | 机器人用户（自己）
        - global 数值 (例如: <code>curr_captcha</code>, <code>the_order</code> 等)
        - 注意，可以调用 <code>await 函数</code>

        范例：
        <code class="language-python">text == "BYPASS"</code>

        解释：
        当对方发送的文字为“BYPASS”时，不继续执行规则

        :param rule: 规则
        """
        if not rule:
            return await self._display_value(
                key="custom_rule",
                display_text=code(setting.get("custom_rule", lang("none"))),
                sub_cmd="custom_rule",
                value_type="vocab_rule",
            )
        rule = " ".join(rule)
        if rule.startswith("-c"):
            setting.delete("custom_rule")
            return await self._edit(lang("custom_rule_reset"))
        setting.set("custom_rule", rule)
        await self._edit(lang("custom_rule_set") % rule)

    async def collect_logs(self, toggle: Optional[str]):
        """查看或设置是否允许 <code>PMCaptcha</code> 收集验证错误相关信息以帮助改进
        默认为 <code>开启</code>，收集的信息包括被验证者的信息以及未通过验证的信息记录

        :param opt toggle: 开关 (y / n)
        :alias: collect, log
        """
        if not toggle:
            status = lang(
                "enabled" if setting.get("collect_logs", True) else "disabled"
            )
            return await self._display_value(
                display_text=f"{lang('collect_logs_curr_rule') % status}\n{lang('collect_logs_note')}",
                sub_cmd="log",
                value_type="vocab_bool",
            )
        await self._set_toggle("collect_logs", toggle, reverse=True)

    async def change_type(self, _type: Optional[str]):
        """切换验证码类型，默认为 <code>计算验证</code>
        验证码类型如下：
        - <code>math</code> | 计算验证
        - <code>img</code> | 图像辨识验证
        - <code>sticker</code> | 贴纸验证

        <b>注意：如果图像验证不能使用将回退到计算验证</b>

        :param opt _type: 验证码类型
        :alias: type, typ
        """
        if not _type:
            return await self._display_value(
                display_text=lang("type_curr_rule")
                % lang(f'type_captcha_{setting.get("type", "math")}'),
                sub_cmd="typ",
                value_type="type_param_name",
            )
        if _type not in ("img", "math", "sticker"):
            return await self.help("typ")
        _type == "math" and setting.delete("type") or setting.set("type", _type)
        await self._edit(lang("type_set") % lang(f"type_captcha_{_type}"))

    async def show_settings(self):
        """显示目前所有的设置

        :alias: settings, setting
        """
        settings_text = []
        text_none = bold(lang("none"))
        for key, default in (
            ("whitelist", text_none),
            ("blacklist", text_none),
            ("timeout", 300 if setting.get("type") == "img" else 30),
            ("disable_pm", bold(lang("disabled"))),
            ("action", bold(lang("action_ban"))),
            ("report", bold(lang("enabled"))),
            ("premium", bold(lang("premium_set_none"))),
            ("groups_in_common", text_none),
            ("chat_history", -1),
            ("initiative", bold(lang("enabled"))),
            ("silent", bold(lang("disabled"))),
            ("flood", 5),
            ("flood_username", bold(lang("disabled"))),
            ("flood_act", bold(lang("flood_act_set_delete"))),
            ("collect_logs", bold(lang("enabled"))),
            ("type", bold(lang("type_captcha_math"))),
            ("img_captcha", bold(lang("img_captcha_type_func"))),
            ("img_captcha_retry", 3),
            ("custom_rule", text_none),
            ("welcome", text_none),
        ):
            lang_text = lang(f"{key}_curr_rule")
            # Timeout (rule: timeout, value: [multiple])
            if key == "timeout":
                captcha_type = setting.get("type", "math")
                key: str = {
                    "sticker": "sticker_timeout",
                    "img": "img_timeout",
                    "math": "timeout",
                }.get(captcha_type)
                default: int = {"sticker": 30, "img": 300, "math": 30}.get(captcha_type)
            # Disable (rule: disable_pm, val: disable)
            elif key == "disable_pm":
                key = "disable"
            # Chat History (rule: chat_history, val: history_count)
            elif key == "chat_history":
                key = "history_count"
            # Flood (key: flood, val: flood_limit)
            elif key == "flood":
                key = "flood_limit"
            # Image Type (rule: img_captcha_type, val: img_type)
            elif key == "img_captcha":
                key = "img_type"
            # Image Retry Chance (rule: img_captcha_retry, value: img_max_retry)
            elif key == "img_captcha_retry":
                key = "img_max_retry"
            value = setting.get(key, default)
            if isinstance(value, bool):
                value = bold(lang("enabled" if value else "disabled"))
            elif key == "premium":
                value = "\n" + value
            if lang_text.find("%") != -1:
                lang_text = lang_text % value
            else:
                lang_text += f": {bold(str(value))}"
            settings_text.append(lang_text)
        await self._edit(lang("settings_lists") % "\n".join(settings_text))

    # Config Export / Import

    async def export_settings(self):
        """导出目前 <code>PMCaptcha</code> 的设置
        请注意，此导出并不包括：
        - 封禁人数缓存
        - 白名单
        - 等待验证缓存

        :alias: export, export_setting
        """
        config = sqlite[setting.key_name]
        config["version"] = get_version()
        for key in ("pass", "banned", "flooded"):
            if config.get(key):
                del config[key]
        file = BytesIO(json.dumps(config, indent=4).encode())
        file.name = f"{str_timestamp(int(time.time()))}.pmc-settings.json"
        await bot.send_document("me", file)
        await self._edit(lang("export_success"))

    async def import_settings(self):
        """导入 <code>PMCaptcha</code> 的设置，对着设置文件回复即可
        请注意，如果导出和导入的版本不一样可能会因为版本兼容问题
        导致有些设置可能会无法导入，届时将会提示

        :alias: import_setting, import
        """
        if not (self.msg.reply_to_message and self.msg.reply_to_message.document):
            return await self.help("import")
        try:
            # noinspection PyUnresolvedReferences
            config = json.loads(
                (await self.msg.reply_to_message.download(in_memory=True)).getvalue()
            )
        except (json.JSONDecodeError, ValueError):
            return await self._edit(lang("import_failed"))
        if config.get("version") != get_version():
            return await self._edit(lang("import_version_mismatch"))
        del config["version"]
        for key, value in config.items():
            setting.set(key, value)
        await self._edit(lang("import_success"))

    # Image Captcha

    async def change_img_type(self, _type: Optional[str]):
        """切换图像辨识使用接口，默认为 <code>funCaptcha</code>
        目前可用的接口：
        - <code>func</code> (<i>ArkLabs funCaptcha</i> )
        - <code>github</code> (<i>GitHub 螺旋星系</i> )
        - <code>rec</code> (<i>Google reCAPTCHA</i> )

        请注意， <code>reCAPTCHA</code> 难度相比前两个<a href="https://t.me/c/1441461877/958395">高出不少</a>，
        因此验证码系统会在尝试过多后提供 <code>funCaptcha</code> 接口让用户选择

        :param opt _type: 验证码类型
        :alias: img_type, img_typ
        """
        if not _type:
            return await self._display_value(
                display_text=lang("type_curr_rule")
                % lang(f'img_captcha_type_{setting.get("img_type", "func")}'),
                sub_cmd="img_typ",
                value_type="type_param_name",
            )
        if _type not in ("func", "github", "rec"):
            return await self.help("img_typ")
        _type == "func" and setting.delete("img_type") or setting.set("img_type", _type)
        await self._edit(lang("type_set") % lang(f"img_captcha_type_{_type}"))

    async def img_retry_chance(self, number: Optional[int]):
        """图形验证码最大可重试次数，默认为 <code>3</code>

        :param opt number: 重试次数
        :alias: img_re
        """
        if number is None:
            return await self._display_value(
                display_text=lang("img_captcha_retry_curr_rule")
                % setting.get("img_max_retry", 3),
                sub_cmd="img_re",
                value_type="vocab_int",
            )
        if number < 0:
            return await self._edit(lang("invalid_param"))
        setting.set("img_max_retry", number)
        await self._edit(lang("img_captcha_retry_set") % number)

    # Web Configure (Sam: I'm not touching this)

    async def web_configure(self, config: Optional[str]):
        """PMCaptcha 网页可视化配置

        :alias: web
        """
        if not config:
            try:
                config = sqlite[setting.key_name]
            except KeyError:
                config = {}
            config["version"] = get_version()
            config["cmd"] = user_cmd_name
            for key in ("pass", "banned", "flooded"):
                if config.get(key):
                    del config[key]
            config = b64encode(json.dumps(config).encode("utf-8")).decode("utf-8")
            await self._edit(f"网页配置链接： https://pmc-config.xtaolabs.com/{config}")
            return
        try:
            nc = json.loads(b64decode(config))
        except Exception:
            await self._edit(lang("import_failed"))
            return
        for i in nc:
            if nc[i] == -1:
                setting.delete(i)
            else:
                setting.set(i, nc[i])
        await self._edit(lang("import_success"))


# region Captcha


@dataclass
class TheOrder:
    """Worker of blocking user (Punishment)"""

    queue = asyncio.Queue()
    task: Optional[asyncio.Task] = None

    def __post_init__(self):
        if pending := setting.pending_ban_list.get_subs():
            console.debug(f"Pending user(s) to ban: {len(pending)}")
            if len(pending) > 0 and not self.task or self.task.done():
                self.task = asyncio.create_task(self.worker())
            for user_id in pending:
                self.queue.put_nowait((user_id, False))

    async def worker(self):
        console.debug("Punishment Worker started")
        while True:
            target = None
            try:
                target, skip_log = await self.queue.get()
                action = setting.get("action", "ban")
                if action in ("ban", "delete"):
                    if not await exec_api(bot.block_user(user_id=target)):
                        console.debug(f"Failed to block user {target}")
                    if action == "delete" and not await exec_api(
                        bot.invoke(
                            messages.DeleteHistory(
                                just_clear=False,
                                revoke=False,
                                peer=await bot.resolve_peer(target),
                                max_id=0,
                            )
                        )
                    ):
                        console.debug(f"Failed to delete user chat {target}")
                setting.pending_ban_list.del_id(target)
                setting.get_challenge_state(target) and setting.del_challenge_state(
                    target
                )
                setting.set("banned", setting.get("banned", 0) + 1)
                chat_link = gen_link(str(target), f"tg://user?id={target}")
                text = f"[PMCaptcha - The Order] {lang('verify_log_punished')} (Punishment)"
                (
                    not skip_log
                    and action not in ("none", "archive")
                    and await log(text % (chat_link, lang(f"action_{action}")), True)
                )
                (
                    skip_log
                    and console.debug(
                        text
                        % (
                            chat_link,
                            lang(f'action_{action == "none" and "set_none" or action}'),
                        )
                    )
                )
            except asyncio.CancelledError:
                break
            except Exception as e:
                await log(
                    f"Error occurred when punishing user: {e}\n{traceback.format_exc()}"
                )
            finally:
                target and self.queue.task_done()

    async def active(self, user_id: int, reason_code: str):
        if not self.task or self.task.done():
            self.task = asyncio.create_task(self.worker())
        try:
            not the_world_eye.triggered and not setting.get(
                "silent"
            ) and await bot.send_message(
                user_id,
                "\n\n".join((lang_full(reason_code), lang_full("verify_blocked"))),
            )
        except FloodWait:
            pass  # Skip waiting
        finally:
            setting.pending_ban_list.add_id(user_id)
            await self.queue.put((user_id, False))
            console.debug(f"User {user_id} added to ban queue")


@dataclass
class TheWorldEye:
    """Anti-Flooding System

    Actual name of each functions:
    - sophitia -> Watcher
    - synchronize -> flood_triggered
    - overload -> flood_ended

    Naming inspired by a anime game
    """

    queue = asyncio.Queue()
    watcher: Optional[asyncio.Task] = None
    timer_task: Optional[asyncio.Task] = None

    # Watcher
    last_challenge_time: Optional[int] = None
    level: int = 0

    # Post Init Value
    channel_id: Optional[int] = None
    username: Optional[str] = None
    triggered: bool = False
    start: Optional[int] = None
    update: Optional[int] = None
    end: Optional[int] = None
    user_ids: Optional[list] = field(init=False)
    auto_archive_enabled_default: Optional[bool] = None

    def __post_init__(self):
        self.user_ids = []
        if state := setting.get_flood_state():  # PMCaptcha restarts, flood keeps going
            # Resume last flood state
            now = int(time.time())
            self.triggered = True
            self.channel_id = state.get("channel_id")
            self.username = state.get("username")
            self.start = state.get("start")
            self.update = state.get("update")
            self.user_ids = state.get("user_ids")
            self.auto_archive_enabled_default = state.get(
                "auto_archive_enabled_default"
            )
            self.reset_timer(360 - (now - self.start))
            console.debug("PMCaptcha restarted, flood state resume")
        self.watcher = asyncio.create_task(self.sophitia())

    # region Timer

    async def _flood_timer(self, interval: int):
        try:
            await asyncio.sleep(interval)
        except asyncio.CancelledError:
            return
        console.debug("Flood timer ended")
        self.end = int(time.time())
        await self.overload()

    def reset_timer(self, interval: int = 270 + randint(30, 60)):
        if self.timer_task:
            self.timer_task.cancel()
        self.update = int(time.time())
        self.timer_task = asyncio.create_task(self._flood_timer(interval))
        console.debug("Flood timer reset")
        return self

    # endregion

    async def _set_channel_username(self):
        console.debug("Creating temporary channel")
        try:
            channel = await bot.create_channel(
                "PMCaptcha Temporary Channel",
                "\n\n".join(
                    (lang("flood_channel_desc", "en"), lang("flood_channel_desc", "zh"))
                ),
            )
            console.debug("Temporary channel created")
            self.channel_id = channel.id
        except Exception as e:
            await log(
                f"Failed to create temporary channel: {e}\n{traceback.format_exc()}"
            )
            return False
        console.debug("Moving username to temporary channel")
        try:
            await bot.set_username(None)
        except Exception as e:
            await log(f"Failed to remove username: {e}\n{traceback.format_exc()}")
            return False
        result = False
        try:
            await bot.invoke(
                UpdateUsername(
                    channel=await bot.resolve_peer(channel.id), username=self.username
                )
            )
            result = True
        except ChannelsAdminPublicTooMuch:
            await log(
                "Failed to move username to temporary channel, too many public channels"
            )
        except Exception as e:
            await log(
                f"Failed to set username for channel: {e}\n{traceback.format_exc()}"
            )
        if not result:
            console.debug("Setting back username")
            try:
                await bot.set_username(self.username)
                await bot.delete_supergroup(channel.id)
            except Exception as e:
                await log(f"Failed to set username back: {e}\n{traceback.format_exc()}")
            self.username = None
        return result

    async def _restore_username(self):
        if self.channel_id:
            console.debug("Deleting temporary channel")
            try:
                await bot.delete_channel(self.channel_id)
            except Exception as e:
                console.debug(
                    f"Failed to delete temporary channel: {e}\n{traceback.format_exc()}"
                )
        if self.username:
            console.debug("Setting back username")
            try:
                await bot.set_username(self.username)
            except Exception as e:
                await log(f"Failed to set username back: {e}\n{traceback.format_exc()}")
        self.username = self.channel_id = None

    # region State

    def save_state(self):
        setting.set_flood_state(
            {
                "start": self.start,
                "update": self.update,
                "user_ids": self.user_ids,
                "auto_archive_enabled_default": self.auto_archive_enabled_default,
                "username": self.username,
                "channel_id": self.channel_id,
            }
        )

    def update_state(self):
        data = setting.get_flood_state()
        data.update(
            {
                "update": self.update,
                "user_ids": self.user_ids,
            }
        )
        setting.set_flood_state(data)

    @staticmethod
    def del_state():
        setting.del_flood_state()

    # endregion

    # noinspection SpellCheckingInspection
    async def sophitia(self):
        """Watches the private message chat (World)"""
        console.debug("Flood Watcher started")
        while True:
            user_id = None
            try:
                (user_id,) = await self.queue.get()
                if self.triggered:  # Continues flooding, add to list and reset timer
                    self.reset_timer()
                    self.user_ids.append(user_id)
                    console.debug(f"User {user_id} added to flood list")
                    self.update_state()
                    continue
                now = int(time.time())
                if self.last_challenge_time and now - self.last_challenge_time < 60:
                    # A user is challenged less than a min
                    self.level += 1
                elif (
                    not self.last_challenge_time or now - self.last_challenge_time > 60
                ):
                    self.level = 1
                self.last_challenge_time = now
                if self.level >= setting.get("flood_limit", 5):
                    console.warn(f"Flooding detected: {self.level} reached in 1 min")
                    self.triggered = True
                    self.start = self.update = now
                    self.reset_timer()
                    await self.synchronize()
            except asyncio.CancelledError:
                break
            except Exception as e:
                await log(
                    f"Error occurred in flood watcher: {e}\n{traceback.format_exc()}"
                )
            finally:
                user_id and self.queue.task_done()

    async def add_synchronize(self, user_id: int):
        await self.queue.put((user_id,))

    async def synchronize(self):
        """Triggered when flood starts (Iris has started synchronizing people)"""
        # Force enable auto archive to reduce api flood
        settings: GlobalPrivacySettings = await bot.invoke(GetGlobalPrivacySettings())
        self.auto_archive_enabled_default = (
            settings.archive_and_mute_new_noncontact_peers
        )
        console.debug("Enabling auto archive")
        try:
            await bot.invoke(
                SetGlobalPrivacySettings(
                    settings=GlobalPrivacySettings(
                        archive_and_mute_new_noncontact_peers=True
                    )
                )
            )
            console.debug("Auto archive enabled")
        except AutoarchiveNotAvailable:
            console.warn(
                "Auto archive is not available, API Flooding may be larger than expected"
            )
        except Exception as e:
            console.error(
                f"Failed to enable auto archive: {e}\n{traceback.format_exc()}"
            )
        if setting.get("flood_username") and bot.me.username:
            self.username = bot.me.username
            console.debug("Moving username to temporary channel")
            if not await self._set_channel_username():
                self.username = None
        # Save state
        self.save_state()

    async def overload(self):
        """Executed after flood ends (Nine has performed load action)"""
        console.info(
            f"Flood ended, {len(self.user_ids)} users were affected, duration: {self.end - self.start}s"
        )
        flood_act = setting.get("flood_act", "delete")
        user_ids, start, end, duration = (
            self.user_ids.copy(),
            self.start,
            self.end,
            self.update - self.start,
        )
        # Reset now so it can handle next flood in time
        rule_lock.locked() and rule_lock.release()
        await rule_lock.acquire()  # Don't process rule until flood state is reset
        self.triggered = False
        self.user_ids.clear()
        self.del_state()
        self.start = self.end = self.update = None
        if self.channel_id or self.username:
            console.debug("Changing back username")
            await self._restore_username()
        try:
            await bot.send_message(
                log_collect_bot,
                "\n".join(
                    (
                        "💣 检测到私聊轰炸",
                        f"设置限制: {code(setting.get('flood_limit', 5))}",
                        f"用户数量: {code(str(len(user_ids)))}",
                        f"开始时间: {code(str_timestamp(start))}",
                        f"结束时间: {code(str_timestamp(end))}",
                        f"轰炸时长: {code(str(duration))} 秒",
                    )
                ),
            )
        except Exception as e:
            console.debug(f"Failed to send flood log: {e}\n{traceback.format_exc()}")
        if not self.auto_archive_enabled_default:  # Restore auto archive setting
            try:
                await bot.invoke(
                    SetGlobalPrivacySettings(
                        settings=GlobalPrivacySettings(
                            archive_and_mute_new_noncontact_peers=False
                        )
                    )
                )
                console.debug("Auto archive disabled")
            except Exception as e:
                console.debug(
                    f"Failed to disable auto archive: {e}\n{traceback.format_exc()}"
                )
        rule_lock.release()  # Let rule processing continue
        setting.set("banned", setting.get("banned", 0) + len(user_ids))
        setting.set("flooded", setting.get("flooded", 0) + 1)
        console.debug(f"Doing flood action: {flood_act}")
        if flood_act == "asis":
            if not the_order.task or the_order.task.done():
                the_order.task = asyncio.create_task(the_order.worker())
            for user_id in user_ids:
                setting.pending_ban_list.add_id(user_id)
            for user_id in user_ids:
                await the_order.queue.put((user_id, True))
        elif flood_act == "captcha":
            if not captcha_task.task or captcha_task.task.done():
                captcha_task.task = asyncio.create_task(captcha_task.worker())
            for user_id in user_ids:
                setting.pending_challenge_list.add_id(user_id)
            for user_id in user_ids:
                if curr_captcha.get(user_id) or setting.get_challenge_state(user_id):
                    continue
                await self.queue.put((user_id, None, None, None))
                console.debug(f"User {user_id} added to challenge queue")
        elif flood_act == "delete":
            from pyrogram.raw.functions import messages

            console.debug(f"Delete and reporting {len(user_ids)} users")
            for user_id in user_ids:
                peer = await bot.resolve_peer(user_id)
                await exec_api(bot.invoke(messages.ReportSpam(peer=peer)))
                await exec_api(
                    bot.invoke(
                        messages.DeleteHistory(
                            just_clear=False, revoke=False, peer=peer, max_id=0
                        )
                    )
                )
        console.debug("Flood action done")


@dataclass
class CaptchaTask:
    """A class to start, resume and verify the captcha challenge
    and contains some nice function like archiving user, getting user's settings

    The main function of this class is to queue & start a captcha for the user
    """

    queue = asyncio.Queue()
    task: Optional[asyncio.Task] = None

    def __post_init__(self):
        if pending := setting.pending_challenge_list.get_subs():
            console.debug(f"Pending user(s) to challenge: {len(pending)}")
            if len(pending) > 0 and not self.task or self.task.done():
                self.task = asyncio.create_task(self.worker())
            for user_id in pending:
                self.queue.put_nowait((user_id, None, None, None))

    @staticmethod
    async def archive(user_id: int, *, un_archive: bool = False):
        from pyrogram.raw.functions.account import UpdateNotifySettings
        from pyrogram.raw.types import InputNotifyPeer, InputPeerNotifySettings

        notify_setting = InputPeerNotifySettings(
            **{
                "mute_until": None if un_archive else 2147483647,
                "show_previews": un_archive,
                "silent": un_archive,
            }
        )
        peer = InputNotifyPeer(peer=await bot.resolve_peer(user_id))
        await exec_api(
            bot.invoke(UpdateNotifySettings(peer=peer, settings=notify_setting))
        )
        await exec_api(
            (bot.unarchive_chats if un_archive else bot.archive_chats)(user_id)
        )

    @staticmethod
    async def get_user_settings(user_id: int) -> (bool, bool):
        can_report = True
        auto_archived = False
        if peer_settings := await exec_api(
            bot.invoke(messages.GetPeerSettings(peer=await bot.resolve_peer(user_id)))
        ):
            can_report = peer_settings.settings.report_spam
            auto_archived = peer_settings.settings.autoarchived
        else:
            console.debug(f"GetPeerSettings failed for {user_id}")
        return can_report, auto_archived

    async def worker(self):
        console.debug("Captcha Challenge Worker started")
        while True:
            user_id: Optional[int] = None
            try:
                user_id, msg, can_report, auto_archived = await self.queue.get()
                user = msg and msg.from_user or await bot.get_users(user_id)
                if can_report is None or auto_archived is None:
                    can_report, auto_archived = await self.get_user_settings(user_id)
                if (
                    last_captcha := setting.get_challenge_state(user_id)
                ) and not curr_captcha.get(user_id):
                    # Resume last captcha challenge
                    if last_captcha["type"] not in captcha_challenges:
                        console.info(
                            "Failed to resume last captcha challenge: "
                            f"Unknown challenge type {last_captcha['type']}"
                        )
                        continue
                    await captcha_challenges[last_captcha["type"]].resume(
                        user=user, msg=msg, state=last_captcha
                    )
                    continue
                # Start a captcha challenge
                await self.archive(user_id)
                captcha = captcha_challenges.get(
                    setting.get("type", "math"), MathChallenge
                )(user, can_report)
                captcha.log_msg(msg and (msg.text or msg.caption or "") or None)
                captcha = await captcha.start() or captcha
                curr_captcha[user_id] = captcha
                setting.pending_challenge_list.del_id(user_id)
            except asyncio.CancelledError:
                break
            except Exception as e:
                await log(
                    f"Error occurred when challenging user: {e}\n{traceback.format_exc()}"
                )
            finally:
                user_id and self.queue.task_done()

    async def add(
        self,
        user_id: int,
        msg: Optional[Message],
        can_report: Optional[bool],
        auto_archived: Optional[bool],
    ):
        await the_world_eye.add_synchronize(user_id)
        if not self.task or self.task.done():
            self.task = asyncio.create_task(self.worker())
        if not (
            setting.pending_challenge_list.check_id(user_id)
            or curr_captcha.get(user_id)
            or setting.get_challenge_state(user_id)
        ):
            setting.pending_challenge_list.add_id(user_id)
            self.queue.put_nowait((user_id, msg, can_report, auto_archived))
            console.debug(f"User {user_id} added to challenge queue")


@dataclass
class CaptchaChallenge:
    type: str
    user: User
    input: bool
    logs: List[str] = field(init=False, default_factory=list)
    captcha_write_lock: asyncio.Lock = field(init=False, default_factory=asyncio.Lock)

    # User Settings
    can_report: bool = True

    # Post Init Value
    captcha_start: int = 0
    captcha_end: int = 0
    challenge_msg_ids: Optional[List[int]] = field(default_factory=list)
    timer_task: Optional[asyncio.Task] = None

    # region Logging

    def log_msg(self, msg: Optional[str]):
        if isinstance(msg, str) and not msg.strip():
            return
        self.logs.append(isinstance(msg, str) and msg.strip() or msg)

    async def send_log(self, ban_code: Optional[str] = None):
        if not setting.get("collect_logs", True):
            return
        user = self.user
        log_file = BytesIO(json.dumps(self.logs, indent=4).encode())
        log_file.name = f"{user.id}_{self.captcha_start}.json"
        caption = [
            f"FROM: {code(str(bot.me.id))}",
            f"UID: {code(str(user.id))}"
            + (f" @{user.username}" if self.user.username else ""),
            f"Mention: {gen_link(str(user.id), f'tg://user?id={user.id}')}",
        ]
        if user.first_name or user.last_name:
            user_full_name = []
            user.first_name and user_full_name.append(user.first_name)
            user.last_name and user_full_name.append(user.last_name)
            caption.append(f"Name: {code(' '.join(user_full_name))}")
        elif user.is_deleted:
            caption.append(f"Name: {bold('Deleted Account')}")
        if user.is_scam or user.is_fake or user.is_premium:
            tags = []
            user.is_scam and tags.append(code("Scam"))
            user.is_fake and tags.append(code("Fake"))
            user.is_premium and tags.append(code("Premium"))
            caption.append(f"Tags: {', '.join(tags)}")
        user.language_code and caption.append(f"Language: {code(user.language_code)}")
        user.dc_id and caption.append(f"DC: {code(str(user.dc_id))}")
        user.phone_number and caption.append(f"Phone: {code(user.phone_number)}")
        self.type and caption.append(f"Type: {code(self.type)}")
        self.can_report and setting.get("report", True) and caption.append(
            f"Spam Reported: {code('Yes')}"
        )
        ban_code and caption.append(f"Block Reason: {code(ban_code)}")
        self.captcha_start and caption.append(f"Start: {code(str(self.captcha_start))}")
        self.captcha_end and caption.append(f"End: {code(str(self.captcha_end))}")
        (
            self.captcha_start
            and self.captcha_end
            and caption.append(
                f"Duration: {code(str(self.captcha_end - self.captcha_start))}s"
            )
        )
        await exec_api(bot.archive_chats(log_collect_bot))
        await exec_api(bot.unblock_user(log_collect_bot))
        if not await exec_api(
            bot.send_document(
                log_collect_bot,
                log_file,
                caption="\n".join(caption),
                parse_mode=ParseMode.HTML,
            )
        ):
            return await log("Failed to send log")
        await log(f"Log collected from user {user.id}")

    # endregion

    # region State

    def save_state(self, extra: Optional[dict] = None):
        self.captcha_start = self.captcha_start or int(time.time())
        data = {
            "type": self.type,
            "start": self.captcha_start,
            "logs": self.logs,
            "msg_ids": self.challenge_msg_ids,
            "report": self.can_report,
        }
        extra and data.update(extra)
        setting.set_challenge_state(self.user.id, data)

    def update_state(self, changes: Optional[dict] = None):
        data = setting.get_challenge_state(self.user.id)
        changes and data.update(changes)
        setting.set_challenge_state(self.user.id, data)

    def del_state(self):
        setting.del_challenge_state(self.user.id)

    # endregion

    # region Verify Result

    async def _del_challenge_msgs(self):
        for challenge_msg_id in self.challenge_msg_ids:
            try:
                await bot.delete_messages(self.user.id, challenge_msg_id)
            except Exception as e:
                console.error(
                    f"Failed to delete challenge message: {e}\n{traceback.format_exc()}"
                )

    async def _verify_success(self, _):
        setting.whitelist.add_id(self.user.id)
        setting.set("pass", setting.get("pass", 0) + 1)
        chat_link = gen_link(str(self.user.id), f"tg://user?id={self.user.id}")
        await log(
            lang("verify_log_passed") % (chat_link, lang(f"type_captcha_{self.type}"))
        )
        success_msg = setting.get("welcome") or lang_full("verify_passed")
        welcome_msg: Optional[Message] = None
        if setting.get("silent"):
            await self._del_challenge_msgs()
            return await CaptchaTask.archive(self.user.id, un_archive=True)
        try:
            self.challenge_msg_ids and await bot.edit_message_text(
                self.user.id, self.challenge_msg_ids[0], success_msg
            )
        except Exception as e:
            console.error(
                f"Failed to edit welcome message: {e}\n{traceback.format_exc()}"
            )
            try:
                welcome_msg = await bot.send_message(self.user.id, success_msg)
            except Exception as e:
                console.error(
                    f"Failed to send welcome message: {e}\n{traceback.format_exc()}"
                )
        await asyncio.sleep(setting.get("welcome") and 5 or 3)
        await self._del_challenge_msgs()
        welcome_msg and await welcome_msg.delete()
        await CaptchaTask.archive(self.user.id, un_archive=True)

    async def _verify_failed(self, reason_code: str):
        try:
            for challenge_msg_id in self.challenge_msg_ids:
                await bot.delete_messages(self.user.id, challenge_msg_id)
            (
                self.can_report
                and setting.get("report", True)
                and await bot.invoke(
                    messages.ReportSpam(peer=await bot.resolve_peer(self.user.id))
                )
            )
        except Exception as e:
            console.debug(
                f"Error occurred when executing verify failed function: {e}\n{traceback.format_exc()}"
            )
        await the_order.active(self.user.id, "verify_failed")
        await self.send_log(reason_code)

    async def action(self, success: bool, reason_code="verify_failed"):
        self.captcha_end = int(time.time())
        self.del_state()
        self.remove_timer()
        await getattr(self, f"_verify_{'success' if success else 'failed'}")(
            reason_code
        )
        console.debug(
            f"User {self.user.id} verify {'success' if success else 'failed'}"
        )

    # endregion

    # region Timer

    async def _challenge_timer(self, timeout: int):
        try:
            await asyncio.sleep(timeout)
        except asyncio.CancelledError:
            return
        except Exception as e:
            console.error(
                f"Error occurred when running challenge timer: {e}\n{traceback.format_exc()}"
            )
        async with self.captcha_write_lock:
            console.debug(f"User {self.user.id} verification timed out")
            await self.action(False, "verify_timeout")
        if curr_captcha.get(self.user.id):
            del curr_captcha[self.user.id]

    def reset_timer(self, timeout: Optional[int] = None):
        self.timer_task and self.timer_task.cancel()
        timeout = (
            timeout is not None
            and timeout
            or setting.get(
                f"{self.type == 'img' and 'img_' or ''}timeout",
                self.type == "img" and 300 or 30,
            )
        )
        if timeout > 0:
            self.timer_task = asyncio.create_task(self._challenge_timer(timeout))
        console.debug(f"User {self.user.id} verification timer reset")
        return self

    def remove_timer(self):
        if task := self.timer_task:
            task.cancel()
        return self

    # endregion

    @classmethod
    async def resume(cls, *, user: User, msg: Optional[Message] = None, state: dict):
        console.debug(f"User {user.id} resumed captcha challenge {state['type']}")

    async def start(self):
        console.debug(f"User {self.user.id} started {self.type} captcha challenge")


class MathChallenge(CaptchaChallenge):
    answer: int

    def __init__(self, user: User, can_report: bool):
        super().__init__("math", user, True, can_report)

    @classmethod
    async def resume(cls, *, user: User, msg: Optional[Message] = None, state: dict):
        captcha = cls(user, state.get("report", True))
        captcha.captcha_start = state["start"]
        captcha.logs = state["logs"]
        captcha.challenge_msg_ids = (
            [state["msg_id"]] if state.get("msg_id") else state.get("msg_ids", [])
        )

        captcha.answer = state["answer"]
        if (timeout := setting.get("timeout", 30)) > 0:
            time_passed = int(time.time()) - int(state["start"])
            if time_passed > timeout:
                # Timeout
                return await captcha.action(False)
            if msg:  # Verify result
                await captcha.verify(msg.text or msg.caption or "")
            else:  # Restore timer
                captcha.reset_timer(timeout - time_passed)
        await super(MathChallenge, captcha).resume(user=user, msg=msg, state=state)

    async def start(self, previous_msg_id: Optional[int] = None):
        if self.captcha_write_lock.locked():
            return
        async with self.captcha_write_lock:
            import random

            first_value, second_value = random.randint(1, 10), random.randint(1, 10)
            timeout = setting.get("timeout", 30)
            operator = random.choice(("+", "-", "*"))
            expression = f"{first_value} {operator} {second_value}"
            params = {
                "chat_id": self.user.id,
                "text": "\n".join(
                    (
                        lang_full("verify_challenge"),
                        "",
                        code(f"{expression} = ?"),
                        "",
                        lang_full(
                            "verify_challenge_timed", timeout if timeout > 0 else ""
                        ),
                    )
                ),
                "parse_mode": ParseMode.HTML,
            }
            if previous_msg_id:
                params["message_id"] = previous_msg_id
            challenge_msg = await exec_api(
                (bot.edit_message_text if previous_msg_id else bot.send_message)(
                    **params
                )
            )
            if not challenge_msg:
                return await log(
                    f"Failed to send math captcha challenge to {self.user.id}"
                )
            self.challenge_msg_ids = [challenge_msg.id]
            self.answer = eval(expression)
            self.save_state({"answer": self.answer})
            self.reset_timer(timeout)
            await super(MathChallenge, self).start()

    async def verify(self, answer: str):
        if self.captcha_write_lock.locked():
            return
        async with self.captcha_write_lock:
            try:
                user_answer = int("".join(re.findall(r"\d+", answer)))
                if "-" in answer:
                    user_answer = -user_answer
            except ValueError:
                return await the_order.active(self.user.id, "verify_failed")
        await self.action(user_answer == self.answer)
        return user_answer == self.answer


class ImageChallenge(CaptchaChallenge):
    try_count: int

    def __init__(self, user: User, can_report: bool):
        super().__init__("img", user, False, can_report)
        self.try_count = 0

    @classmethod
    async def resume(cls, *, user: User, msg: Optional[Message] = None, state: dict):
        captcha = cls(user, state["report"])
        captcha.captcha_start = state["start"]
        captcha.logs = state["logs"]
        captcha.challenge_msg_ids = (
            [state["msg_id"]] if state.get("msg_id") else state.get("msg_ids", [])
        )

        captcha.try_count = state["try_count"]
        if captcha.try_count >= setting.get("img_max_retry", 3):
            return await captcha.action(False)
        if (timeout := setting.get("img_timeout", 300)) > 0:  # Restore timer
            time_passed = int(time.time()) - int(state["last_active"])
            if time_passed > timeout:
                # Timeout
                return await captcha.action(False)
            captcha.reset_timer(timeout - time_passed)
        curr_captcha[user.id] = captcha
        await super(ImageChallenge, captcha).resume(user=user, msg=msg, state=state)

    async def start(self):
        from pyrogram.raw.types import UpdateMessageID

        if self.captcha_write_lock.locked():
            return
        async with self.captcha_write_lock:
            while True:
                try:
                    if (
                        not (
                            result := await bot.get_inline_bot_results(
                                img_captcha_bot, setting.get("img_type", "func")
                            )
                        )
                        or not result.results
                    ):
                        console.debug(
                            f"Failed to get captcha results from {img_captcha_bot}, fallback"
                        )
                        break  # Fallback
                    # From now on, wait for bot result
                    timeout = setting.get("timeout", 300)
                    challenge_msg = await exec_api(
                        bot.send_message(
                            self.user.id,
                            "\n".join(
                                (
                                    lang_full("verify_challenge"),
                                    "",
                                    code(lang_full("verify_complete_image")),
                                    "",
                                    lang_full(
                                        "verify_challenge_timed",
                                        timeout if timeout > 0 else "",
                                    ),
                                )
                            ),
                            parse_mode=ParseMode.HTML,
                        )
                    )
                    updates = await bot.send_inline_bot_result(
                        self.user.id, result.query_id, result.results[0].id
                    )
                    for update in updates.updates:
                        if isinstance(update, UpdateMessageID):
                            self.challenge_msg_ids = [challenge_msg.id, update.id]
                            self.save_state(
                                {
                                    "try_count": self.try_count,
                                    "last_active": int(time.time()),
                                }
                            )
                            await bot.block_user(self.user.id)
                            self.reset_timer()
                            await super(ImageChallenge, self).start()
                            return
                    console.debug(
                        f"Failed to send image captcha challenge to {self.user.id}, fallback"
                    )
                    challenge_msg and await challenge_msg.safe_delete()
                    break
                except (TimeoutError, BotResponseTimeout):
                    console.debug(
                        f"Image captcha bot timeout for {self.user.id}, fallback"
                    )
                    break  # Fallback
                except FloodWait as e:
                    console.debug(
                        f"Image captcha triggered flood for {self.user.id}, wait {e.value}"
                    )
                    await asyncio.sleep(e.value)
                except Exception as e:
                    console.error(
                        f"Failed to send image captcha challenge to {self.user.id}: {e}\n{traceback.format_exc()}"
                    )
                    await asyncio.sleep(10)
            console.debug("Failed to get image captcha, fallback to math captcha.")
            fallback_captcha = MathChallenge(self.user, self.can_report)
            await fallback_captcha.start()
            return fallback_captcha

    async def verify(self, success: bool):
        if success:
            await bot.unblock_user(self.user.id)
            self.challenge_msg_id = 0
            return await self.action(success)
        else:
            self.try_count += 1
            if self.try_count >= setting.get("img_max_retry", 3):
                await self.action(False)
                return True
            console.debug(
                f"User failed to complete image captcha challenge, try count: {self.try_count}"
            )
            self.update_state({"try_count": self.try_count})


class StickerChallenge(CaptchaChallenge):
    def __init__(self, user: User, can_report: bool):
        super().__init__("sticker", user, True, can_report)

    @classmethod
    async def resume(cls, *, user: User, msg: Optional[Message] = None, state: dict):
        captcha = cls(user, state["report"])
        captcha.captcha_start = state["start"]
        captcha.logs = state["logs"]
        captcha.challenge_msg_ids = (
            [state["msg_id"]] if state.get("msg_id") else state.get("msg_ids", [])
        )

        if (timeout := setting.get("timeout", 30)) > 0:
            time_passed = int(time.time()) - int(state["start"])
            if time_passed > timeout:
                # Timeout
                return await captcha.action(False)
            if msg:  # Verify result
                await captcha.verify(msg.text or msg.caption or "")
            else:  # Restore timer
                captcha.reset_timer(timeout - time_passed)
        await super(StickerChallenge, captcha).resume(user=user, msg=msg, state=state)

    async def start(self):
        if self.captcha_write_lock.locked():
            return
        async with self.captcha_write_lock:
            timeout = setting.get("timeout", 30)
            challenge_msg = await exec_api(
                bot.send_message(
                    self.user.id,
                    "\n".join(
                        (
                            lang_full("verify_challenge"),
                            "",
                            code(lang_full("verify_send_sticker")),
                            "",
                            lang_full(
                                "verify_challenge_timed", timeout if timeout > 0 else ""
                            ),
                        )
                    ),
                    parse_mode=ParseMode.HTML,
                )
            )
            if not challenge_msg:
                return await log(
                    f"Failed to send sticker captcha challenge to {self.user.id}"
                )
            self.challenge_msg_ids = [challenge_msg.id]
            self.save_state()
            self.reset_timer(timeout)
            await super(StickerChallenge, self).start()

    async def verify(self, response: Optional[Sticker]):
        if self.captcha_write_lock.locked():
            return
        async with self.captcha_write_lock:
            if not response:
                return await the_order.active(self.user.id, "verify_failed")
        await self.action(bool(response))
        return bool(response)


# endregion


@dataclass
class Rule:
    user: User
    msg: Message

    can_report: Optional[bool] = None
    auto_archived: Optional[bool] = None

    def _precondition(self) -> bool:
        return (
            # Skip for PGM/PMC Developers
            self.user.id in (347437156, 583325201, 1148248480, 751686745, 676660002)
            or self.msg.from_user.is_contact
            or self.msg.from_user.is_verified
            or self.msg.chat.type == ChatType.BOT
            or self.msg.service is not None  # skip service message
            or setting.is_verified(self.user.id)
        )

    def _get_text(self) -> str:
        return self.msg.text or self.msg.caption or ""

    async def _get_user_settings(self) -> (bool, bool):
        if isinstance(self.can_report, bool):
            return self.can_report, self.auto_archived
        return await captcha_task.get_user_settings(self.user.id)

    async def _run_rules(self, *, outgoing: bool = False):
        if not outgoing and self._precondition():
            return
        members = inspect.getmembers(self, inspect.iscoroutinefunction)
        members.sort(key=_sort_line_number)
        async with rule_lock:
            for name, func in members:
                docs = func.__doc__ or ""
                try:
                    if not name.startswith("_") and (
                        "outgoing" in docs
                        and outgoing
                        and await func()
                        or "outgoing" not in docs
                        and not self.user.is_self
                        and await func()
                    ):
                        console.debug(
                            f"Rule triggered: `{name}` (user: {self.user.id} chat: {self.msg.chat.id})"
                        )
                        break
                except Exception as e:
                    console.error(
                        f"Failed to run rule `{name}`: {e}\n{traceback.format_exc()}"
                    )

    @staticmethod
    def _get_rules_priority() -> tuple:
        prio_list = []
        members = inspect.getmembers(Rule, inspect.iscoroutinefunction)
        members.sort(key=_sort_line_number)
        for name, func in members:
            if name.startswith("_"):
                continue
            docs = func.__doc__ or ""
            if "no_prio" not in docs:
                if result := re.search(r"name:\s?(.+)", docs):
                    name = result[1]
                prio_list.append(name)
        return tuple(prio_list)

    async def initiative(self) -> bool:
        """outgoing"""
        initiative = setting.get("initiative", True)
        initiative and not setting.whitelist.check_id(
            self.msg.chat.id
        ) and setting.whitelist.add_id(self.msg.chat.id)
        return initiative

    async def user_defined(self) -> bool:
        if custom_rule := setting.get("custom_rule"):
            try:
                exec(f"async def _(msg, text, user, me, bot):\n return {custom_rule}")
                return bool(
                    await locals()["_"](
                        self.msg, self._get_text(), self.user, bot.me, bot
                    )
                )
            except Exception as e:
                await log(
                    f"{lang('custom_rule_exec_err')}: {e}\n{traceback.format_exc()}"
                )
        return False

    async def disable_pm(self) -> bool:
        if disabled := setting.get("disable"):
            await the_order.active(self.user.id, "disable_pm_enabled")
            captcha = CaptchaChallenge("none", self.user, False)
            captcha.log_msg(self.msg.text or self.msg.caption or "")
            await captcha.send_log("pm_disabled")
        return disabled

    async def chat_history(self) -> bool:
        if (history_count := setting.get("history_count", -1)) > 0:
            count = 0
            async for msg in bot.get_chat_history(
                self.user.id, limit=history_count + 1
            ):
                if msg.id != self.msg.id:
                    count += 1
            if count >= history_count:
                setting.whitelist.add_id(self.user.id)
                return True
        return False

    async def groups_in_common(self) -> bool:
        from pyrogram.raw.functions.users import GetFullUser

        if (common_groups := setting.get("groups_in_common")) is not None:
            if user_full := await exec_api(
                bot.invoke(GetFullUser(id=await bot.resolve_peer(self.user.id)))
            ):
                if user_full.full_user.common_chats_count >= common_groups:
                    setting.whitelist.add_id(self.user.id)
                    return True
            else:
                console.warn(f"Failed to Get Common Groups for user {self.user.id}")
        return False

    async def premium(self) -> bool:
        if premium := setting.get("premium"):
            captcha = CaptchaChallenge("disable_pm", self.user, False)
            captcha.log_msg(self.msg.text or self.msg.caption or "")
            if premium == "only" and not self.msg.from_user.is_premium:
                await the_order.active(self.user.id, "premium_only")
                await captcha.send_log("premium_only")
            elif not self.msg.from_user.is_premium:
                return False
            elif premium == "ban":
                await the_order.active(self.user.id, "premium_ban")
                await captcha.send_log("premium_ban")
        return premium

    # Whitelist / Blacklist
    async def word_filter(self) -> bool:
        """name: whitelist > blacklist"""
        text = self._get_text()
        if text is None:
            return False
        if array := setting.get("whitelist"):
            for word in array:
                if word not in text:
                    continue
                setting.whitelist.add_id(self.user.id)
                return True
        if array := setting.get("blacklist"):
            for word in array:
                if word not in text:
                    continue
                reason_code = "blacklist_triggered"
                await the_order.active(self.user.id, reason_code)
                # Collect logs
                can_report, _ = await self._get_user_settings()
                captcha = CaptchaChallenge("blacklist", self.user, False, can_report)
                captcha.log_msg(text)
                await captcha.send_log(reason_code)
                return True
        return False

    async def flooding(self) -> bool:
        """name: flood"""
        if the_world_eye.triggered:
            _, auto_archived = await self._get_user_settings()
            not auto_archived and await captcha_task.archive(self.user.id)
            await the_world_eye.add_synchronize(self.user.id)
        return the_world_eye.triggered

    async def add_captcha(self) -> bool:
        """name: captcha"""
        user_id = self.user.id
        if (
            setting.get_challenge_state(user_id)
            and not curr_captcha.get(user_id)
            or not curr_captcha.get(user_id)
        ):
            # Put in challenge queue
            await captcha_task.add(
                user_id, self.msg, *(await self._get_user_settings())
            )
            return True
        return False

    async def verify_challenge_answer(self) -> bool:
        """no_priority"""
        user_id = self.user.id
        if (
            (captcha := curr_captcha.get(user_id))
            and captcha.input
            and captcha.type == "math"
        ):
            text = self._get_text()
            captcha.log_msg(text)
            await captcha.verify(text) and await self.msg.safe_delete()
            del curr_captcha[user_id]
            return True
        return False

    async def verify_sticker_response(self) -> bool:
        """no_priority"""
        if (
            (captcha := curr_captcha.get(user_id := self.user.id))
            and captcha.input
            and captcha.type == "sticker"
        ):
            captcha.log_msg(self._get_text())
            await captcha.verify(self.msg.sticker) and await self.msg.safe_delete()
            del curr_captcha[user_id]
            return True
        return False


# Watches every image captcha result
@listener(is_plugin=False, incoming=True, outgoing=True, privates_only=True)
async def image_captcha_listener(_, msg: Message):
    # Ignores non-private chat, not via bot, username not equal to image bot
    if (
        msg.chat.type != ChatType.PRIVATE
        or not msg.via_bot
        or msg.via_bot.username != img_captcha_bot
    ):
        return
    user_id = msg.chat.id
    if (
        last_captcha := sqlite.get(f"pmcaptcha.challenge.{user_id}")
    ) and not curr_captcha.get(user_id):
        # Resume last captcha challenge
        if last_captcha["type"] != "img":
            return await log(
                "Failed to resume last captcha challenge: "
                f"Unknown challenge type {last_captcha['type']}"
            )
        await ImageChallenge.resume(user=msg.from_user, state=last_captcha)
    if not curr_captcha.get(user_id):  # User not in verify state
        return
    captcha, msg_text = curr_captcha[user_id], msg.caption or ""
    captcha.reset_timer().update_state({"last_active": int(time.time())})
    if "CAPTCHA_SOLVED" in msg_text:
        await msg.safe_delete()
        await captcha.verify(True)
        del curr_captcha[user_id]
    elif "CAPTCHA_FAILED" in msg_text:
        if "forced" in msg.caption:
            await captcha.action(False)
            del curr_captcha[user_id]
            return
        if await captcha.verify(False):
            del curr_captcha[user_id]
            await msg.safe_delete()
    elif "CAPTCHA_FALLBACK" in msg_text:
        await msg.safe_delete()
        # Fallback to selected captcha type
        captcha_type = msg.caption.replace("CAPTCHA_FALLBACK", "").strip()
        console.debug(f"Image bot return fallback request, fallback to {captcha_type}")

        # Unstuck
        await bot.unblock_user(user_id)
        if captcha := curr_captcha.get(user_id):
            captcha.timer_task and captcha.timer_task.cancel()
        setting.get_challenge_state(user_id) and setting.del_challenge_state(user_id)

        if captcha_type == "math":
            new_captcha = MathChallenge(captcha.user, captcha.can_report)
            challenge_msg_id = captcha.challenge_msg_ids[0]
            await new_captcha.start(challenge_msg_id)
            curr_captcha[user_id] = new_captcha
            return


@listener(is_plugin=False, outgoing=True, privates_only=True)
async def initiative_listener(_, msg: Message):
    await Rule(bot.me, msg)._run_rules(outgoing=True)


@listener(
    is_plugin=False,
    incoming=True,
    outgoing=False,
    ignore_edited=True,
    privates_only=True,
)
async def chat_listener(_, msg: Message):
    await Rule(msg.from_user, msg)._run_rules()


@listener(
    is_plugin=True,
    outgoing=True,
    command=user_cmd_name,
    parameters=f"<{lang('vocab_cmd')}> [{lang('cmd_param')}]",
    need_admin=True,
    description=f"{lang('plugin_desc')}\n{(lang('check_usage') % code(f',{user_cmd_name} h'))}",
)
async def cmd_entry(_, msg: Message):
    cmd = Command(msg.from_user, msg)
    result, err_code, extra = await cmd._run_command()
    if not result:
        if err_code == "NOT_FOUND":
            return await cmd._edit(
                f"{lang('cmd_not_found')}: {code(extra)}\n"
                + lang("check_usage") % code(f",{user_cmd_name} h")
            )
        elif err_code == "INVALID_PARAM":
            return await cmd._edit(lang("invalid_param"))


async def resume_states():
    console.debug("Resuming Captcha States")
    for key, value in sqlite.items():  # type: str, dict
        if key.startswith("pmcaptcha.challenge"):
            user_id = int(key.split(".")[2])
            if user_id not in curr_captcha and (
                challenge := captcha_challenges.get(value.get("type"))
            ):
                # Resume challenge state
                try:
                    user = await bot.get_users(user_id)
                    await challenge.resume(user=user, state=value)
                except (KeyError, PeerIdInvalid):
                    del sqlite[key]
                    console.debug(f"User {user_id} not found, deleted challenge state")
                except Exception as e:
                    console.error(
                        f"Error occurred when resuming captcha state: {e}\n{traceback.format_exc()}"
                    )
    console.debug("Captcha State Resume Completed")


if __name__ == "plugins.pmcaptcha":
    # Force disabled for old PMCaptcha
    globals().get("SubCommand") and exit(0)
    # Flood Username confirm
    user_want_set_flood_username = None
    # Logger
    console = logs.getChild(cmd_name)
    globals().get("console") is None and exit(0)  # Old version
    # Rule lock
    rule_lock = asyncio.Lock()
    captcha_challenges = {
        "math": MathChallenge,
        "img": ImageChallenge,
        "sticker": StickerChallenge,
    }
    _cancel_task = lambda task: task and task.cancel()  # noqa
    gbl = globals()
    # noinspection PyRedeclaration
    curr_captcha: Dict[
        int, Union[MathChallenge, ImageChallenge, StickerChallenge]
    ] = globals().get("curr_captcha", {})
    if setting := globals().get("setting"):
        del setting
    # noinspection PyRedeclaration
    setting = Setting("pmcaptcha")
    if logging := gbl.get("logging"):
        _cancel_task(logging.task)
        del logging
    # noinspection PyRedeclaration
    logging = Log()
    if the_world_eye := gbl.get("the_world_eye"):
        _cancel_task(the_world_eye.watcher)
        del the_world_eye
    # noinspection PyRedeclaration
    the_world_eye = TheWorldEye()
    if the_order := gbl.get("the_order"):
        _cancel_task(the_order.task)
        del the_order
    # noinspection PyRedeclaration
    the_order = TheOrder()
    if captcha_task := gbl.get("captcha_task"):
        _cancel_task(captcha_task.task)
        del captcha_task
    # noinspection PyRedeclaration
    captcha_task = CaptchaTask()
    if resume_task := gbl.get("resume_task"):
        _cancel_task(resume_task)
    resume_task = asyncio.create_task(resume_states())
    gc.collect()
elif __name__ == "__main__":
    with open("command_list.md", "wb") as f:
        f.write(Command._generate_markdown().encode())
    print("MarkDown Generated.")