TalkAPI: added send_file; receive_messages can return TalkFileMessage (#135)

Added simple APIs that will make life much easier for chats: sending a
file/directory and converting a text message containing the
file/directory directly into FsNode

Later will write and examples and add them to documentation, but these
API are really simple.

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

Author: bigcat88

CHANGELOG.md

@@ -6,6 +6,9 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
+- TalkAPI:
+  * `send_file` to easy send `FsNode` to Talk chat.
+  * `receive_messages` can return the `TalkFileMessage` subclass of usual `TalkMessage` with additional functionality.
 - NextcloudApp: The `ex_app.verify_version` function to simply check whether the application has been updated.
 
 ### Changed

docs/reference/ExApp.rst

@@ -1,7 +1,7 @@
 .. py:currentmodule:: nc_py_api.ex_app
 
-AppAPI Application
-==================
+External Application
+====================
 
 Constants
 ---------

docs/reference/Talk.rst

@@ -9,6 +9,10 @@ Talk API
     :members:
     :inherited-members:
 
+.. autoclass:: nc_py_api.talk.TalkFileMessage
+    :members:
+    :inherited-members:
+
 .. autoclass:: nc_py_api._talk_api._TalkAPI
     :members:
 

nc_py_api/_talk_api.py

@@ -10,6 +10,7 @@
     require_capabilities,
 )
 from ._session import NcSessionBasic
+from .files import FsNode, Share, ShareType
 from .talk import (
     BotInfo,
     BotInfoBasic,
@@ -18,6 +19,7 @@
     MessageReactions,
     NotificationLevel,
     Poll,
+    TalkFileMessage,
     TalkMessage,
 )
 
@@ -191,7 +193,7 @@ def delete_conversation(self, conversation: typing.Union[Conversation, str]) ->
         """Deletes a conversation.
 
         .. note:: Deleting a conversation that is the parent of breakout rooms, will also delete them.
-            ``ONE_TO_ONE`` conversations can not be deleted for them
+            ``ONE_TO_ONE`` conversations cannot be deleted for them
             :py:class:`~nc_py_api._talk_api._TalkAPI.leave_conversation` should be used.
 
         :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`.
@@ -244,6 +246,23 @@ def send_message(
         r = self._session.ocs("POST", self._ep_base + f"/api/v1/chat/{token}", json=params)
         return TalkMessage(r)
 
+    def send_file(
+        self,
+        path: typing.Union[str, FsNode],
+        conversation: typing.Union[Conversation, str] = "",
+    ) -> tuple[Share, str]:
+        require_capabilities("files_sharing.api_enabled", self._session.capabilities)
+        token = conversation.token if isinstance(conversation, Conversation) else conversation
+        reference_id = hashlib.sha256(random_string(32).encode("UTF-8")).hexdigest()
+        params = {
+            "shareType": ShareType.TYPE_ROOM,
+            "shareWith": token,
+            "path": path.user_path if isinstance(path, FsNode) else path,
+            "referenceId": reference_id,
+        }
+        r = self._session.ocs("POST", "/ocs/v1.php/apps/files_sharing/api/v1/shares", json=params)
+        return Share(r), reference_id
+
     def receive_messages(
         self,
         conversation: typing.Union[Conversation, str],
@@ -268,7 +287,7 @@ def receive_messages(
             "noStatusUpdate": int(no_status_update),
         }
         result = self._session.ocs("GET", self._ep_base + f"/api/v1/chat/{token}", params=params)
-        return [TalkMessage(i) for i in result]
+        return [TalkFileMessage(i, self._session.user) if i["message"] == "{file}" else TalkMessage(i) for i in result]
 
     def delete_message(
         self, message: typing.Union[TalkMessage, str], conversation: typing.Union[Conversation, str] = ""

nc_py_api/ex_app/ui/files.py

@@ -8,7 +8,7 @@
 from ..._exceptions import NextcloudExceptionNotFound
 from ..._misc import require_capabilities
 from ..._session import NcSessionApp
-from ...files import FilePermissions, FsNode
+from ...files import FsNode, permissions_to_str
 
 
 class UiActionFileInfo(BaseModel):
@@ -51,16 +51,7 @@ def to_fs_node(self) -> FsNode:
         file_id = str(self.fileId).rjust(8, "0")
 
         permissions = "S" if self.shareOwnerId else ""
-        if self.permissions & FilePermissions.PERMISSION_SHARE:
-            permissions += "R"
-        if self.permissions & FilePermissions.PERMISSION_READ:
-            permissions += "G"
-        if self.permissions & FilePermissions.PERMISSION_DELETE:
-            permissions += "D"
-        if self.permissions & FilePermissions.PERMISSION_UPDATE:
-            permissions += "NV" if is_dir else "NVW"
-        if is_dir and self.permissions & FilePermissions.PERMISSION_CREATE:
-            permissions += "CK"
+        permissions += permissions_to_str(self.permissions, is_dir)
         return FsNode(
             full_path,
             etag=self.etag,

nc_py_api/files/__init__.py

@@ -6,6 +6,8 @@
 import enum
 import typing
 
+from .. import _misc
+
 
 @dataclasses.dataclass
 class FsNodeInfo:
@@ -216,6 +218,26 @@ class FilePermissions(enum.IntFlag):
     """Access to re-share object(s)"""
 
 
+def permissions_to_str(permissions: int, is_dir: bool = False) -> str:
+    """Converts integer permissions to string permissions.
+
+    :param permissions: concatenation of ``FilePermissions`` integer flags.
+    :param is_dir: Flag indicating is permissions related to the directory object or not.
+    """
+    r = ""
+    if permissions & FilePermissions.PERMISSION_SHARE:
+        r += "R"
+    if permissions & FilePermissions.PERMISSION_READ:
+        r += "G"
+    if permissions & FilePermissions.PERMISSION_DELETE:
+        r += "D"
+    if permissions & FilePermissions.PERMISSION_UPDATE:
+        r += "NV" if is_dir else "NVW"
+    if is_dir and permissions & FilePermissions.PERMISSION_CREATE:
+        r += "CK"
+    return r
+
+
 @dataclasses.dataclass
 class SystemTag:
     """Nextcloud System Tag."""
@@ -242,3 +264,113 @@ def user_visible(self) -> bool:
     def user_assignable(self) -> bool:
         """Flag indicating if User can assign this Tag."""
         return bool(self._raw_data.get("oc:user-assignable", "false").lower() == "true")
+
+
+class ShareType(enum.IntEnum):
+    """Type of the object that will receive share."""
+
+    TYPE_USER = 0
+    """Share to the user"""
+    TYPE_GROUP = 1
+    """Share to the group"""
+    TYPE_LINK = 3
+    """Share by link"""
+    TYPE_EMAIL = 4
+    """Share by the email"""
+    TYPE_REMOTE = 6
+    """Share to the Federation"""
+    TYPE_CIRCLE = 7
+    """Share to the Nextcloud Circle"""
+    TYPE_GUEST = 8
+    """Share to `Guest`"""
+    TYPE_REMOTE_GROUP = 9
+    """Share to the Federation group"""
+    TYPE_ROOM = 10
+    """Share to the Talk room"""
+    TYPE_DECK = 11
+    """Share to the Nextcloud Deck"""
+    TYPE_SCIENCE_MESH = 15
+    """Share to the Reva instance(Science Mesh)"""
+
+
+class Share:
+    """Information about Share."""
+
+    def __init__(self, raw_data: dict):
+        self.raw_data = raw_data
+
+    @property
+    def share_id(self) -> int:
+        """Unique ID of the share."""
+        return int(self.raw_data["id"])
+
+    @property
+    def share_type(self) -> ShareType:
+        """Type of the share."""
+        return ShareType(int(self.raw_data["share_type"]))
+
+    @property
+    def share_with(self) -> str:
+        """To whom Share was created."""
+        return self.raw_data["share_with"]
+
+    @property
+    def permissions(self) -> FilePermissions:
+        """Recipient permissions."""
+        return FilePermissions(int(self.raw_data["permissions"]))
+
+    @property
+    def url(self) -> str:
+        """URL at which Share is avalaible."""
+        return self.raw_data.get("url", "")
+
+    @property
+    def path(self) -> str:
+        """Share path relative to the user's root directory."""
+        return self.raw_data.get("path", "").lstrip("/")
+
+    @property
+    def label(self) -> str:
+        """Label for the Shared object."""
+        return self.raw_data.get("label", "")
+
+    @property
+    def note(self) -> str:
+        """Note for the Shared object."""
+        return self.raw_data.get("note", "")
+
+    @property
+    def mimetype(self) -> str:
+        """Mimetype of the Shared object."""
+        return self.raw_data.get("mimetype", "")
+
+    @property
+    def share_owner(self) -> str:
+        """Share's creator ID."""
+        return self.raw_data.get("uid_owner", "")
+
+    @property
+    def file_owner(self) -> str:
+        """File/directory owner ID."""
+        return self.raw_data.get("uid_file_owner", "")
+
+    @property
+    def password(self) -> str:
+        """Password to access share."""
+        return self.raw_data.get("password", "")
+
+    @property
+    def send_password_by_talk(self) -> bool:
+        """Flag indicating was password send by Talk."""
+        return self.raw_data.get("send_password_by_talk", False)
+
+    @property
+    def expire_date(self) -> datetime.datetime:
+        """Share expiration time."""
+        return _misc.nc_iso_time_to_datetime(self.raw_data.get("expiration", ""))
+
+    def __str__(self):
+        return (
+            f"{self.share_type.name}: `{self.path}` with id={self.share_id}"
+            f" from {self.share_owner} to {self.share_with}"
+        )

