import inspect import re import asyncio from .common import EventBuilder, EventCommon, name_inner_event from .. import utils, helpers from ..tl import types, functions, custom from ..tl.custom.sendergetter import SenderGetter @name_inner_event class InlineQuery(EventBuilder): """ Occurs whenever you sign in as a bot and a user sends an inline query such as ``@bot query``. Args: users (`entity`, optional): May be one or more entities (username/peer/etc.), preferably IDs. By default, only inline queries from these users will be handled. blacklist_users (`bool`, optional): Whether to treat the users as a blacklist instead of as a whitelist (default). This means that every chat will be handled *except* those specified in ``users`` which will be ignored if ``blacklist_users=True``. pattern (`str`, `callable`, `Pattern`, optional): If set, only queries matching this pattern will be handled. You can specify a regex-like string which will be matched against the message, a callable function that returns `True` if a message is acceptable, or a compiled regex pattern. Example .. code-block:: python from telethon import events @client.on(events.InlineQuery) async def handler(event): builder = event.builder # Two options (convert user text to UPPERCASE or lowercase) await event.answer([ builder.article('UPPERCASE', text=event.text.upper()), builder.article('lowercase', text=event.text.lower()), ]) """ def __init__( self, users=None, *, blacklist_users=False, func=None, pattern=None): super().__init__(users, blacklist_chats=blacklist_users, func=func) if isinstance(pattern, str): self.pattern = re.compile(pattern).match elif not pattern or callable(pattern): self.pattern = pattern elif hasattr(pattern, 'match') and callable(pattern.match): self.pattern = pattern.match else: raise TypeError('Invalid pattern type given') @classmethod def build(cls, update, others=None, self_id=None): if isinstance(update, types.UpdateBotInlineQuery): return cls.Event(update) def filter(self, event): if self.pattern: match = self.pattern(event.text) if not match: return event.pattern_match = match return super().filter(event) class Event(EventCommon, SenderGetter): """ Represents the event of a new callback query. Members: query (:tl:`UpdateBotInlineQuery`): The original :tl:`UpdateBotInlineQuery`. Make sure to access the `text` property of the query if you want the text rather than the actual query object. pattern_match (`obj`, optional): The resulting object from calling the passed ``pattern`` function, which is ``re.compile(...).match`` by default. """ def __init__(self, query): super().__init__(chat_peer=types.PeerUser(query.user_id)) SenderGetter.__init__(self, query.user_id) self.query = query self.pattern_match = None self._answered = False def _set_client(self, client): super()._set_client(client) self._sender, self._input_sender = utils._get_entity_pair( self.sender_id, self._entities, client._mb_entity_cache) @property def id(self): """ Returns the unique identifier for the query ID. """ return self.query.query_id @property def text(self): """ Returns the text the user used to make the inline query. """ return self.query.query @property def offset(self): """ The string the user's client used as an offset for the query. This will either be empty or equal to offsets passed to `answer`. """ return self.query.offset @property def geo(self): """ If the user location is requested when using inline mode and the user's device is able to send it, this will return the :tl:`GeoPoint` with the position of the user. """ return self.query.geo @property def builder(self): """ Returns a new `InlineBuilder ` instance. """ return custom.InlineBuilder(self._client) async def answer( self, results=None, cache_time=0, *, gallery=False, next_offset=None, private=False, switch_pm=None, switch_pm_param=''): """ Answers the inline query with the given results. See the documentation for `builder` to know what kind of answers can be given. Args: results (`list`, optional): A list of :tl:`InputBotInlineResult` to use. You should use `builder` to create these: .. code-block:: python builder = inline.builder r1 = builder.article('Be nice', text='Have a nice day') r2 = builder.article('Be bad', text="I don't like you") await inline.answer([r1, r2]) You can send up to 50 results as documented in https://core.telegram.org/bots/api#answerinlinequery. Sending more will raise ``ResultsTooMuchError``, and you should consider using `next_offset` to paginate them. cache_time (`int`, optional): For how long this result should be cached on the user's client. Defaults to 0 for no cache. gallery (`bool`, optional): Whether the results should show as a gallery (grid) or not. next_offset (`str`, optional): The offset the client will send when the user scrolls the results and it repeats the request. private (`bool`, optional): Whether the results should be cached by Telegram (not private) or by the user's client (private). switch_pm (`str`, optional): If set, this text will be shown in the results to allow the user to switch to private messages. switch_pm_param (`str`, optional): Optional parameter to start the bot with if `switch_pm` was used. Example: .. code-block:: python @bot.on(events.InlineQuery) async def handler(event): builder = event.builder rev_text = event.text[::-1] await event.answer([ builder.article('Reverse text', text=rev_text), builder.photo('/path/to/photo.jpg') ]) """ if self._answered: return if results: futures = [self._as_future(x) for x in results] await asyncio.wait(futures) # All futures will be in the `done` *set* that `wait` returns. # # Precisely because it's a `set` and not a `list`, it # will not preserve the order, but since all futures # completed we can use our original, ordered `list`. results = [x.result() for x in futures] else: results = [] if switch_pm: switch_pm = types.InlineBotSwitchPM(switch_pm, switch_pm_param) return await self._client( functions.messages.SetInlineBotResultsRequest( query_id=self.query.query_id, results=results, cache_time=cache_time, gallery=gallery, next_offset=next_offset, private=private, switch_pm=switch_pm ) ) @staticmethod def _as_future(obj): if inspect.isawaitable(obj): return asyncio.ensure_future(obj) f = helpers.get_running_loop().create_future() f.set_result(obj) return f