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

Pyrofork: Move pyromod's client bound methods to its own class

Browse source 
Signed-off-by: wulan17 <wulan17@nusantararom.org>
This commit is contained in:
wulan17 2023-12-13 09:01:02 +07:00
parent b7cf8cd3fd
commit 7a7225b08f
No known key found for this signature in database
GPG key ID: 318CD6CD3A6AC0A5
13 changed files with 609 additions and 307 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

305
pyrogram/client.py
View file

@ -57,8 +57,8 @@ except Exception:
pass pass
else: else:
from pyrogram.storage import MongoStorage from pyrogram.storage import MongoStorage
from pyrogram.types import User, TermsOfService, ListenerTypes, Identifier, Listener from pyrogram.types import User, TermsOfService, ListenerTypes
from pyrogram.utils import ainput, PyromodConfig from pyrogram.utils import ainput
from .dispatcher import Dispatcher from .dispatcher import Dispatcher
from .file_id import FileId, FileType, ThumbnailSource from .file_id import FileId, FileType, ThumbnailSource
from .filters import Filter from .filters import Filter
@ -365,307 +365,6 @@ class Client(Methods):
if datetime.now() - self.last_update_time > timedelta(seconds=self.UPDATES_WATCHDOG_INTERVAL): if datetime.now() - self.last_update_time > timedelta(seconds=self.UPDATES_WATCHDOG_INTERVAL):
await self.invoke(raw.functions.updates.GetState()) 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: async def authorize(self) -> User:
if self.bot_token: if self.bot_token:
return await self.sign_in_bot(self.bot_token) return await self.sign_in_bot(self.bot_token)

11
pyrogram/methods/__init__.py
View file

