import inspect
import re
import typing
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, Optional, Union
from babel.support import LazyProxy
from aiogram import types
from aiogram.dispatcher.filters.filters import BoundFilter, Filter
from aiogram.types import CallbackQuery, Message, InlineQuery, Poll
[docs]class Command(Filter):
"""
You can handle commands by using this filter.
If filter is successful processed the :obj:`Command.CommandObj` will be passed to the handler arguments.
By default this filter is registered for messages and edited messages handlers.
"""
def __init__(self, commands: Union[Iterable, str],
prefixes: Union[Iterable, str] = '/',
ignore_case: bool = True,
ignore_mention: bool = False):
"""
Filter can be initialized from filters factory or by simply creating instance of this class.
Examples:
.. code-block:: python
@dp.message_handler(commands=['myCommand'])
@dp.message_handler(Command(['myCommand']))
@dp.message_handler(commands=['myCommand'], commands_prefix='!/')
:param commands: Command or list of commands always without leading slashes (prefix)
:param prefixes: Allowed commands prefix. By default is slash.
If you change the default behavior pass the list of prefixes to this argument.
:param ignore_case: Ignore case of the command
:param ignore_mention: Ignore mention in command
(By default this filter pass only the commands addressed to current bot)
"""
if isinstance(commands, str):
commands = (commands,)
self.commands = list(map(str.lower, commands)) if ignore_case else commands
self.prefixes = prefixes
self.ignore_case = ignore_case
self.ignore_mention = ignore_mention
[docs] @classmethod
def validate(cls, full_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Validator for filters factory
From filters factory this filter can be registered with arguments:
- ``command``
- ``commands_prefix`` (will be passed as ``prefixes``)
- ``commands_ignore_mention`` (will be passed as ``ignore_mention``
:param full_config:
:return: config or empty dict
"""
config = {}
if 'commands' in full_config:
config['commands'] = full_config.pop('commands')
if config and 'commands_prefix' in full_config:
config['prefixes'] = full_config.pop('commands_prefix')
if config and 'commands_ignore_mention' in full_config:
config['ignore_mention'] = full_config.pop('commands_ignore_mention')
return config
[docs] async def check(self, message: types.Message):
return await self.check_command(message, self.commands, self.prefixes, self.ignore_case, self.ignore_mention)
@staticmethod
async def check_command(message: types.Message, commands, prefixes, ignore_case=True, ignore_mention=False):
full_command = message.text.split()[0]
prefix, (command, _, mention) = full_command[0], full_command[1:].partition('@')
if not ignore_mention and mention and (await message.bot.me).username.lower() != mention.lower():
return False
elif prefix not in prefixes:
return False
elif (command.lower() if ignore_case else command) not in commands:
return False
return {'command': Command.CommandObj(command=command, prefix=prefix, mention=mention)}
[docs] @dataclass
class CommandObj:
"""
Instance of this object is always has command and it prefix.
Can be passed as keyword argument ``command`` to the handler
"""
"""Command prefix"""
prefix: str = '/'
"""Command without prefix and mention"""
command: str = ''
"""Mention (if available)"""
mention: str = None
"""Command argument"""
args: str = field(repr=False, default=None)
@property
def mentioned(self) -> bool:
"""
This command has mention?
:return:
"""
return bool(self.mention)
@property
def text(self) -> str:
"""
Generate original text from object
:return:
"""
line = self.prefix + self.command
if self.mentioned:
line += '@' + self.mention
if self.args:
line += ' ' + self.args
return line
[docs]class CommandStart(Command):
"""
This filter based on :obj:`Command` filter but can handle only ``/start`` command.
"""
def __init__(self, deep_link: typing.Optional[typing.Union[str, re.Pattern]] = None):
"""
Also this filter can handle `deep-linking <https://core.telegram.org/bots#deep-linking>`_ arguments.
Example:
.. code-block:: python
@dp.message_handler(CommandStart(re.compile(r'ref-([\\d]+)')))
:param deep_link: string or compiled regular expression (by ``re.compile(...)``).
"""
super(CommandStart, self).__init__(['start'])
self.deep_link = deep_link
[docs] async def check(self, message: types.Message):
"""
If deep-linking is passed to the filter result of the matching will be passed as ``deep_link`` to the handler
:param message:
:return:
"""
check = await super(CommandStart, self).check(message)
if check and self.deep_link is not None:
if not isinstance(self.deep_link, re.Pattern):
return message.get_args() == self.deep_link
match = self.deep_link.match(message.get_args())
if match:
return {'deep_link': match}
return False
return check
[docs]class CommandHelp(Command):
"""
This filter based on :obj:`Command` filter but can handle only ``/help`` command.
"""
def __init__(self):
super(CommandHelp, self).__init__(['help'])
[docs]class CommandSettings(Command):
"""
This filter based on :obj:`Command` filter but can handle only ``/settings`` command.
"""
def __init__(self):
super(CommandSettings, self).__init__(['settings'])
[docs]class CommandPrivacy(Command):
"""
This filter based on :obj:`Command` filter but can handle only ``/privacy`` command.
"""
def __init__(self):
super(CommandPrivacy, self).__init__(['privacy'])
[docs]class Text(Filter):
"""
Simple text filter
"""
def __init__(self,
equals: Optional[Union[str, LazyProxy]] = None,
contains: Optional[Union[str, LazyProxy]] = None,
startswith: Optional[Union[str, LazyProxy]] = None,
endswith: Optional[Union[str, LazyProxy]] = None,
ignore_case=False):
"""
Check text for one of pattern. Only one mode can be used in one filter.
:param equals:
:param contains:
:param startswith:
:param endswith:
:param ignore_case: case insensitive
"""
# Only one mode can be used. check it.
check = sum(map(bool, (equals, contains, startswith, endswith)))
if check > 1:
args = "' and '".join([arg[0] for arg in [('equals', equals),
('contains', contains),
('startswith', startswith),
('endswith', endswith)
] if arg[1]])
raise ValueError(f"Arguments '{args}' cannot be used together.")
elif check == 0:
raise ValueError(f"No one mode is specified!")
self.equals = equals
self.contains = contains
self.endswith = endswith
self.startswith = startswith
self.ignore_case = ignore_case
[docs] @classmethod
def validate(cls, full_config: Dict[str, Any]):
if 'text' in full_config:
return {'equals': full_config.pop('text')}
elif 'text_contains' in full_config:
return {'contains': full_config.pop('text_contains')}
elif 'text_startswith' in full_config:
return {'startswith': full_config.pop('text_startswith')}
elif 'text_endswith' in full_config:
return {'endswith': full_config.pop('text_endswith')}
[docs] async def check(self, obj: Union[Message, CallbackQuery, InlineQuery]):
if isinstance(obj, Message):
text = obj.text or obj.caption or ''
if not text and obj.poll:
text = obj.poll.question
elif isinstance(obj, CallbackQuery):
text = obj.data
elif isinstance(obj, InlineQuery):
text = obj.query
elif isinstance(obj, Poll):
text = obj.question
else:
return False
if self.ignore_case:
text = text.lower()
if self.equals:
return text == str(self.equals)
elif self.contains:
return str(self.contains) in text
elif self.startswith:
return text.startswith(str(self.startswith))
elif self.endswith:
return text.endswith(str(self.endswith))
return False
[docs]class HashTag(Filter):
"""
Filter for hashtag's and cashtag's
"""
# TODO: allow to use regexp
def __init__(self, hashtags=None, cashtags=None):
if not hashtags and not cashtags:
raise ValueError('No one hashtag or cashtag is specified!')
if hashtags is None:
hashtags = []
elif isinstance(hashtags, str):
hashtags = [hashtags]
if cashtags is None:
cashtags = []
elif isinstance(cashtags, str):
cashtags = [cashtags.upper()]
else:
cashtags = list(map(str.upper, cashtags))
self.hashtags = hashtags
self.cashtags = cashtags
[docs] @classmethod
def validate(cls, full_config: Dict[str, Any]):
config = {}
if 'hashtags' in full_config:
config['hashtags'] = full_config.pop('hashtags')
if 'cashtags' in full_config:
config['cashtags'] = full_config.pop('cashtags')
return config
[docs] async def check(self, message: types.Message):
if message.caption:
text = message.caption
entities = message.caption_entities
elif message.text:
text = message.text
entities = message.entities
else:
return False
hashtags, cashtags = self._get_tags(text, entities)
if self.hashtags and set(hashtags) & set(self.hashtags) \
or self.cashtags and set(cashtags) & set(self.cashtags):
return {'hashtags': hashtags, 'cashtags': cashtags}
def _get_tags(self, text, entities):
hashtags = []
cashtags = []
for entity in entities:
if entity.type == types.MessageEntityType.HASHTAG:
value = entity.get_text(text).lstrip('#')
hashtags.append(value)
elif entity.type == types.MessageEntityType.CASHTAG:
value = entity.get_text(text).lstrip('$')
cashtags.append(value)
return hashtags, cashtags
[docs]class Regexp(Filter):
"""
Regexp filter for messages and callback query
"""
def __init__(self, regexp):
if not isinstance(regexp, re.Pattern):
regexp = re.compile(regexp, flags=re.IGNORECASE | re.MULTILINE)
self.regexp = regexp
[docs] @classmethod
def validate(cls, full_config: Dict[str, Any]):
if 'regexp' in full_config:
return {'regexp': full_config.pop('regexp')}
[docs] async def check(self, obj: Union[Message, CallbackQuery]):
if isinstance(obj, Message):
content = obj.text or obj.caption or ''
if not content and obj.poll:
content = obj.poll.question
elif isinstance(obj, CallbackQuery) and obj.data:
content = obj.data
else:
return False
match = self.regexp.search(content)
if match:
return {'regexp': match}
return False
[docs]class RegexpCommandsFilter(BoundFilter):
"""
Check commands by regexp in message
"""
key = 'regexp_commands'
def __init__(self, regexp_commands):
self.regexp_commands = [re.compile(command, flags=re.IGNORECASE | re.MULTILINE) for command in regexp_commands]
[docs] async def check(self, message):
if not message.is_command():
return False
command = message.text.split()[0][1:]
command, _, mention = command.partition('@')
if mention and mention != (await message.bot.me).username:
return False
for command in self.regexp_commands:
search = command.search(message.text)
if search:
return {'regexp_command': search}
return False
[docs]class ContentTypeFilter(BoundFilter):
"""
Check message content type
"""
key = 'content_types'
required = True
default = types.ContentTypes.TEXT
def __init__(self, content_types):
self.content_types = content_types
[docs] async def check(self, message):
return types.ContentType.ANY in self.content_types or \
message.content_type in self.content_types
[docs]class StateFilter(BoundFilter):
"""
Check user state
"""
key = 'state'
required = True
ctx_state = ContextVar('user_state')
def __init__(self, dispatcher, state):
from aiogram.dispatcher.filters.state import State, StatesGroup
self.dispatcher = dispatcher
states = []
if not isinstance(state, (list, set, tuple, frozenset)) or state is None:
state = [state, ]
for item in state:
if isinstance(item, State):
states.append(item.state)
elif inspect.isclass(item) and issubclass(item, StatesGroup):
states.extend(item.all_states_names)
else:
states.append(item)
self.states = states
def get_target(self, obj):
return getattr(getattr(obj, 'chat', None), 'id', None), getattr(getattr(obj, 'from_user', None), 'id', None)
[docs] async def check(self, obj):
if '*' in self.states:
return {'state': self.dispatcher.current_state()}
try:
state = self.ctx_state.get()
except LookupError:
chat, user = self.get_target(obj)
if chat or user:
state = await self.dispatcher.storage.get_state(chat=chat, user=user)
self.ctx_state.set(state)
if state in self.states:
return {'state': self.dispatcher.current_state(), 'raw_state': state}
else:
if state in self.states:
return {'state': self.dispatcher.current_state(), 'raw_state': state}
return False
[docs]class ExceptionsFilter(BoundFilter):
"""
Filter for exceptions
"""
key = 'exception'
def __init__(self, exception):
self.exception = exception
[docs] async def check(self, update, exception):
try:
raise exception
except self.exception:
return True
except:
return False