yasirarism/pyrofork
1
0
Fork 0
mirror of https://github.com/Mayuri-Chan/pyrofork.git synced 2025-12-29 12:04:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Added Conversation Support Based on Pyromodv2

Browse source 
Co-authored-by: Alisson Lauffer <alissonvitortc@gmail.com>
Co-authored-by: Yasir Aris M <git@yasirdev.my.id>
Co-authored-by: wulan17 <wulan17@nusantararom.org>
Signed-off-by: Yasir Aris M <git@yasirdev.my.id>
Signed-off-by: wulan17 <wulan17@nusantararom.org>
This commit is contained in:
Cezar H 2023-11-03 11:01:15 +07:00 committed by wulan17
parent e3e84d91e8
commit 8aee144313
No known key found for this signature in database
GPG key ID: 318CD6CD3A6AC0A5
26 changed files with 1317 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

308
pyrogram/client.py
View file

@ -57,10 +57,11 @@ except Exception:
pass
else:
from pyrogram.storage import MongoStorage
from pyrogram.types import User, TermsOfService
from pyrogram.utils import ainput
from pyrogram.types import User, TermsOfService, ListenerTypes, Identifier, Listener
from pyrogram.utils import ainput, PyromodConfig
from .dispatcher import Dispatcher
from .file_id import FileId, FileType, ThumbnailSource
from .filters import Filter
from .mime_types import mime_types
from .parser import Parser
from .session.internals import MsgId
@ -331,7 +332,7 @@ class Client(Methods):
self.updates_watchdog_task = None
self.updates_watchdog_event = asyncio.Event()
self.last_update_time = datetime.now()
self.listeners = {listener_type: [] for listener_type in ListenerTypes}
self.loop = asyncio.get_event_loop()
def __enter__(self):
@ -364,6 +365,307 @@ class Client(Methods):
if datetime.now() - self.last_update_time > timedelta(seconds=self.UPDATES_WATCHDOG_INTERVAL):
await self.invoke(raw.functions.updates.GetState())
async def listen(
self,
filters: Optional[Filter] = None,
listener_type: ListenerTypes = ListenerTypes.MESSAGE,
timeout: Optional[int] = None,
unallowed_click_alert: bool = True,
chat_id: Union[Union[int, str], List[Union[int, str]]] = None,
user_id: Union[Union[int, str], List[Union[int, str]]] = None,
message_id: Union[int, List[int]] = None,
inline_message_id: Union[str, List[str]] = None,
):
"""
Creates a listener and waits for it to be fulfilled.
:param filters: A filter to check if the listener should be fulfilled.
:param listener_type: The type of listener to create. Defaults to :attr:`pyromod.types.ListenerTypes.MESSAGE`.
:param timeout: The maximum amount of time to wait for the listener to be fulfilled. Defaults to ``None``.
:param unallowed_click_alert: Whether to alert the user if they click on a button that is not intended for them. Defaults to ``True``.
:param chat_id: The chat ID(s) to listen for. Defaults to ``None``.
:param user_id: The user ID(s) to listen for. Defaults to ``None``.
:param message_id: The message ID(s) to listen for. Defaults to ``None``.
:param inline_message_id: The inline message ID(s) to listen for. Defaults to ``None``.
:return: The Message or CallbackQuery that fulfilled the listener.
"""
pattern = Identifier(
from_user_id=user_id,
chat_id=chat_id,
message_id=message_id,
inline_message_id=inline_message_id,
)
loop = asyncio.get_event_loop()
future = loop.create_future()
listener = Listener(
future=future,
filters=filters,
unallowed_click_alert=unallowed_click_alert,
identifier=pattern,
listener_type=listener_type,
)
future.add_done_callback(lambda _future: self.remove_listener(listener))
self.listeners[listener_type].append(listener)
try:
return await asyncio.wait_for(future, timeout)
except asyncio.exceptions.TimeoutError:
if callable(PyromodConfig.timeout_handler):
if inspect.iscoroutinefunction(PyromodConfig.timeout_handler.__call__):
await PyromodConfig.timeout_handler(pattern, listener, timeout)
else:
await self.loop.run_in_executor(
None, PyromodConfig.timeout_handler, pattern, listener, timeout
)
elif PyromodConfig.throw_exceptions:
raise ListenerTimeout(timeout)
async def ask(
self,
chat_id: Union[Union[int, str], List[Union[int, str]]],
text: str,
filters: Optional[Filter] = None,
listener_type: ListenerTypes = ListenerTypes.MESSAGE,
timeout: Optional[int] = None,
unallowed_click_alert: bool = True,
user_id: Union[Union[int, str], List[Union[int, str]]] = None,
message_id: Union[int, List[int]] = None,
inline_message_id: Union[str, List[str]] = None,
*args,
**kwargs,
):
"""
Sends a message and waits for a response.
:param chat_id: The chat ID(s) to wait for a message from. The first chat ID will be used to send the message.
:param text: The text to send.
:param filters: Same as :meth:`pyromod.types.Client.listen`.
:param listener_type: Same as :meth:`pyromod.types.Client.listen`.
:param timeout: Same as :meth:`pyromod.types.Client.listen`.
:param unallowed_click_alert: Same as :meth:`pyromod.types.Client.listen`.
:param user_id: Same as :meth:`pyromod.types.Client.listen`.
:param message_id: Same as :meth:`pyromod.types.Client.listen`.
:param inline_message_id: Same as :meth:`pyromod.types.Client.listen`.
:param args: Additional arguments to pass to :meth:`pyrogram.Client.send_message`.
:param kwargs: Additional keyword arguments to pass to :meth:`pyrogram.Client.send_message`.
:return:
Same as :meth:`pyromod.types.Client.listen`. The sent message is returned as the attribute ``sent_message``.
"""
sent_message = None
if text.strip() != "":
chat_to_ask = chat_id[0] if isinstance(chat_id, list) else chat_id
sent_message = await self.send_message(chat_to_ask, text, *args, **kwargs)
response = await self.listen(
filters=filters,
listener_type=listener_type,
timeout=timeout,
unallowed_click_alert=unallowed_click_alert,
chat_id=chat_id,
user_id=user_id,
message_id=message_id,
inline_message_id=inline_message_id,
)
if response:
response.sent_message = sent_message
return response
def remove_listener(self, listener: Listener):
"""
Removes a listener from the :meth:`pyromod.types.Client.listeners` dictionary.
:param listener: The listener to remove.
:return: ``void``
"""
try:
self.listeners[listener.listener_type].remove(listener)
except ValueError:
pass
def get_listener_matching_with_data(
self, data: Identifier, listener_type: ListenerTypes
) -> Optional[Listener]:
"""
Gets a listener that matches the given data.
:param data: A :class:`pyromod.types.Identifier` to match against.
:param listener_type: The type of listener to get. Must be a value from :class:`pyromod.types.ListenerTypes`.
:return: The listener that matches the given data or ``None`` if no listener matches.
"""
matching = []
for listener in self.listeners[listener_type]:
if listener.identifier.matches(data):
matching.append(listener)
# in case of multiple matching listeners, the most specific should be returned
def count_populated_attributes(listener_item: Listener):
return listener_item.identifier.count_populated()
return max(matching, key=count_populated_attributes, default=None)
def get_listener_matching_with_identifier_pattern(
self, pattern: Identifier, listener_type: ListenerTypes
) -> Optional[Listener]:
"""
Gets a listener that matches the given identifier pattern.
The difference from :meth:`pyromod.types.Client.get_listener_matching_with_data` is that this method
intends to get a listener by passing partial info of the listener identifier, while the other method
intends to get a listener by passing the full info of the update data, which the listener should match with.
:param pattern: A :class:`pyromod.types.Identifier` to match against.
:param listener_type: The type of listener to get. Must be a value from :class:`pyromod.types.ListenerTypes`.
:return: The listener that matches the given identifier pattern or ``None`` if no listener matches.
"""
matching = []
for listener in self.listeners[listener_type]:
if pattern.matches(listener.identifier):
matching.append(listener)
# in case of multiple matching listeners, the most specific should be returned
def count_populated_attributes(listener_item: Listener):
return listener_item.identifier.count_populated()
return max(matching, key=count_populated_attributes, default=None)
def get_many_listeners_matching_with_data(
self,
data: Identifier,
listener_type: ListenerTypes,
) -> List[Listener]:
"""
Same of :meth:`pyromod.types.Client.get_listener_matching_with_data` but returns a list of listeners instead of one.
:param data: Same as :meth:`pyromod.types.Client.get_listener_matching_with_data`.
:param listener_type: Same as :meth:`pyromod.types.Client.get_listener_matching_with_data`.
:return: A list of listeners that match the given data.
"""
listeners = []
for listener in self.listeners[listener_type]:
if listener.identifier.matches(data):
listeners.append(listener)
return listeners
def get_many_listeners_matching_with_identifier_pattern(
self,
pattern: Identifier,
listener_type: ListenerTypes,
) -> List[Listener]:
"""
Same of :meth:`pyromod.types.Client.get_listener_matching_with_identifier_pattern` but returns a list of listeners instead of one.
:param pattern: Same as :meth:`pyromod.types.Client.get_listener_matching_with_identifier_pattern`.
:param listener_type: Same as :meth:`pyromod.types.Client.get_listener_matching_with_identifier_pattern`.
:return: A list of listeners that match the given identifier pattern.
"""
listeners = []
for listener in self.listeners[listener_type]:
if pattern.matches(listener.identifier):
listeners.append(listener)
return listeners
async def stop_listening(
self,
listener_type: ListenerTypes = ListenerTypes.MESSAGE,
chat_id: Union[Union[int, str], List[Union[int, str]]] = None,
user_id: Union[Union[int, str], List[Union[int, str]]] = None,
message_id: Union[int, List[int]] = None,
inline_message_id: Union[str, List[str]] = None,
):
"""
Stops all listeners that match the given identifier pattern.
Uses :meth:`pyromod.types.Client.get_many_listeners_matching_with_identifier_pattern`.
:param listener_type: The type of listener to stop. Must be a value from :class:`pyromod.types.ListenerTypes`.
:param chat_id: The chat_id to match against.
:param user_id: The user_id to match against.
:param message_id: The message_id to match against.
:param inline_message_id: The inline_message_id to match against.
:return: ``void``
"""
pattern = Identifier(
from_user_id=user_id,
chat_id=chat_id,
message_id=message_id,
inline_message_id=inline_message_id,
)
listeners = self.get_many_listeners_matching_with_identifier_pattern(
pattern, listener_type
)
for listener in listeners:
await self.stop_listener(listener)
async def stop_listener(self, listener: Listener):
"""
Stops a listener, calling stopped_handler if applicable or raising ListenerStopped if throw_exceptions is True.
:param listener: The :class:`pyromod.types.Listener` to stop.
:return: ``void``
:raises ListenerStopped: If throw_exceptions is True.
"""
self.remove_listener(listener)
if listener.future.done():
return
if callable(PyromodConfig.stopped_handler):
if inspect.iscoroutinefunction(PyromodConfig.stopped_handler.__call__):
await PyromodConfig.stopped_handler(None, listener)
else:
await self.loop.run_in_executor(
None, PyromodConfig.stopped_handler, None, listener
)
elif PyromodConfig.throw_exceptions:
listener.future.set_exception(ListenerStopped())
def register_next_step_handler(
self,
callback: Callable,
filters: Optional[Filter] = None,
listener_type: ListenerTypes = ListenerTypes.MESSAGE,
unallowed_click_alert: bool = True,
chat_id: Union[Union[int, str], List[Union[int, str]]] = None,
user_id: Union[Union[int, str], List[Union[int, str]]] = None,
message_id: Union[int, List[int]] = None,
inline_message_id: Union[str, List[str]] = None,
):
"""
Registers a listener with a callback to be called when the listener is fulfilled.
:param callback: The callback to call when the listener is fulfilled.
:param filters: Same as :meth:`pyromod.types.Client.listen`.
:param listener_type: Same as :meth:`pyromod.types.Client.listen`.
:param unallowed_click_alert: Same as :meth:`pyromod.types.Client.listen`.
:param chat_id: Same as :meth:`pyromod.types.Client.listen`.
:param user_id: Same as :meth:`pyromod.types.Client.listen`.
:param message_id: Same as :meth:`pyromod.types.Client.listen`.
:param inline_message_id: Same as :meth:`pyromod.types.Client.listen`.
:return: ``void``
"""
pattern = Identifier(
from_user_id=user_id,
chat_id=chat_id,
message_id=message_id,
inline_message_id=inline_message_id,
)
listener = Listener(
callback=callback,
filters=filters,
unallowed_click_alert=unallowed_click_alert,
identifier=pattern,
listener_type=listener_type,
)
self.listeners[listener_type].append(listener)
async def authorize(self) -> User:
if self.bot_token:
return await self.sign_in_bot(self.bot_token)

5
pyrogram/dispatcher.py
View file

@ -26,7 +26,7 @@ import pyrogram
from pyrogram import utils
from pyrogram.handlers import (
CallbackQueryHandler, MessageHandler, EditedMessageHandler, DeletedMessagesHandler,
UserStatusHandler, RawUpdateHandler, InlineQueryHandler, PollHandler,
UserStatusHandler, RawUpdateHandler, InlineQueryHandler, PollHandler, ConversationHandler,
ChosenInlineResultHandler, ChatMemberUpdatedHandler, ChatJoinRequestHandler, StoryHandler
)
from pyrogram.raw.types import (
@ -65,6 +65,9 @@ class Dispatcher:
self.updates_queue = asyncio.Queue()
self.groups = OrderedDict()
self.conversation_handler = ConversationHandler()
self.groups[0] = [self.conversation_handler]
async def message_parser(update, users, chats):
return (
await pyrogram.types.Message._parse(self.client, update.message, users, chats,

1
pyrogram/errors/__init__.py
View file

@ -17,6 +17,7 @@
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
from .exceptions import *
from .pyromod import *
from .rpc_error import UnknownError

4
pyrogram/errors/pyromod/__init__.py Normal file
View file

@ -0,0 +1,4 @@
from .listener_stopped import ListenerStopped
from .listener_timeout import ListenerTimeout
__all__ = ["ListenerStopped", "ListenerTimeout"]

2
pyrogram/errors/pyromod/listener_stopped.py Normal file
View file

@ -0,0 +1,2 @@
class ListenerStopped(Exception):
pass

2
pyrogram/errors/pyromod/listener_timeout.py Normal file
View file

@ -0,0 +1,2 @@
class ListenerTimeout(Exception):
pass

1
pyrogram/handlers/__init__.py
View file

@ -20,6 +20,7 @@
from .callback_query_handler import CallbackQueryHandler
from .chat_join_request_handler import ChatJoinRequestHandler
from .chat_member_updated_handler import ChatMemberUpdatedHandler
from .conversation_handler import ConversationHandler
from .chosen_inline_result_handler import ChosenInlineResultHandler
from .deleted_messages_handler import DeletedMessagesHandler
from .disconnect_handler import DisconnectHandler

161
pyrogram/handlers/callback_query_handler.py
View file

@ -16,7 +16,13 @@
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
from typing import Callable
from asyncio import iscoroutinefunction
from typing import Callable, Tuple
import pyrogram
from pyrogram.utils import PyromodConfig
from pyrogram.types import ListenerTypes, CallbackQuery, Identifier, Listener
from .handler import Handler
@ -46,4 +52,155 @@ class CallbackQueryHandler(Handler):
"""
def __init__(self, callback: Callable, filters=None):
super().__init__(callback, filters)
self.original_callback = callback
super().__init__(self.resolve_future_or_callback, filters)
def compose_data_identifier(self, query: CallbackQuery):
"""
Composes an Identifier object from a CallbackQuery object.
:param query: The CallbackQuery object to compose of.
:return: An Identifier object.
"""
from_user = query.from_user
from_user_id = from_user.id if from_user else None
from_user_username = from_user.username if from_user else None
chat_id = None
message_id = None
if query.message:
message_id = getattr(
query.message, "id", getattr(query.message, "message_id", None)
)
if query.message.chat:
chat_id = [query.message.chat.id, query.message.chat.username]
return Identifier(
message_id=message_id,
chat_id=chat_id,
from_user_id=[from_user_id, from_user_username],
inline_message_id=query.inline_message_id,
)
async def check_if_has_matching_listener(
self, client: "pyrogram.Client", query: CallbackQuery
) -> Tuple[bool, Listener]:
"""
Checks if the CallbackQuery object has a matching listener.
:param client: The Client object to check with.
:param query: The CallbackQuery object to check with.
:return: A tuple of a boolean and a Listener object. The boolean indicates whether
the found listener has filters and its filters matches with the CallbackQuery object.
The Listener object is the matching listener.
"""
data = self.compose_data_identifier(query)
listener = client.get_listener_matching_with_data(
data, ListenerTypes.CALLBACK_QUERY
)
listener_does_match = False
if listener:
filters = listener.filters
if callable(filters):
if iscoroutinefunction(filters.__call__):
listener_does_match = await filters(client, query)
else:
listener_does_match = await client.loop.run_in_executor(
None, filters, client, query
)
else:
listener_does_match = True
return listener_does_match, listener
async def check(self, client: "pyrogram.Client", query: CallbackQuery):
"""
Checks if the CallbackQuery object has a matching listener or handler.
:param client: The Client object to check with.
:param query: The CallbackQuery object to check with.
:return: A boolean indicating whether the CallbackQuery object has a matching listener or the handler
filter matches.
"""
listener_does_match, listener = await self.check_if_has_matching_listener(
client, query
)
if callable(self.filters):
if iscoroutinefunction(self.filters.__call__):
handler_does_match = await self.filters(client, query)
else:
handler_does_match = await client.loop.run_in_executor(
None, self.filters, client, query
)
else:
handler_does_match = True
data = self.compose_data_identifier(query)
if PyromodConfig.unallowed_click_alert:
# matches with the current query but from any user
permissive_identifier = Identifier(
chat_id=data.chat_id,
message_id=data.message_id,
inline_message_id=data.inline_message_id,
from_user_id=None,
)
matches = permissive_identifier.matches(data)
if (
listener
and (matches and not listener_does_match)
and listener.unallowed_click_alert
):
alert = (
listener.unallowed_click_alert
if isinstance(listener.unallowed_click_alert, str)
else PyromodConfig.unallowed_click_alert_text
)
await query.answer(alert)
return False
# let handler get the chance to handle if listener
# exists but its filters doesn't match
return listener_does_match or handler_does_match
async def resolve_future_or_callback(
self, client: "pyrogram.Client", query: CallbackQuery, *args
):
"""
Resolves the future or calls the callback of the listener. Will call the original handler if no listener.
:param client: The Client object to resolve or call with.
:param query: The CallbackQuery object to resolve or call with.
:param args: The arguments to call the callback with.
:return: None
"""
listener_does_match, listener = await self.check_if_has_matching_listener(
client, query
)
if listener and listener_does_match:
client.remove_listener(listener)
if listener.future and not listener.future.done():
listener.future.set_result(query)
raise pyrogram.StopPropagation
elif listener.callback:
if iscoroutinefunction(listener.callback):
await listener.callback(client, query, *args)
else:
listener.callback(client, query, *args)
raise pyrogram.StopPropagation
else:
raise ValueError("Listener must have either a future or a callback")
else:
await self.original_callback(client, query, *args)

68
pyrogram/handlers/conversation_handler.py Normal file
View file

@ -0,0 +1,68 @@
# Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-present Dan <https://github.com/delivrance>
#
# This file is part of Pyrogram.
#
# Pyrogram is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pyrogram 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 General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import inspect
from typing import Union
import pyrogram
from pyrogram.types import Message, CallbackQuery
from .message_handler import MessageHandler
from .callback_query_handler import CallbackQueryHandler
class ConversationHandler(MessageHandler, CallbackQueryHandler):
"""The Conversation handler class."""
def __init__(self):
self.waiters = {}
async def check(self, client: "pyrogram.Client", update: Union[Message, CallbackQuery]):
if isinstance(update, Message) and update.outgoing:
return False
try:
chat_id = update.chat.id if isinstance(update, Message) else update.message.chat.id
except AttributeError:
return False
waiter = self.waiters.get(chat_id)
if not waiter or not isinstance(update, waiter['update_type']) or waiter['future'].done():
return False
filters = waiter.get('filters')
if callable(filters):
if inspect.iscoroutinefunction(filters.__call__):
filtered = await filters(client, update)
else:
filtered = await client.loop.run_in_executor(
client.executor,
filters,
client, update
)
if not filtered or waiter['future'].done():
return False
waiter['future'].set_result(update)
return True
@staticmethod
async def callback(_, __):
pass
def delete_waiter(self, chat_id, future):
if future == self.waiters[chat_id]['future']:
del self.waiters[chat_id]

105
pyrogram/handlers/message_handler.py
View file

@ -15,8 +15,11 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
from inspect import iscoroutinefunction
from typing import Callable
import pyrogram
from pyrogram.types import ListenerTypes, Message, Identifier
from .handler import Handler
@ -46,4 +49,102 @@ class MessageHandler(Handler):
"""
def __init__(self, callback: Callable, filters=None):
super().__init__(callback, filters)
self.original_callback = callback
super().__init__(self.resolve_future_or_callback, filters)
async def check_if_has_matching_listener(self, client: "pyrogram.Client", message: Message):
"""
Checks if the message has a matching listener.
:param client: The Client object to check with.
:param message: The Message object to check with.
:return: A tuple of whether the message has a matching listener and its filters does match with the Message
and the matching listener;
"""
from_user = message.from_user
from_user_id = from_user.id if from_user else None
from_user_username = from_user.username if from_user else None
message_id = getattr(message, "id", getattr(message, "message_id", None))
data = Identifier(
message_id=message_id,
chat_id=[message.chat.id, message.chat.username],
from_user_id=[from_user_id, from_user_username],
)
listener = client.get_listener_matching_with_data(data, ListenerTypes.MESSAGE)
listener_does_match = False
if listener:
filters = listener.filters
if callable(filters):
if iscoroutinefunction(filters.__call__):
listener_does_match = await filters(client, message)
else:
listener_does_match = await client.loop.run_in_executor(
None, filters, client, message
)
else:
listener_does_match = True
return listener_does_match, listener
async def check(self, client: "pyrogram.Client", message: Message):
"""
Checks if the message has a matching listener or handler and its filters does match with the Message.
:param client: Client object to check with.
:param message: Message object to check with.
:return: Whether the message has a matching listener or handler and its filters does match with the Message.
"""
listener_does_match = (
await self.check_if_has_matching_listener(client, message)
)[0]
if callable(self.filters):
if iscoroutinefunction(self.filters.__call__):
handler_does_match = await self.filters(client, message)
else:
handler_does_match = await client.loop.run_in_executor(
None, self.filters, client, message
)
else:
handler_does_match = True
# let handler get the chance to handle if listener
# exists but its filters doesn't match
return listener_does_match or handler_does_match
async def resolve_future_or_callback(self, client: "pyrogram.Client", message: Message, *args):
"""
Resolves the future or calls the callback of the listener if the message has a matching listener.
:param client: Client object to resolve or call with.
:param message: Message object to resolve or call with.
:param args: Arguments to call the callback with.
:return: None
"""
listener_does_match, listener = await self.check_if_has_matching_listener(
client, message
)
if listener and listener_does_match:
client.remove_listener(listener)
if listener.future and not listener.future.done():
listener.future.set_result(message)
raise pyrogram.StopPropagation
elif listener.callback:
if iscoroutinefunction(listener.callback):
await listener.callback(client, message, *args)
else:
listener.callback(client, message, *args)
raise pyrogram.StopPropagation
else:
raise ValueError("Listener must have either a future or a callback")
else:
await self.original_callback(client, message, *args)

16
pyrogram/helpers/__init__.py Normal file
View file

@ -0,0 +1,16 @@
"""
pyromod - A monkeypatcher add-on for Pyrogram
Copyright (C) 2020 Cezar H. <https://github.com/usernein>
This file is part of pyromod.
pyromod is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
pyromod 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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with pyromod. If not, see <https://www.gnu.org/licenses/>.
"""
from .helpers import ikb, bki, ntb, btn, kb, kbtn, array_chunk, force_reply

138
pyrogram/helpers/helpers.py Normal file
View file

@ -0,0 +1,138 @@
from pyrogram.types import (
InlineKeyboardButton,
InlineKeyboardMarkup,
KeyboardButton,
ReplyKeyboardMarkup,
ForceReply,
)
def ikb(rows=None):
"""
Create an InlineKeyboardMarkup from a list of lists of buttons.
:param rows: List of lists of buttons. Defaults to empty list.
:return: InlineKeyboardMarkup
"""
if rows is None:
rows = []
lines = []
for row in rows:
line = []
for button in row:
button = (
btn(button, button) if isinstance(button, str) else btn(*button)
) # InlineKeyboardButton
line.append(button)
lines.append(line)
return InlineKeyboardMarkup(inline_keyboard=lines)
# return {'inline_keyboard': lines}
def btn(text, value, type="callback_data"):
"""
Create an InlineKeyboardButton.
:param text: Text of the button.
:param value: Value of the button.
:param type: Type of the button. Defaults to "callback_data".
:return: InlineKeyboardButton
"""
return InlineKeyboardButton(text, **{type: value})
# return {'text': text, type: value}
# The inverse of above
def bki(keyboard):
"""
Create a list of lists of buttons from an InlineKeyboardMarkup.
:param keyboard: InlineKeyboardMarkup
:return: List of lists of buttons
"""
lines = []
for row in keyboard.inline_keyboard:
line = []
for button in row:
button = ntb(button) # btn() format
line.append(button)
lines.append(line)
return lines
# return ikb() format
def ntb(button):
"""
Create a button list from an InlineKeyboardButton.
:param button: InlineKeyboardButton
:return: Button as a list to be used in btn()
"""
for btn_type in [
"callback_data",
"url",
"switch_inline_query",
"switch_inline_query_current_chat",
"callback_game",
]:
value = getattr(button, btn_type)
if value:
break
button = [button.text, value]
if btn_type != "callback_data":
button.append(btn_type)
return button
# return {'text': text, type: value}
def kb(rows=None, **kwargs):
"""
Create a ReplyKeyboardMarkup from a list of lists of buttons.
:param rows: List of lists of buttons. Defaults to empty list.
:param kwargs: Other arguments to pass to ReplyKeyboardMarkup.
:return: ReplyKeyboardMarkup
"""
if rows is None:
rows = []
lines = []
for row in rows:
line = []
for button in row:
button_type = type(button)
if button_type == str:
button = KeyboardButton(button)
elif button_type == dict:
button = KeyboardButton(**button)
line.append(button)
lines.append(line)
return ReplyKeyboardMarkup(keyboard=lines, **kwargs)
kbtn = KeyboardButton
"""
Create a KeyboardButton.
"""
def force_reply(selective=True):
"""
Create a ForceReply.
:param selective: Whether the reply should be selective. Defaults to True.
:return: ForceReply
"""
return ForceReply(selective=selective)
def array_chunk(input_array, size):
"""
Split an array into chunks.
:param input_array: The array to split.
:param size: The size of each chunk.
:return: List of chunks.
"""
return [input_array[i: i + size] for i in range(0, len(input_array), size)]

6
pyrogram/methods/messages/__init__.py
View file

@ -65,6 +65,8 @@ from .send_web_page import SendWebPage
from .stop_poll import StopPoll
from .stream_media import StreamMedia
from .vote_poll import VotePoll
from .wait_for_message import WaitForMessage
from .wait_for_callback_query import WaitForCallbackQuery
class Messages(
@ -116,6 +118,8 @@ class Messages(
GetDiscussionReplies,
GetDiscussionRepliesCount,
StreamMedia,
GetCustomEmojiStickers
GetCustomEmojiStickers,
WaitForMessage,
WaitForCallbackQuery
):
pass

71
pyrogram/methods/messages/wait_for_callback_query.py Normal file
View file

@ -0,0 +1,71 @@
# Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-present Dan <https://github.com/delivrance>
#
# This file is part of Pyrogram.
#
# Pyrogram is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pyrogram 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 General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import asyncio
from typing import Union
from functools import partial
from pyrogram import types
from pyrogram.filters import Filter
class WaitForCallbackQuery:
async def wait_for_callback_query(
self,
chat_id: Union[int, str],
filters: Filter = None,
timeout: int = None
) -> "types.CallbackQuery":
"""Wait for callback query.
Parameters:
chat_id (``int`` | ``str``):
Unique identifier (int) or username (str) of the target chat.
filters (:obj:`Filters`):
Pass one or more filters to allow only a subset of callback queries to be passed
in your callback function.
timeout (``int``, *optional*):
Timeout in seconds.
Returns:
:obj:`~pyrogram.types.CallbackQuery`: On success, the callback query is returned.
Raises:
asyncio.TimeoutError: In case callback query not received within the timeout.
Example:
.. code-block:: python
# Simple example
callback_query = app.wait_for_callback_query(chat_id)
# Example with filter
callback_query = app.wait_for_callback_query(chat_id, filters=filters.user(user_id))
# Example with timeout
callback_query = app.wait_for_callback_query(chat_id, timeout=60)
"""
if not isinstance(chat_id, int):
chat = await self.get_chat(chat_id)
chat_id = chat.id
conversation_handler = self.dispatcher.conversation_handler
future = self.loop.create_future()
future.add_done_callback(
partial(
conversation_handler.delete_waiter,
chat_id
)
)
waiter = dict(future=future, filters=filters, update_type=types.CallbackQuery)
conversation_handler.waiters[chat_id] = waiter
return await asyncio.wait_for(future, timeout=timeout)

71
pyrogram/methods/messages/wait_for_message.py Normal file
View file

@ -0,0 +1,71 @@
# Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-present Dan <https://github.com/delivrance>
#
# This file is part of Pyrogram.
#
# Pyrogram is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pyrogram 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 General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import asyncio
from typing import Union
from functools import partial
from pyrogram import types
from pyrogram.filters import Filter
class WaitForMessage:
async def wait_for_message(
self,
chat_id: Union[int, str],
filters: Filter = None,
timeout: int = None
) -> "types.Message":
"""Wait for message.
Parameters:
chat_id (``int`` | ``str``):
Unique identifier (int) or username (str) of the target chat.
filters (:obj:`Filters`):
Pass one or more filters to allow only a subset of callback queries to be passed
in your callback function.
timeout (``int``, *optional*):
Timeout in seconds.
Returns:
:obj:`~pyrogram.types.Message`: On success, the reply message is returned.
Raises:
asyncio.TimeoutError: In case message not received within the timeout.
Example:
.. code-block:: python
# Simple example
reply_message = app.wait_for_message(chat_id)
# Example with filter
reply_message = app.wait_for_message(chat_id, filters=filters.text)
# Example with timeout
reply_message = app.wait_for_message(chat_id, timeout=60)
"""
if not isinstance(chat_id, int):
chat = await self.get_chat(chat_id)
chat_id = chat.id
conversation_handler = self.dispatcher.conversation_handler
future = self.loop.create_future()
future.add_done_callback(
partial(
conversation_handler.delete_waiter,
chat_id
)
)
waiter = dict(future=future, filters=filters, update_type=types.Message)
conversation_handler.waiters[chat_id] = waiter
return await asyncio.wait_for(future, timeout=timeout)

16
pyrogram/nav/__init__.py Normal file
View file

@ -0,0 +1,16 @@
"""
pyromod - A monkeypatcher add-on for Pyrogram
Copyright (C) 2020 Cezar H. <https://github.com/usernein>
This file is part of pyromod.
pyromod is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
pyromod 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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with pyromod. If not, see <https://www.gnu.org/licenses/>.
"""
from .pagination import Pagination

94
pyrogram/nav/pagination.py Normal file
View file

@ -0,0 +1,94 @@
"""
pyromod - A monkeypatcher add-on for Pyrogram
Copyright (C) 2020 Cezar H. <https://github.com/usernein>
This file is part of pyromod.
pyromod is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
pyromod 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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with pyromod. If not, see <https://www.gnu.org/licenses/>.
"""
import math
from pyrogram.helpers import array_chunk
class Pagination:
def __init__(self, objects, page_data=None, item_data=None, item_title=None):
def default_page_callback(x):
return str(x)
def default_item_callback(i, pg):
return f"[{pg}] {i}"
self.objects = objects
self.page_data = page_data or default_page_callback
self.item_data = item_data or default_item_callback
self.item_title = item_title or default_item_callback
def create(self, page, lines=5, columns=1):
quant_per_page = lines * columns
page = 1 if page <= 0 else page
offset = (page - 1) * quant_per_page
stop = offset + quant_per_page
cutted = self.objects[offset:stop]
total = len(self.objects)
pages_range = [
*range(1, math.ceil(total / quant_per_page) + 1)
] # each item is a page
last_page = len(pages_range)
nav = []
if page <= 3:
for n in [1, 2, 3]:
if n not in pages_range:
continue
text = f"· {n} ·" if n == page else n
nav.append((text, self.page_data(n)))
if last_page >= 4:
nav.append(("4 ›" if last_page > 5 else 4, self.page_data(4)))
if last_page > 4:
nav.append(
(
f"{last_page} »" if last_page > 5 else last_page,
self.page_data(last_page),
)
)
elif page >= last_page - 2:
nav.extend(
[
("« 1" if last_page - 4 > 1 else 1, self.page_data(1)),
(
f"‹ {last_page - 3}" if last_page - 4 > 1 else last_page - 3,
self.page_data(last_page - 3),
),
]
)
for n in range(last_page - 2, last_page + 1):
text = f"· {n} ·" if n == page else n
nav.append((text, self.page_data(n)))
else:
nav = [
("« 1", self.page_data(1)),
(f"‹ {page - 1}", self.page_data(page - 1)),
(f"· {page} ·", "noop"),
(f"{page + 1} ›", self.page_data(page + 1)),
(f"{last_page} »", self.page_data(last_page)),
]
buttons = []
for item in cutted:
buttons.append((self.item_title(item, page), self.item_data(item, page)))
kb_lines = array_chunk(buttons, columns)
if last_page > 1:
kb_lines.append(nav)
return kb_lines

1
pyrogram/types/__init__.py
View file

@ -26,3 +26,4 @@ from .messages_and_media import *
from .object import Object
from .update import *
from .user_and_chats import *
from .pyromod import *

114
pyrogram/types/messages_and_media/message.py
View file

@ -546,6 +546,35 @@ class Message(Object, Update):
self.web_app_data = web_app_data
self.reactions = reactions
async def wait_for_click(
self,
from_user_id: Optional[Union[Union[int, str], List[Union[int, str]]]] = None,
timeout: Optional[int] = None,
filters=None,
alert: Union[str, bool] = True,
):
"""
Waits for a callback query to be clicked on the message.
:param from_user_id: The user ID(s) to wait for. If None, waits for any user.
:param timeout: The timeout in seconds. If None, waits forever.
:param filters: The filters to pass to Client.listen().
:param alert: The alert to show when the button is clicked by users that are not allowed in from_user_id.
:return: The callback query that was clicked.
"""
message_id = getattr(self, "id", getattr(self, "message_id", None))
return await self._client.listen(
(self.chat.id, from_user_id, self.id),
listener_type=types.ListenerTypes.CALLBACK_QUERY,
timeout=timeout,
filters=filters,
unallowed_click_alert=alert,
chat_id=self.chat.id,
user_id=from_user_id,
message_id=message_id,
)
@staticmethod
async def _parse(
client: "pyrogram.Client",
@ -4358,3 +4387,88 @@ class Message(Object, Update):
chat_id=self.chat.id,
message_id=self.id
)
async def ask(
self,
text: str,
quote: bool = None,
parse_mode: Optional["enums.ParseMode"] = None,
entities: List["types.MessageEntity"] = None,
disable_web_page_preview: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
reply_markup=None,
filters=None,
timeout: int = None
) -> "Message":
"""Bound method *ask* of :obj:`~pyrogram.types.Message`.
Use as a shortcut for:
.. code-block:: python
client.send_message(chat_id, "What is your name?")
client.wait_for_message(chat_id)
Example:
.. code-block:: python
message.ask("What is your name?")
Parameters:
text (``str``):
Text of the message to be sent.
quote (``bool``, *optional*):
If ``True``, the message will be sent as a reply to this message.
If *reply_to_message_id* is passed, this parameter will be ignored.
Defaults to ``True`` in group chats and ``False`` in private chats.
parse_mode (:obj:`~pyrogram.enums.ParseMode`, *optional*):
By default, texts are parsed using both Markdown and HTML styles.
You can combine both syntaxes together.
Pass "markdown" or "md" to enable Markdown-style parsing only.
Pass "html" to enable HTML-style parsing only.
Pass None to completely disable style parsing.
entities (List of :obj:`~pyrogram.types.MessageEntity`):
List of special entities that appear in message text, which can be specified instead of *parse_mode*.
disable_web_page_preview (``bool``, *optional*):
Disables link previews for links in this message.
disable_notification (``bool``, *optional*):
Sends the message silently.
Users will receive a notification with no sound.
reply_to_message_id (``int``, *optional*):
If the message is a reply, ID of the original message.
reply_markup (:obj:`~pyrogram.types.InlineKeyboardMarkup` | :obj:`~pyrogram.types.ReplyKeyboardMarkup` | :obj:`~pyrogram.types.ReplyKeyboardRemove` | :obj:`~pyrogram.types.ForceReply`, *optional*):
Additional interface options. An object for an inline keyboard, custom reply keyboard,
instructions to remove reply keyboard or to force a reply from the user.
filters (:obj:`Filters`):
Pass one or more filters to allow only a subset of callback queries to be passed
in your callback function.
timeout (``int``, *optional*):
Timeout in seconds.
Returns:
:obj:`~pyrogram.types.Message`: On success, the reply message is returned.
Raises:
RPCError: In case of a Telegram RPC error.
asyncio.TimeoutError: In case reply not received within the timeout.
"""
if quote is None:
quote = self.chat.type != "private"
if reply_to_message_id is None and quote:
reply_to_message_id = self.id
request = await self._client.send_message(
chat_id=self.chat.id,
text=text,
parse_mode=parse_mode,
entities=entities,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
reply_markup=reply_markup
)
reply_message = await self._client.wait_for_message(
self.chat.id,
filters=filters,
timeout=timeout
)
reply_message.request = request
return reply_message

5
pyrogram/types/pyromod/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from .identifier import Identifier
from .listener import Listener
from .listener_types import ListenerTypes
__all__ = ["Identifier", "Listener", "ListenerTypes"]

41
pyrogram/types/pyromod/identifier.py Normal file
View file

@ -0,0 +1,41 @@
from dataclasses import dataclass
from typing import Optional, Union, List
@dataclass
class Identifier:
inline_message_id: Optional[Union[str, List[str]]] = None
chat_id: Optional[Union[Union[int, str], List[Union[int, str]]]] = None
message_id: Optional[Union[int, List[int]]] = None
from_user_id: Optional[Union[Union[int, str], List[Union[int, str]]]] = None
def matches(self, update: "Identifier") -> bool:
# Compare each property of other with the corresponding property in self
# If the property in self is None, the property in other can be anything
# If the property in self is not None, the property in other must be the same
for field in self.__annotations__:
pattern_value = getattr(self, field)
update_value = getattr(update, field)
if pattern_value is not None:
if isinstance(update_value, list):
if isinstance(pattern_value, list):
if not set(update_value).intersection(set(pattern_value)):
return False
elif pattern_value not in update_value:
return False
elif isinstance(pattern_value, list):
if update_value not in pattern_value:
return False
elif update_value != pattern_value:
return False
return True
def count_populated(self):
non_null_count = 0
for attr in self.__annotations__:
if getattr(self, attr) is not None:
non_null_count += 1
return non_null_count

19
pyrogram/types/pyromod/listener.py Normal file
View file

@ -0,0 +1,19 @@
from asyncio import Future
from dataclasses import dataclass
from typing import Callable
import pyrogram
from pyrogram import filters
from .identifier import Identifier
from .listener_types import ListenerTypes
@dataclass
class Listener:
listener_type: ListenerTypes
filters: "pyrogram.filters.Filter"
unallowed_click_alert: bool
identifier: Identifier
future: Future = None
callback: Callable = None

6
pyrogram/types/pyromod/listener_types.py Normal file
View file

@ -0,0 +1,6 @@
from enum import Enum
class ListenerTypes(Enum):
MESSAGE = "message"
CALLBACK_QUERY = "callback_query"

30
pyrogram/types/user_and_chats/chat.py
View file

@ -402,6 +402,36 @@ class Chat(Object):
else:
return Chat._parse_channel_chat(client, chat)
def listen(self, *args, **kwargs):
"""
Listens for messages in the chat. Calls Client.listen() with the chat_id set to the chat's id.
:param args: Arguments to pass to Client.listen().
:param kwargs: Keyword arguments to pass to Client.listen().
:return: The return value of Client.listen().
"""
return self._client.listen(*args, chat_id=self.id, **kwargs)
def ask(self, text, *args, **kwargs):
"""
Asks a question in the chat. Calls Client.ask() with the chat_id set to the chat's id.
:param text: The text to send.
:param args: Arguments to pass to Client.ask().
:param kwargs: Keyword arguments to pass to Client.ask().
:return: The return value of Client.ask().
"""
return self._client.ask(self.id, text, *args, **kwargs)
def stop_listening(self, *args, **kwargs):
"""
Stops listening for messages in the chat. Calls Client.stop_listening() with the chat_id set to the chat's id.
:param args: Arguments to pass to Client.stop_listening().
:param kwargs: Keyword arguments to pass to Client.stop_listening().
:return: The return value of Client.stop_listening().
"""
return self._client.stop_listening(*args, chat_id=self.id, **kwargs)
async def archive(self):
"""Bound method *archive* of :obj:`~pyrogram.types.Chat`.

31
pyrogram/types/user_and_chats/user.py
View file

@ -308,6 +308,37 @@ class User(Object, Update):
client=client
)
def listen(self, *args, **kwargs):
"""
Listens for messages from the user. Calls Client.listen() with the user_id set to the user's id.
:param args: Arguments to pass to Client.listen().
:param kwargs: Keyword arguments to pass to Client.listen().
:return: The return value of Client.listen().
"""
return self._client.listen(*args, user_id=self.id, **kwargs)
def ask(self, text, *args, **kwargs):
"""
Asks a question to the user. Calls Client.ask() with both chat_id and user_id set to the user's id.
:param text: The text to send.
:param args: Arguments to pass to Client.ask().
:param kwargs: Keyword arguments to pass to Client.ask().
:return: The return value of Client.ask().
"""
return self._client.ask(self.id, text, *args, user_id=self.id, **kwargs)
def stop_listening(self, *args, **kwargs):
"""
Stops listening for messages from the user. Calls Client.stop_listening() with the user_id set to the user's id.
:param args: Arguments to pass to Client.stop_listening().
:param kwargs: Keyword arguments to pass to Client.stop_listening().
:return: The return value of Client.stop_listening().
"""
return self._client.stop_listening(*args, user_id=self.id, **kwargs)
async def archive(self):
"""Bound method *archive* of :obj:`~pyrogram.types.User`.

10
pyrogram/utils.py
View file

@ -27,6 +27,7 @@ from concurrent.futures.thread import ThreadPoolExecutor
from datetime import datetime, timezone
from getpass import getpass
from typing import Union, List, Dict, Optional, Any, Callable, TypeVar
from types import SimpleNamespace
import pyrogram
from pyrogram import raw, enums
@ -34,6 +35,15 @@ from pyrogram import types
from pyrogram.file_id import FileId, FileType, PHOTO_TYPES, DOCUMENT_TYPES
PyromodConfig = SimpleNamespace(
timeout_handler=None,
stopped_handler=None,
throw_exceptions=True,
unallowed_click_alert=True,
unallowed_click_alert_text=("[pyromod] You're not expected to click this button."),
)
async def ainput(prompt: str = "", *, hide: bool = False):
"""Just like the built-in input, but async"""
with ThreadPoolExecutor(1) as executor: