#!/usr/bin/env python # # A library that provides a Python interface to the Telegram Bot API # Copyright (C) 2015-2022 # Leandro Toledo de Souza # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Lesser Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Lesser Public License for more details. # # You should have received a copy of the GNU Lesser Public License # along with this program. If not, see [http://www.gnu.org/licenses/]. """This module contains the StringRegexHandler class.""" import re from typing import TYPE_CHECKING, Callable, Dict, Match, Optional, Pattern, TypeVar, Union from telegram.utils.helpers import DefaultValue, DEFAULT_FALSE from .handler import Handler from .utils.types import CCT if TYPE_CHECKING: from telegram.ext import Dispatcher RT = TypeVar('RT') class StringRegexHandler(Handler[str, CCT]): """Handler class to handle string updates based on a regex which checks the update content. Read the documentation of the ``re`` module for more information. The ``re.match`` function is used to determine if an update should be handled by this handler. Note: This handler is not used to handle Telegram :attr:`telegram.Update`, but strings manually put in the queue. For example to send messages with the bot using command line or API. Warning: When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info. Args: pattern (:obj:`str` | :obj:`Pattern`): The regex pattern. callback (:obj:`callable`): The callback function for this handler. Will be called when :attr:`check_update` has determined that an update should be processed by this handler. Callback signature for context based API: ``def callback(update: Update, context: CallbackContext)`` The return value of the callback is usually ignored except for the special case of :class:`telegram.ext.ConversationHandler`. pass_groups (:obj:`bool`, optional): If the callback should be passed the result of ``re.match(pattern, data).groups()`` as a keyword argument called ``groups``. Default is :obj:`False` DEPRECATED: Please switch to context based callbacks. pass_groupdict (:obj:`bool`, optional): If the callback should be passed the result of ``re.match(pattern, data).groupdict()`` as a keyword argument called ``groupdict``. Default is :obj:`False` DEPRECATED: Please switch to context based callbacks. pass_update_queue (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called ``update_queue`` will be passed to the callback function. It will be the ``Queue`` instance used by the :class:`telegram.ext.Updater` and :class:`telegram.ext.Dispatcher` that contains new updates which can be used to insert updates. Default is :obj:`False`. DEPRECATED: Please switch to context based callbacks. pass_job_queue (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called ``job_queue`` will be passed to the callback function. It will be a :class:`telegram.ext.JobQueue` instance created by the :class:`telegram.ext.Updater` which can be used to schedule new jobs. Default is :obj:`False`. DEPRECATED: Please switch to context based callbacks. run_async (:obj:`bool`): Determines whether the callback will run asynchronously. Defaults to :obj:`False`. Attributes: pattern (:obj:`str` | :obj:`Pattern`): The regex pattern. callback (:obj:`callable`): The callback function for this handler. pass_groups (:obj:`bool`): Determines whether ``groups`` will be passed to the callback function. pass_groupdict (:obj:`bool`): Determines whether ``groupdict``. will be passed to the callback function. pass_update_queue (:obj:`bool`): Determines whether ``update_queue`` will be passed to the callback function. pass_job_queue (:obj:`bool`): Determines whether ``job_queue`` will be passed to the callback function. run_async (:obj:`bool`): Determines whether the callback will run asynchronously. """ __slots__ = ('pass_groups', 'pass_groupdict', 'pattern') def __init__( self, pattern: Union[str, Pattern], callback: Callable[[str, CCT], RT], pass_groups: bool = False, pass_groupdict: bool = False, pass_update_queue: bool = False, pass_job_queue: bool = False, run_async: Union[bool, DefaultValue] = DEFAULT_FALSE, ): super().__init__( callback, pass_update_queue=pass_update_queue, pass_job_queue=pass_job_queue, run_async=run_async, ) if isinstance(pattern, str): pattern = re.compile(pattern) self.pattern = pattern self.pass_groups = pass_groups self.pass_groupdict = pass_groupdict def check_update(self, update: object) -> Optional[Match]: """Determines whether an update should be passed to this handlers :attr:`callback`. Args: update (:obj:`object`): The incoming update. Returns: :obj:`bool` """ if isinstance(update, str): match = re.match(self.pattern, update) if match: return match return None def collect_optional_args( self, dispatcher: 'Dispatcher', update: str = None, check_result: Optional[Match] = None, ) -> Dict[str, object]: """Pass the results of ``re.match(pattern, update).{groups(), groupdict()}`` to the callback as a keyword arguments called ``groups`` and ``groupdict``, respectively, if needed. """ optional_args = super().collect_optional_args(dispatcher, update, check_result) if self.pattern: if self.pass_groups and check_result: optional_args['groups'] = check_result.groups() if self.pass_groupdict and check_result: optional_args['groupdict'] = check_result.groupdict() return optional_args def collect_additional_context( self, context: CCT, update: str, dispatcher: 'Dispatcher', check_result: Optional[Match], ) -> None: """Add the result of ``re.match(pattern, update)`` to :attr:`CallbackContext.matches` as list with one element. """ if self.pattern and check_result: context.matches = [check_result]