Telethon/telethon/events/inlinequery.py

222 lines
7.9 KiB
Python
Raw Normal View History

2018-07-15 12:31:14 +03:00
import inspect
import re
import asyncio
from .common import EventBuilder, EventCommon, name_inner_event
from .. import utils
from ..tl import types, functions, custom
from ..tl.custom.sendergetter import SenderGetter
@name_inner_event
class InlineQuery(EventBuilder):
"""
Represents an inline query event (when someone writes ``'@my_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.
"""
def __init__(
self, users=None, *, blacklist_users=False, func=None, pattern=None):
super().__init__(users, blacklist_chats=blacklist_users, func=func)
2018-07-15 12:31:14 +03:00
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')
2018-07-19 02:47:32 +03:00
@classmethod
def build(cls, update):
2018-07-15 12:31:14 +03:00
if isinstance(update, types.UpdateBotInlineQuery):
2018-07-19 02:47:32 +03:00
event = cls.Event(update)
2018-07-15 12:31:14 +03:00
else:
return
event._entities = update._entities
return event
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:`UpdateBotCallbackQuery`):
The original :tl:`UpdateBotCallbackQuery`.
2018-10-12 13:38:46 +03:00
Make sure to access the `text` of the query if
that's what you want instead working with this.
2018-07-15 12:31:14 +03:00
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))
self.query = query
self.pattern_match = None
self._answered = False
self._sender_id = query.user_id
self._input_sender = None
self._sender = None
def _load_entities(self):
self._sender, self._input_sender = self._get_entity_pair(self.sender_id)
return super()._load_entities() and self._input_sender is not None
2018-07-15 12:31:14 +03:00
@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):
"""
2018-10-17 12:47:51 +03:00
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`.
2018-07-15 12:31:14 +03:00
"""
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
@property
def builder(self):
"""
2018-10-12 13:38:46 +03:00
Returns a new `InlineBuilder
<telethon.tl.custom.inlinebuilder.InlineBuilder>` instance.
2018-07-15 12:31:14 +03:00
"""
return custom.InlineBuilder(self._client)
async def answer(
self, results=None, cache_time=0, *,
gallery=False, next_offset=None, private=False,
2018-07-15 12:31:14 +03:00
switch_pm=None, switch_pm_param=''):
"""
Answers the inline query with the given results.
Args:
results (`list`, optional):
A list of :tl:`InputBotInlineResult` to use.
You should use `builder` to create these:
2018-10-12 13:38:46 +03:00
.. code-block:: python
2018-07-15 12:31:14 +03:00
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.
2018-07-15 12:31:14 +03:00
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.
2018-07-15 12:31:14 +03:00
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.
"""
if self._answered:
return
if results:
futures = [self._as_future(x, self._client.loop)
for x in results]
await asyncio.wait(futures, loop=self._client.loop)
# 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 = []
2018-07-15 12:31:14 +03:00
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,
2018-07-15 12:31:14 +03:00
private=private,
switch_pm=switch_pm
)
)
@staticmethod
def _as_future(obj, loop):
2018-07-15 12:31:14 +03:00
if inspect.isawaitable(obj):
return asyncio.ensure_future(obj, loop=loop)
2018-07-15 12:31:14 +03:00
f = loop.create_future()
2018-07-15 12:31:14 +03:00
f.set_result(obj)
return f