Source code for nc_py_api.talk_bot

"""Nextcloud Talk API for bots."""

import dataclasses
import hashlib
import hmac
import json
import os
import typing

import httpx

from . import options
from ._misc import random_string
from ._session import BasicConfig
from .nextcloud import AsyncNextcloudApp, NextcloudApp


[docs] class ObjectContent(typing.TypedDict): """Object content of :py:class:`~nc_py_api.talk_bot.TalkBotMessage`.""" message: str parameters: dict
[docs] @dataclasses.dataclass class TalkBotMessage: """Talk message received by bots.""" def __init__(self, raw_data: dict): self._raw_data = raw_data @property def message_type(self) -> str: """The type of message like Join, Leave, Create, Activity, etc.""" return self._raw_data["type"] @property def actor_id(self) -> str: """One of the attendee types followed by the ``/`` character and a unique identifier within the given type. For the users it is the Nextcloud user ID, for guests a **sha1** value. """ return self._raw_data["actor"]["id"] @property def actor_display_name(self) -> str: """The display name of the attendee sending the message.""" return self._raw_data["actor"]["name"] @property def object_id(self) -> int: """The message ID of the given message on the origin server. It can be used to react or reply to the given message. """ return self._raw_data["object"]["id"] @property def object_name(self) -> str: """For normal written messages ``message``, otherwise one of the known ``system message identifiers``.""" return self._raw_data["object"]["name"] @property def object_content(self) -> ObjectContent: """Dictionary with a ``message`` and ``parameters`` keys.""" return json.loads(self._raw_data["object"]["content"]) @property def object_media_type(self) -> str: """``text/markdown`` when the message should be interpreted as **Markdown**, otherwise ``text/plain``.""" return self._raw_data["object"]["mediaType"] @property def conversation_token(self) -> str: """The token of the conversation in which the message was posted. It can be used to react or reply to the given message. """ return self._raw_data["target"]["id"] @property def conversation_name(self) -> str: """The name of the conversation in which the message was posted.""" return self._raw_data["target"]["name"] def __repr__(self): return f"<{self.__class__.__name__} conversation={self.conversation_name}, actor={self.actor_display_name}>"
[docs] class TalkBot: """A class that implements the TalkBot functionality.""" _ep_base: str = "/ocs/v2.php/apps/spreed/api/v1/bot" def __init__(self, callback_url: str, display_name: str, description: str = ""): """Class implementing Nextcloud Talk Bot functionality. :param callback_url: FastAPI endpoint which will be assigned to bot. :param display_name: The display name of the bot that is shown as author when it posts a message or reaction. :param description: Description of the bot helping moderators to decide if they want to enable this bot. """ self.callback_url = callback_url.lstrip("/") self.display_name = display_name self.description = description
[docs] def enabled_handler(self, enabled: bool, nc: NextcloudApp) -> None: """Handles the app ``on``/``off`` event in the context of the bot. :param enabled: Value that was passed to ``/enabled`` handler. :param nc: **NextcloudApp** class that was passed ``/enabled`` handler. """ if enabled: bot_id, bot_secret = nc.register_talk_bot(self.callback_url, self.display_name, self.description) os.environ[bot_id] = bot_secret else: nc.unregister_talk_bot(self.callback_url)
[docs] def send_message( self, message: str, reply_to_message: int | TalkBotMessage, silent: bool = False, token: str = "" ) -> tuple[httpx.Response, str]: """Send a message and returns a "reference string" to identify the message again in a "get messages" request. :param message: The message to say. :param reply_to_message: The message ID this message is a reply to. .. note:: Only allowed when the message type is not ``system`` or ``command``. :param silent: Flag controlling if the message should create a chat notifications for the users. :param token: Token of the conversation. Can be empty if ``reply_to_message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :returns: Tuple, where fist element is :py:class:`httpx.Response` and second is a "reference string". :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(reply_to_message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") token = reply_to_message.conversation_token if isinstance(reply_to_message, TalkBotMessage) else token reference_id = hashlib.sha256(random_string(32).encode("UTF-8")).hexdigest() params = { "message": message, "replyTo": reply_to_message.object_id if isinstance(reply_to_message, TalkBotMessage) else reply_to_message, "referenceId": reference_id, "silent": silent, } return self._sign_send_request("POST", f"/{token}/message", params, message), reference_id
[docs] def react_to_message(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """React to a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to react to. :param reaction: A single emoji. :param token: Token of the conversation. Can be empty if ``message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") message_id = message.object_id if isinstance(message, TalkBotMessage) else message token = message.conversation_token if isinstance(message, TalkBotMessage) else token params = { "reaction": reaction, } return self._sign_send_request("POST", f"/{token}/reaction/{message_id}", params, reaction)
[docs] def delete_reaction(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """Removes reaction from a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to remove reaction from. :param reaction: A single emoji. :param token: Token of the conversation. Can be empty if ``message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") message_id = message.object_id if isinstance(message, TalkBotMessage) else message token = message.conversation_token if isinstance(message, TalkBotMessage) else token params = { "reaction": reaction, } return self._sign_send_request("DELETE", f"/{token}/reaction/{message_id}", params, reaction)
def _sign_send_request(self, method: str, url_suffix: str, data: dict, data_to_sign: str) -> httpx.Response: secret = get_bot_secret(self.callback_url) if secret is None: raise RuntimeError("Can't find the 'secret' of the bot. Has the bot been installed?") talk_bot_random = random_string(32) hmac_sign = hmac.new(secret, talk_bot_random.encode("UTF-8"), digestmod=hashlib.sha256) hmac_sign.update(data_to_sign.encode("UTF-8")) nc_app_cfg = BasicConfig() with httpx.Client(verify=nc_app_cfg.options.nc_cert) as client: return client.request( method, url=nc_app_cfg.endpoint + "/ocs/v2.php/apps/spreed/api/v1/bot" + url_suffix, json=data, headers={ "X-Nextcloud-Talk-Bot-Random": talk_bot_random, "X-Nextcloud-Talk-Bot-Signature": hmac_sign.hexdigest(), "OCS-APIRequest": "true", }, cookies={"XDEBUG_SESSION": options.XDEBUG_SESSION} if options.XDEBUG_SESSION else {}, timeout=nc_app_cfg.options.timeout, )
class AsyncTalkBot: """A class that implements the async TalkBot functionality.""" _ep_base: str = "/ocs/v2.php/apps/spreed/api/v1/bot" def __init__(self, callback_url: str, display_name: str, description: str = ""): """Class implementing Nextcloud Talk Bot functionality. :param callback_url: FastAPI endpoint which will be assigned to bot. :param display_name: The display name of the bot that is shown as author when it posts a message or reaction. :param description: Description of the bot helping moderators to decide if they want to enable this bot. """ self.callback_url = callback_url.lstrip("/") self.display_name = display_name self.description = description async def enabled_handler(self, enabled: bool, nc: AsyncNextcloudApp) -> None: """Handles the app ``on``/``off`` event in the context of the bot. :param enabled: Value that was passed to ``/enabled`` handler. :param nc: **NextcloudApp** class that was passed ``/enabled`` handler. """ if enabled: bot_id, bot_secret = await nc.register_talk_bot(self.callback_url, self.display_name, self.description) os.environ[bot_id] = bot_secret else: await nc.unregister_talk_bot(self.callback_url) async def send_message( self, message: str, reply_to_message: int | TalkBotMessage, silent: bool = False, token: str = "" ) -> tuple[httpx.Response, str]: """Send a message and returns a "reference string" to identify the message again in a "get messages" request. :param message: The message to say. :param reply_to_message: The message ID this message is a reply to. .. note:: Only allowed when the message type is not ``system`` or ``command``. :param silent: Flag controlling if the message should create a chat notifications for the users. :param token: Token of the conversation. Can be empty if ``reply_to_message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :returns: Tuple, where fist element is :py:class:`httpx.Response` and second is a "reference string". :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(reply_to_message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") token = reply_to_message.conversation_token if isinstance(reply_to_message, TalkBotMessage) else token reference_id = hashlib.sha256(random_string(32).encode("UTF-8")).hexdigest() params = { "message": message, "replyTo": reply_to_message.object_id if isinstance(reply_to_message, TalkBotMessage) else reply_to_message, "referenceId": reference_id, "silent": silent, } return await self._sign_send_request("POST", f"/{token}/message", params, message), reference_id async def react_to_message(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """React to a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to react to. :param reaction: A single emoji. :param token: Token of the conversation. Can be empty if ``message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") message_id = message.object_id if isinstance(message, TalkBotMessage) else message token = message.conversation_token if isinstance(message, TalkBotMessage) else token params = { "reaction": reaction, } return await self._sign_send_request("POST", f"/{token}/reaction/{message_id}", params, reaction) async def delete_reaction(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """Removes reaction from a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to remove reaction from. :param reaction: A single emoji. :param token: Token of the conversation. Can be empty if ``message`` is :py:class:`~nc_py_api.talk_bot.TalkBotMessage`. :raises ValueError: in case of an invalid usage. :raises RuntimeError: in case of a broken installation. """ if not token and not isinstance(message, TalkBotMessage): raise ValueError("Either specify 'token' value or provide 'TalkBotMessage'.") message_id = message.object_id if isinstance(message, TalkBotMessage) else message token = message.conversation_token if isinstance(message, TalkBotMessage) else token params = { "reaction": reaction, } return await self._sign_send_request("DELETE", f"/{token}/reaction/{message_id}", params, reaction) async def _sign_send_request(self, method: str, url_suffix: str, data: dict, data_to_sign: str) -> httpx.Response: secret = await aget_bot_secret(self.callback_url) if secret is None: raise RuntimeError("Can't find the 'secret' of the bot. Has the bot been installed?") talk_bot_random = random_string(32) hmac_sign = hmac.new(secret, talk_bot_random.encode("UTF-8"), digestmod=hashlib.sha256) hmac_sign.update(data_to_sign.encode("UTF-8")) nc_app_cfg = BasicConfig() async with httpx.AsyncClient(verify=nc_app_cfg.options.nc_cert) as aclient: return await aclient.request( method, url=nc_app_cfg.endpoint + "/ocs/v2.php/apps/spreed/api/v1/bot" + url_suffix, json=data, headers={ "X-Nextcloud-Talk-Bot-Random": talk_bot_random, "X-Nextcloud-Talk-Bot-Signature": hmac_sign.hexdigest(), "OCS-APIRequest": "true", }, cookies={"XDEBUG_SESSION": options.XDEBUG_SESSION} if options.XDEBUG_SESSION else {}, timeout=nc_app_cfg.options.timeout, ) def __get_bot_secret(callback_url: str) -> str: sha_1 = hashlib.sha1(usedforsecurity=False) string_to_hash = os.environ["APP_ID"] + "_" + callback_url.lstrip("/") sha_1.update(string_to_hash.encode("UTF-8")) return sha_1.hexdigest()
[docs] def get_bot_secret(callback_url: str) -> bytes | None: """Returns the bot's secret from an environment variable or from the application's configuration on the server.""" secret_key = __get_bot_secret(callback_url) if secret_key in os.environ: return os.environ[secret_key].encode("UTF-8") secret_value = NextcloudApp().appconfig_ex.get_value(secret_key) if secret_value is not None: os.environ[secret_key] = secret_value return secret_value.encode("UTF-8") return None
async def aget_bot_secret(callback_url: str) -> bytes | None: """Returns the bot's secret from an environment variable or from the application's configuration on the server.""" secret_key = __get_bot_secret(callback_url) if secret_key in os.environ: return os.environ[secret_key].encode("UTF-8") secret_value = await AsyncNextcloudApp().appconfig_ex.get_value(secret_key) if secret_value is not None: os.environ[secret_key] = secret_value return secret_value.encode("UTF-8") return None