nc_py_api/files/sharing.py

@@ -1,120 +1,9 @@
 """Nextcloud API for working with the files shares."""
-import datetime
-import enum
+
 import typing
 
 from .. import _misc, _session
-from . import FilePermissions, FsNode
-
-
-class ShareType(enum.IntEnum):
-    """Type of the object that will receive share."""
-
-    TYPE_USER = 0
-    """Share to the user"""
-    TYPE_GROUP = 1
-    """Share to the group"""
-    TYPE_LINK = 3
-    """Share by link"""
-    TYPE_EMAIL = 4
-    """Share by the email"""
-    TYPE_REMOTE = 6
-    """Share to the Federation"""
-    TYPE_CIRCLE = 7
-    """Share to the Nextcloud Circle"""
-    TYPE_GUEST = 8
-    """Share to `Guest`"""
-    TYPE_REMOTE_GROUP = 9
-    """Share to the Federation group"""
-    TYPE_ROOM = 10
-    """Share to the Talk room"""
-    TYPE_DECK = 11
-    """Share to the Nextcloud Deck"""
-    TYPE_SCIENCE_MESH = 15
-    """Share to the Reva instance(Science Mesh)"""
-
-
-class Share:
-    """Information about Share."""
-
-    def __init__(self, raw_data: dict):
-        self.raw_data = raw_data
-
-    @property
-    def share_id(self) -> int:
-        """Unique ID of the share."""
-        return int(self.raw_data["id"])
-
-    @property
-    def share_type(self) -> ShareType:
-        """Type of the share."""
-        return ShareType(int(self.raw_data["share_type"]))
-
-    @property
-    def share_with(self) -> str:
-        """To whom Share was created."""
-        return self.raw_data["share_with"]
-
-    @property
-    def permissions(self) -> FilePermissions:
-        """Recipient permissions."""
-        return FilePermissions(int(self.raw_data["permissions"]))
-
-    @property
-    def url(self) -> str:
-        """URL at which Share is avalaible."""
-        return self.raw_data.get("url", "")
-
-    @property
-    def path(self) -> str:
-        """Share path relative to the user's root directory."""
-        return self.raw_data.get("path", "").lstrip("/")
-
-    @property
-    def label(self) -> str:
-        """Label for the Shared object."""
-        return self.raw_data.get("label", "")
-
-    @property
-    def note(self) -> str:
-        """Note for the Shared object."""
-        return self.raw_data.get("note", "")
-
-    @property
-    def mimetype(self) -> str:
-        """Mimetype of the Shared object."""
-        return self.raw_data.get("mimetype", "")
-
-    @property
-    def share_owner(self) -> str:
-        """Share's creator ID."""
-        return self.raw_data.get("uid_owner", "")
-
-    @property
-    def file_owner(self) -> str:
-        """File/directory owner ID."""
-        return self.raw_data.get("uid_file_owner", "")
-
-    @property
-    def password(self) -> str:
-        """Password to access share."""
-        return self.raw_data.get("password", "")
-
-    @property
-    def send_password_by_talk(self) -> bool:
-        """Flag indicating was password send by Talk."""
-        return self.raw_data.get("send_password_by_talk", False)
-
-    @property
-    def expire_date(self) -> datetime.datetime:
-        """Share expiration time."""
-        return _misc.nc_iso_time_to_datetime(self.raw_data.get("expiration", ""))
-
-    def __str__(self):
-        return (
-            f"{self.share_type.name}: `{self.path}` with id={self.share_id}"
-            f" from {self.share_owner} to {self.share_with}"
-        )
+from . import FilePermissions, FsNode, Share, ShareType
 
 
 class _FilesSharingAPI:
@@ -193,9 +82,8 @@ def create(
             * ``label`` - string with label, if any. default = ``""``
         """
         _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
-        path = path.user_path if isinstance(path, FsNode) else path
         params = {
-            "path": path,
+            "path": path.user_path if isinstance(path, FsNode) else path,
             "shareType": int(share_type),
         }
         if permissions is not None: