import datetime
import logging
from multiprocessing import Queue as PQueue
from queue import Empty, Queue as TQueue
from typing import Dict, List, Optional, Union
from uuid import UUID
from platypush.plugins import RunnablePlugin, action
from platypush.plugins.chat import ChatPlugin
from platypush.schemas.telegram import (
TelegramChatSchema,
TelegramFileSchema,
)
from platypush.utils import wait_for_either
from ._bridge import ResultBridge
from ._model import Command, Resource, END_OF_SERVICE
from ._service import TelegramService
from ._utils import dump_msg, dump_user
[docs]
class TelegramPlugin(ChatPlugin, RunnablePlugin):
"""
Plugin to programmatically send Telegram messages through a Telegram bot.
In order to send messages to contacts, groups or channels you'll first need
to register a bot. To do so:
1. Open a Telegram conversation with the `@BotFather
<https://telegram.me/BotFather>`_.
2. Send ``/start`` followed by ``/newbot``. Choose a display name and a
username for your bot.
3. Copy the provided API token in the configuration of this plugin.
4. Open a conversation with your newly created bot.
"""
_DEFAULT_TIMEOUT = 20
[docs]
def __init__(
self,
api_token: str,
authorized_chat_ids: Optional[List[Union[str, int]]] = None,
**kwargs,
):
"""
:param api_token: API token as returned by the `@BotFather
<https://telegram.me/BotFather>`_
:param authorized_chat_ids: Optional list of chat_id/user_id which are
authorized to send messages to the bot. If nothing is specified
then no restrictions are applied.
"""
super().__init__(**kwargs)
self._authorized_chat_ids = set(authorized_chat_ids or [])
self._api_token = api_token
self._cmd_queue = PQueue()
self._result_queue = PQueue()
self._service: Optional[TelegramService] = None
self._response_queues: Dict[UUID, TQueue] = {}
self._result_bridge: Optional[ResultBridge] = None
# Set httpx logging to WARNING, or it will log a line for every request
logging.getLogger("httpx").setLevel(logging.WARNING)
def _exec(
self, cmd: str, *args, timeout: Optional[float] = _DEFAULT_TIMEOUT, **kwargs
):
assert self._service, "Telegram service not running"
cmd_obj = Command(cmd, args=args, kwargs=kwargs, timeout=timeout)
self._response_queues[cmd_obj.id] = TQueue()
self._cmd_queue.put_nowait(cmd_obj)
try:
result = self._response_queues[cmd_obj.id].get(timeout=timeout)
except (TimeoutError, Empty) as e:
raise TimeoutError(f"Timeout while executing command {cmd}") from e
finally:
self._response_queues.pop(cmd_obj.id, None)
assert not isinstance(
result, Exception
), f'Error while executing command {cmd}: {result}'
return result
@property
def response_queues(self):
return self._response_queues
@property
def result_queue(self):
return self._result_queue
[docs]
@action
def send_message( # pylint: disable=arguments-differ
self,
chat_id: Union[str, int],
text: str,
*_,
parse_mode: Optional[str] = None,
disable_web_page_preview: bool = False,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
**__,
) -> dict:
"""
Send a message to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param text: Text to be sent.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_web_page_preview: If True then web previews for URLs
will be disabled.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Request timeout (default: 20 seconds).
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_message',
chat_id=chat_id,
text=text,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
)
)
[docs]
@action
def send_photo(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
caption: Optional[str] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send a picture to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param caption: Optional caption for the picture.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_photo',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='photo',
caption=caption,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_audio(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
caption: Optional[str] = None,
performer: Optional[str] = None,
title: Optional[str] = None,
duration: Optional[float] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send audio to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param caption: Optional caption for the picture.
:param performer: Optional audio performer.
:param title: Optional audio title.
:param duration: Duration of the audio in seconds.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_audio',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='audio',
caption=caption,
disable_notification=disable_notification,
performer=performer,
title=title,
duration=duration,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_document(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
filename: Optional[str] = None,
caption: Optional[str] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send a document to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param filename: Name of the file as it will be shown in Telegram.
:param caption: Optional caption for the picture.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_document',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='document',
filename=filename,
caption=caption,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_video(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
duration: Optional[int] = None,
caption: Optional[str] = None,
width: Optional[int] = None,
height: Optional[int] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send a video to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param duration: Duration in seconds.
:param caption: Optional caption for the picture.
:param width: Video width.
:param height: Video height.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_video',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='video',
duration=duration,
caption=caption,
width=width,
height=height,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_animation(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
duration: Optional[int] = None,
caption: Optional[str] = None,
width: Optional[int] = None,
height: Optional[int] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send an animation (GIF or H.264/MPEG-4 AVC video without sound) to a
chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param duration: Duration in seconds.
:param caption: Optional caption for the picture.
:param width: Video width.
:param height: Video height.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_animation',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='animation',
duration=duration,
caption=caption,
width=width,
height=height,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_voice(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
caption: Optional[str] = None,
duration: Optional[float] = None,
parse_mode: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send audio to a chat as a voice file. For this to work, your audio must
be in an .ogg file encoded with OPUS (other formats may be sent as
Audio or Document).
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or ``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param caption: Optional caption for the picture.
:param duration: Duration of the voice in seconds.
:param parse_mode: Set to 'Markdown' or 'HTML' to send either Markdown
or HTML content.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_voice',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='voice',
caption=caption,
disable_notification=disable_notification,
duration=duration,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
parse_mode=parse_mode,
)
)
[docs]
@action
def send_video_note(
self,
chat_id: Union[str, int],
file_id: Optional[int] = None,
url: Optional[str] = None,
path: Optional[str] = None,
duration: Optional[int] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send a video note to a chat. As of v.4.0, Telegram clients support
rounded square mp4 videos of up to 1 minute long.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param file_id: Set it if the file already exists on Telegram servers
and has a file_id. Note that you'll have to specify either
``file_id``, ``url`` or ``path``.
:param url: Set it if you want to send a file from a remote URL. Note
that you'll have to specify either ``file_id``, ``url`` or
``path``.
:param path: Set it if you want to send a file from the local
filesystem. Note that you'll have to specify either ``file_id``,
``url`` or ``path``.
:param duration: Duration in seconds.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_video_note',
chat_id=chat_id,
resource=Resource(file_id=file_id, url=url, path=path),
resource_attr='video',
duration=duration,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
)
)
[docs]
@action
def send_location(
self,
chat_id: Union[str, int],
latitude: float,
longitude: float,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send a location to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param latitude: Latitude
:param longitude: Longitude
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_location',
chat_id=chat_id,
latitude=latitude,
longitude=longitude,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
)
)
[docs]
@action
def send_venue(
self,
chat_id: Union[str, int],
latitude: float,
longitude: float,
title: str,
address: str,
foursquare_id: Optional[str] = None,
foursquare_type: Optional[str] = None,
disable_notification: bool = False,
reply_to_message_id: Optional[int] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Send the address of a venue to a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param latitude: Latitude
:param longitude: Longitude
:param title: Venue name.
:param address: Venue address.
:param foursquare_id: Foursquare ID.
:param foursquare_type: Foursquare type.
:param disable_notification: If True then no notification will be sent
to the users.
:param reply_to_message_id: If set then the message will be sent as a
response to the specified message.
:param timeout: Upload timeout (default: 20 seconds)
:return: .. schema:: telegram.TelegramMessageSchema
"""
return dump_msg(
self._exec(
'send_venue',
chat_id=chat_id,
latitude=latitude,
longitude=longitude,
title=title,
address=address,
foursquare_id=foursquare_id,
foursquare_type=foursquare_type,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
timeout=timeout,
)
)
[docs]
@action
def get_file(self, file_id: str, timeout: Optional[float] = None) -> dict:
"""
Get the info and URL of an uploaded file by file_id.
:param file_id: File ID.
:param timeout: Upload timeout (default: 20 seconds).
:return: .. schema:: telegram.TelegramFileSchema
"""
return dict(
TelegramFileSchema().dump(self._exec('get_file', file_id, timeout=timeout))
)
[docs]
@action
def get_chat(
self, chat_id: Union[int, str], timeout: Optional[float] = _DEFAULT_TIMEOUT
) -> dict:
"""
Get the info about a Telegram chat.
:param chat_id: Chat ID.
:param timeout: Upload timeout (default: 20 seconds).
"""
return dict(
TelegramChatSchema().dump(self._exec('get_chat', chat_id, timeout=timeout))
)
[docs]
@action
def get_chat_user(
self,
chat_id: Union[int, str],
user_id: int,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
) -> dict:
"""
Get the info about a user connected to a chat.
:param chat_id: Chat ID.
:param user_id: User ID.
:param timeout: Upload timeout (default: 20 seconds).
"""
return dump_user(
self._exec('get_chat_member', chat_id, user_id, timeout=timeout)
)
[docs]
@action
def get_chat_administrators(
self, chat_id: Union[int, str], timeout: Optional[float] = _DEFAULT_TIMEOUT
) -> List[dict]:
"""
Get the list of the administrators of a chat.
:param chat_id: Chat ID.
:param timeout: Upload timeout (default: 20 seconds).
:return: .. schema:: telegram.TelegramUserSchema(many=True)
"""
return [
dump_user(user.user)
for user in self._exec('get_chat_administrators', chat_id, timeout=timeout)
]
[docs]
@action
def get_chat_members_count(
self, chat_id: Union[int, str], timeout: Optional[float] = _DEFAULT_TIMEOUT
) -> int:
"""
Get the number of users in a chat.
:param chat_id: Chat ID.
:param timeout: Upload timeout (default: 20 seconds).
"""
return self._exec('get_chat_members_count', chat_id, timeout=timeout)
[docs]
@action
def kick_chat_member(
self,
chat_id: Union[str, int],
user_id: int,
until_date: Optional[datetime.datetime] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Kick a user from a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param user_id: Unique user ID.
:param until_date: End date for the ban.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'kick_chat_member',
chat_id=chat_id,
user_id=user_id,
until_date=until_date,
timeout=timeout,
)
[docs]
@action
def unban_chat_member(
self,
chat_id: Union[str, int],
user_id: int,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Lift the ban from a chat member.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param user_id: Unique user ID.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'unban_chat_member', chat_id=chat_id, user_id=user_id, timeout=timeout
)
[docs]
@action
def set_chat_title(
self,
chat_id: Union[str, int],
title: str,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Set the title of a channel/group.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param title: New chat title.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'set_chat_title', chat_id=chat_id, description=title, timeout=timeout
)
[docs]
@action
def set_chat_description(
self,
chat_id: Union[str, int],
description: str,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Set the description of a channel/group.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param description: New chat description.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'set_chat_description',
chat_id=chat_id,
description=description,
timeout=timeout,
)
[docs]
@action
def set_chat_photo(
self,
chat_id: Union[str, int],
path: str,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Set the photo of a channel/group.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param path: Path of the new image.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'set_chat_photo',
chat_id=chat_id,
resource=Resource(path=path),
resource_attr='photo',
timeout=timeout,
)
[docs]
@action
def delete_chat_photo(
self, chat_id: Union[str, int], timeout: Optional[float] = _DEFAULT_TIMEOUT
):
"""
Delete the photo of a channel/group.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec('delete_chat_photo', chat_id=chat_id, timeout=timeout)
[docs]
@action
def pin_chat_message(
self,
chat_id: Union[str, int],
message_id: int,
disable_notification: Optional[bool] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
):
"""
Pin a message in a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param message_id: Message ID.
:param disable_notification: If True then no notification will be sent
to the users.
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec(
'pin_chat_message',
chat_id=chat_id,
message_id=message_id,
disable_notification=disable_notification,
timeout=timeout,
)
[docs]
@action
def unpin_chat_message(
self, chat_id: Union[str, int], timeout: Optional[float] = _DEFAULT_TIMEOUT
):
"""
Unpin the message of a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``. In order to get your own
Telegram chat_id open a conversation with `@IDBot
<https://telegram.me/IDBot>`_ and type ``/start`` followed by
``/getid``. Similar procedures also exist to get a group or channel
chat_id - just Google for "Telegram get channel/group chat_id".
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec('unpin_chat_message', chat_id=chat_id, timeout=timeout)
[docs]
@action
def leave_chat(
self, chat_id: Union[str, int], timeout: Optional[float] = _DEFAULT_TIMEOUT
):
"""
Leave a chat.
:param chat_id: Chat ID. Can be either a numerical ID or a unique
identifier in the format ``@channelname``.
In order to get your own Telegram chat_id open a conversation with
`@IDBot <https://telegram.me/IDBot>`_ and type ``/start`` followed
by ``/getid``. Similar procedures also exist to get a group or
channel chat_id - just Google for "Telegram get channel/group
chat_id".
:param timeout: Request timeout (default: 20 seconds)
"""
self._exec('leave_chat', chat_id=chat_id, timeout=timeout)
[docs]
def main(self):
self._result_bridge = ResultBridge(self)
self._result_bridge.start()
while not self.should_stop():
self._service = TelegramService(
self._api_token,
cmd_queue=self._cmd_queue,
result_queue=self._result_queue,
authorized_chat_ids=self._authorized_chat_ids,
)
self._service.start()
wait_for_either(self._should_stop, self._service.stop_event)
self.wait_stop(10)
[docs]
def stop(self):
# Send an END_OF_SERVICE to the result queue to signal that the service
# is stopping
self._cmd_queue.put_nowait(END_OF_SERVICE)
if self._service:
self._service.stop()
self._service = None
if self._result_bridge and self._result_bridge.is_alive():
self._result_bridge.join()
self._result_bridge = None
super().stop()
# vim:sw=4:ts=4:et:
