mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2024-11-10 19:46:36 +03:00
248 lines
8.8 KiB
Python
248 lines
8.8 KiB
Python
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
|
|
<telethon.tl.custom.inlinebuilder.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
|