From 76cdf8095b41cbdb01146713a6d620852425ed5d Mon Sep 17 00:00:00 2001 From: Alexander Piskun Date: Tue, 26 Sep 2023 19:55:19 +0300 Subject: [PATCH] Nextcloud Talk: Conversations Avatar API Signed-off-by: Alexander Piskun --- CHANGELOG.md | 3 +- nc_py_api/_session.py | 3 ++ nc_py_api/_talk_api.py | 49 +++++++++++++++++++++++++++++++++ nc_py_api/options.py | 2 +- tests/actual_tests/talk_test.py | 30 ++++++++++++++++++++ 5 files changed, 85 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a99f540..36580db4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,13 +2,14 @@ All notable changes to this project will be documented in this file. -## [0.2.2 - 2023-09-2x] +## [0.2.2 - 2023-09-26] ### Added - FilesAPI: [Chunked v2 upload](https://docs.nextcloud.com/server/latest/developer_manual/client_apis/WebDAV/chunking.html#chunked-upload-v2) support, enabled by default. - New option to disable `chunked v2 upload` if there is a need for that: `CHUNKED_UPLOAD_V2` - TalkAPI: Poll API support(create_poll, get_poll, vote_poll, close_poll). +- TalkAPI: Conversation avatar API(get_conversation_avatar, set_conversation_avatar, delete_conversation_avatar) ### Changed diff --git a/nc_py_api/_session.py b/nc_py_api/_session.py index 8ab790eb..917c1218 100644 --- a/nc_py_api/_session.py +++ b/nc_py_api/_session.py @@ -190,6 +190,7 @@ def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[byte url_params = f"{self.cfg.endpoint}{path_params}" info = f"request: method={method}, url={url_params}" nested_req = kwargs.pop("nested_req", False) + not_parse = kwargs.pop("not_parse", False) try: timeout = kwargs.pop("timeout", self.cfg.options.timeout) if method == "GET": @@ -203,6 +204,8 @@ def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[byte self.response_headers = response.headers check_error(response.status_code, info) + if not_parse: + return response response_data = loads(response.text) ocs_meta = response_data["ocs"]["meta"] if ocs_meta["status"] != "ok": diff --git a/nc_py_api/_talk_api.py b/nc_py_api/_talk_api.py index 84192dd6..9eb61951 100644 --- a/nc_py_api/_talk_api.py +++ b/nc_py_api/_talk_api.py @@ -463,6 +463,55 @@ def close_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[C token = conversation.token if isinstance(conversation, Conversation) else conversation return Poll(self._session.ocs("DELETE", self._ep_base + f"/api/v1/poll/{token}/{poll_id}"), token) + def set_conversation_avatar( + self, + conversation: typing.Union[Conversation, str], + avatar: typing.Union[bytes, tuple[str, typing.Union[str, None]]], + ) -> Conversation: + """Set image or emoji as avatar for the conversation. + + :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. + :param avatar: Squared image with mimetype equal to PNG or JPEG or a tuple with emoji and optional + HEX color code(6 times ``0-9A-F``) without the leading ``#`` character. + + .. note:: Color omit to fallback to the default bright/dark mode icon background color. + """ + require_capabilities("spreed.features.avatar", self._session.capabilities) + token = conversation.token if isinstance(conversation, Conversation) else conversation + if isinstance(avatar, bytes): + r = self._session.ocs("POST", self._ep_base + f"/api/v1/room/{token}/avatar", files={"file": avatar}) + else: + r = self._session.ocs( + "POST", + self._ep_base + f"/api/v1/room/{token}/avatar/emoji", + json={ + "emoji": avatar[0], + "color": avatar[1], + }, + ) + return Conversation(r) + + def delete_conversation_avatar(self, conversation: typing.Union[Conversation, str]) -> Conversation: + """Delete conversation avatar. + + :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. + """ + require_capabilities("spreed.features.avatar", self._session.capabilities) + token = conversation.token if isinstance(conversation, Conversation) else conversation + return Conversation(self._session.ocs("DELETE", self._ep_base + f"/api/v1/room/{token}/avatar")) + + def get_conversation_avatar(self, conversation: typing.Union[Conversation, str], dark=False) -> bytes: + """Get conversation avatar (binary). + + :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. + :param dark: boolean indicating should be or not avatar fetched for dark theme. + """ + require_capabilities("spreed.features.avatar", self._session.capabilities) + token = conversation.token if isinstance(conversation, Conversation) else conversation + ep_suffix = "/dark" if dark else "" + response = self._session.ocs("GET", self._ep_base + f"/api/v1/room/{token}/avatar" + ep_suffix, not_parse=True) + return response.content + @staticmethod def _get_token(message: typing.Union[TalkMessage, str], conversation: typing.Union[Conversation, str]) -> str: if not conversation and not isinstance(message, TalkMessage): diff --git a/nc_py_api/options.py b/nc_py_api/options.py index fd786237..09ac8cc1 100644 --- a/nc_py_api/options.py +++ b/nc_py_api/options.py @@ -39,7 +39,7 @@ elif str_val.lower() not in ("true", "1"): NPA_NC_CERT = str_val -CHUNKED_UPLOAD_V2 = True +CHUNKED_UPLOAD_V2 = environ.get("CHUNKED_UPLOAD_V2", True) """Option to enable/disable **version 2** chunked upload(better Object Storages support). Additional information can be found in Nextcloud documentation: diff --git a/tests/actual_tests/talk_test.py b/tests/actual_tests/talk_test.py index 767492f6..29ff0d5b 100644 --- a/tests/actual_tests/talk_test.py +++ b/tests/actual_tests/talk_test.py @@ -1,6 +1,8 @@ +from io import BytesIO from os import environ import pytest +from PIL import Image from nc_py_api import Nextcloud, talk, talk_bot @@ -322,3 +324,31 @@ def test_vote_poll(nc_any): assert isinstance(poll.details[0].actor_display_name, str) finally: nc_any.talk.delete_conversation(conversation) + + +@pytest.mark.require_nc(major=27) +def test_conversation_avatar(nc_any): + if nc_any.talk.available is False: + pytest.skip("Nextcloud Talk is not installed") + + conversation = nc_any.talk.create_conversation(talk.ConversationType.GROUP, "admin") + try: + assert conversation.is_custom_avatar is False + r = nc_any.talk.get_conversation_avatar(conversation) + assert isinstance(r, bytes) + im = Image.effect_mandelbrot((512, 512), (-3, -2.5, 2, 2.5), 100) + buffer = BytesIO() + im.save(buffer, format="PNG") + buffer.seek(0) + r = nc_any.talk.set_conversation_avatar(conversation, buffer.read()) + assert r.is_custom_avatar is True + r = nc_any.talk.get_conversation_avatar(conversation) + assert isinstance(r, bytes) + r = nc_any.talk.delete_conversation_avatar(conversation) + assert r.is_custom_avatar is False + r = nc_any.talk.set_conversation_avatar(conversation, ("🫡", None)) + assert r.is_custom_avatar is True + r = nc_any.talk.get_conversation_avatar(conversation, dark=True) + assert isinstance(r, bytes) + finally: + nc_any.talk.delete_conversation(conversation)