@ -1,20 +1,21 @@
# Pyrogram - Telegram MTProto API Client Library for Python # Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-present Dan <https://github.com/delivrance> # Copyright (C) 2017-present Dan <https://github.com/delivrance>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
# #
# This file is part of Pyrogram. # This file is part of Pyrofork.
# #
# Pyrogram is free software: you can redistribute it and/or modify # Pyrofork is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published # 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 # by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version. # (at your option) any later version.
# #
# Pyrogram is distributed in the hope that it will be useful, # Pyrofork is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of # but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details. # GNU Lesser General Public License for more details.
# #
# You should have received a copy of the GNU Lesser General Public License # You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>. # along with Pyrofork. If not, see <http://www.gnu.org/licenses/>.
from .advanced import Advanced from .advanced import Advanced
from .auth import Auth from .auth import Auth
@ -25,6 +26,7 @@ from .decorators import Decorators
from .invite_links import InviteLinks from .invite_links import InviteLinks
from .messages import Messages from .messages import Messages
from .password import Password from .password import Password
from .pyromod import Pyromod
from .stickers import Stickers from .stickers import Stickers
from .users import Users from .users import Users
from .utilities import Utilities from .utilities import Utilities
@ -36,6 +38,7 @@ class Methods(
Bots, Bots,
Contacts, Contacts,
Password, Password,
Pyromod,
Chats, Chats,
Stickers, Stickers,
Users, Users,

42
pyrogram/methods/pyromod/__init__.py Normal file
View file

@ -0,0 +1,42 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
from .ask import Ask
from .get_listener_matching_with_data import GetListenerMatchingWithData
from .get_listener_matching_with_identifier_pattern import GetListenerMatchingWithIdentifierPattern
from .get_many_listeners_matching_with_data import GetManyListenersMatchingWithData
from .get_many_listeners_matching_with_identifier_pattern import GetManyListenersMatchingWithIdentifierPattern
from .listen import Listen
from .register_next_step_handler import RegisterNextStepHandler
from .remove_listerner import RemoveListener
from .stop_listener import StopListener
from .stop_listening import StopListening
class Pyromod(
Ask,
GetListenerMatchingWithData,
GetListenerMatchingWithIdentifierPattern,
GetManyListenersMatchingWithData,
GetManyListenersMatchingWithIdentifierPattern,
Listen,
RegisterNextStepHandler,
RemoveListener,
StopListener,
StopListening
):
pass

76
pyrogram/methods/pyromod/ask.py Normal file
View file

@ -0,0 +1,76 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from pyrogram.filters import Filter
from typing import List, Optional, Union
from pyrogram.types import ListenerTypes
class Ask:
async def ask(
self: "pyrogram.Client",
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

47
pyrogram/methods/pyromod/get_listener_matching_with_data.py Normal file
View file

@ -0,0 +1,47 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from typing import Optional
from pyrogram.types import ListenerTypes, Identifier, Listener
class GetListenerMatchingWithData:
def get_listener_matching_with_data(
self: "pyrogram.Client",
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)

52
pyrogram/methods/pyromod/get_listener_matching_with_identifier_pattern.py Normal file
View file

@ -0,0 +1,52 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from typing import Optional
from pyrogram.types import ListenerTypes, Identifier, Listener
class GetListenerMatchingWithIdentifierPattern:
def get_listener_matching_with_identifier_pattern(
self: "pyrogram.Client",
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)

42
pyrogram/methods/pyromod/get_many_listeners_matching_with_data.py Normal file
View file

@ -0,0 +1,42 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from typing import List
from pyrogram.types import ListenerTypes, Identifier, Listener
class GetManyListenersMatchingWithData:
def get_many_listeners_matching_with_data(
self: "pyrogram.Client",
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

42
pyrogram/methods/pyromod/get_many_listeners_matching_with_identifier_pattern.py Normal file
View file

@ -0,0 +1,42 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from typing import List
from pyrogram.types import ListenerTypes, Identifier, Listener
class GetManyListenersMatchingWithIdentifierPattern:
def get_many_listeners_matching_with_identifier_pattern(
self: "pyrogram.Client",
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

88
pyrogram/methods/pyromod/listen.py Normal file
View file

@ -0,0 +1,88 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import inspect
import pyrogram
from pyrogram.errors import ListenerTimeout
from pyrogram.filters import Filter
from typing import List, Optional, Union
from pyrogram.types import ListenerTypes, Identifier, Listener
from pyrogram.utils import PyromodConfig
class Listen:
async def listen(
self: "pyrogram.Client",
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)

66
pyrogram/methods/pyromod/register_next_step_handler.py Normal file
View file

@ -0,0 +1,66 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from pyrogram.filters import Filter
from typing import Callable, List, Optional, Union
from pyrogram.types import ListenerTypes, Identifier, Listener
class RegisterNextStepHandler:
def register_next_step_handler(
self: "pyrogram.Client",
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)

37
pyrogram/methods/pyromod/remove_listerner.py Normal file
View file

@ -0,0 +1,37 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from pyrogram.types import Listener
class RemoveListener:
def remove_listener(
self: "pyrogram.Client",
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

52
pyrogram/methods/pyromod/stop_listener.py Normal file
View file

@ -0,0 +1,52 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import inspect
import pyrogram
from pyrogram.errors import ListenerStopped
from pyrogram.types import Listener
from pyrogram.utils import PyromodConfig
class StopListener:
async def stop_listener(
self: "pyrogram.Client",
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())

56
pyrogram/methods/pyromod/stop_listening.py Normal file
View file

@ -0,0 +1,56 @@
# Pyrofork - Telegram MTProto API Client Library for Python
# Copyright (C) 2020 Cezar H. <https://github.com/usernein>
# Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
#
# This file is part of Pyrofork.
#
# Pyrofork 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.
#
# Pyrofork 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 Pyrofork. If not, see <http://www.gnu.org/licenses/>.
import pyrogram
from typing import List, Union
from pyrogram.types import ListenerTypes, Identifier
class StopListening:
async def stop_listening(
self: "pyrogram.Client",
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)