async def __call__(self, client: "pyrogram.Client", update: Update):

raise NotImplementedError

def __invert__(self):

return InvertFilter(self)

def __and__(self, other):

return AndFilter(self, other)

def __or__(self, other):

def __init__(self, base):

self.base = base

async def __call__(self, client: "pyrogram.Client", update: Update):

if inspect.iscoroutinefunction(self.base.__call__):

x = await self.base(client, update)

else:

x = await client.loop.run_in_executor(

client.executor,

self.base,

client, update

)

def __init__(self, base, other):

self.base = base

self.other = other

async def __call__(self, client: "pyrogram.Client", update: Update):

if inspect.iscoroutinefunction(self.base.__call__):

x = await self.base(client, update)

else:

x = await client.loop.run_in_executor(

client.executor,

self.base,

client, update

)

# short circuit

if not x:

return False

if inspect.iscoroutinefunction(self.other.__call__):

y = await self.other(client, update)

else:

y = await client.loop.run_in_executor(

client.executor,

self.other,

client, update

)

def __init__(self, base, other):

self.base = base

self.other = other

async def __call__(self, client: "pyrogram.Client", update: Update):

if inspect.iscoroutinefunction(self.base.__call__):

x = await self.base(client, update)

else:

x = await client.loop.run_in_executor(

client.executor,

self.base,

client, update

)

# short circuit

if x:

return True

if inspect.iscoroutinefunction(self.other.__call__):

y = await self.other(client, update)

else:

y = await client.loop.run_in_executor(

client.executor,

self.other,

client, update

)

"""Easily create a custom filter.

return type(

name or func.__name__ or CUSTOM_FILTER_NAME,

(Filter,),

{"__call__": func, **kwargs}

"""Filter commands, i.e.: text messages starting with "/" or any other custom prefix.

command_re = re.compile(r"([\"'])(.*?)(?<!\\)\1|(\S+)")

async def func(flt, client: pyrogram.Client, message: Message):

# Username shared among all commands; used for mention commands, e.g.: /start@username

global username

if username is None:

username = (await client.get_me()).username or ""

text = message.text or message.caption

message.command = None

if not text:

return False

for prefix in flt.prefixes:

if not text.startswith(prefix):

continue

without_prefix = text[len(prefix):]

for cmd in flt.commands:

if not re.match(rf"^(?:{cmd}(?:@?{username})?)(?:\s|$)", without_prefix,

flags=re.IGNORECASE if not flt.case_sensitive else 0):

continue

without_command = re.sub(rf"{cmd}(?:@?{username})?\s?", "", without_prefix, count=1,

flags=re.IGNORECASE if not flt.case_sensitive else 0)

# match.groups are 1-indexed, group(1) is the quote, group(2) is the text

# between the quotes, group(3) is unquoted, whitespace-split text

# Remove the escape character from the arguments

message.command = [cmd] + [

re.sub(r"\\([\"'])", r"\1", m.group(2) or m.group(3) or "")

for m in command_re.finditer(without_command)

]

return True

return False

commands = commands if isinstance(commands, list) else [commands]

commands = {c if case_sensitive else c.lower() for c in commands}

prefixes = [] if prefixes is None else prefixes

prefixes = prefixes if isinstance(prefixes, list) else [prefixes]

prefixes = set(prefixes) if prefixes else {""}

return create(

func,

"CommandFilter",

commands=commands,

prefixes=prefixes,

case_sensitive=case_sensitive

"""Filter updates that match a given regular expression pattern.

async def func(flt, _, update: Update):

if isinstance(update, Message):

value = update.text or update.caption

elif isinstance(update, CallbackQuery):

value = update.data

elif isinstance(update, InlineQuery):

value = update.query

else:

raise ValueError(f"Regex filter doesn't work with {type(update)}")

if value:

update.matches = list(flt.p.finditer(value)) or None

return bool(update.matches)

return create(

func,

"RegexFilter",

p=pattern if isinstance(pattern, Pattern) else re.compile(pattern, flags)

"""Filter messages coming from one or more users.

def __init__(self, users: Union[int, str, List[Union[int, str]]] = None):

users = [] if users is None else users if isinstance(users, list) else [users]

super().__init__(

"me" if u in ["me", "self"]

else u.lower().strip("@") if isinstance(u, str)

else u for u in users

)

async def __call__(self, _, message: Message):

return (message.from_user

and (message.from_user.id in self

or (message.from_user.username

and message.from_user.username.lower() in self)

or ("me" in self

"""Filter messages coming from one or more chats.

def __init__(self, chats: Union[int, str, List[Union[int, str]]] = None):

chats = [] if chats is None else chats if isinstance(chats, list) else [chats]

super().__init__(

"me" if c in ["me", "self"]

else c.lower().strip("@") if isinstance(c, str)

else c for c in chats

)

async def __call__(self, _, message: Message):

return (message.chat

and (message.chat.id in self

or (message.chat.username

and message.chat.username.lower() in self)

or ("me" in self

and message.from_user

and message.from_user.is_self