Nextcloud Talk: Conversations Avatar API (#133)

Relatively small and simple API.

Signed-off-by: Alexander Piskun <[email protected]>

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
 

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":

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):

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:

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)