Nextcloud Talk: Polls API

CHANGELOG.md

@@ -7,7 +7,8 @@ All notable changes to this project will be documented in this file.
 ### 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 need for that: `CHUNKED_UPLOAD_V2`
+- 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).
 
 ### Changed
 

docs/reference/Talk.rst

@@ -63,3 +63,9 @@ Talk API
 
 .. autoclass:: nc_py_api.talk.BotInfoBasic
     :members:
+
+.. autoclass:: nc_py_api.talk.Poll
+    :members:
+
+.. autoclass:: nc_py_api.talk.PollDetail
+    :members:

nc_py_api/_talk_api.py

@@ -17,6 +17,7 @@
     ConversationType,
     MessageReactions,
     NotificationLevel,
+    Poll,
     TalkMessage,
 )
 
@@ -386,6 +387,82 @@ def disable_bot(self, conversation: typing.Union[Conversation, str], bot: typing
         bot_id = bot.bot_id if isinstance(bot, BotInfoBasic) else bot
         self._session.ocs("DELETE", self._ep_base + f"/api/v1/bot/{token}/{bot_id}")
 
+    def create_poll(
+        self,
+        conversation: typing.Union[Conversation, str],
+        question: str,
+        options: list[str],
+        hidden_results: bool = True,
+        max_votes: int = 1,
+    ) -> Poll:
+        """Creates a poll in a conversation.
+
+        :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`.
+        :param question: The question of the poll.
+        :param options: Array of strings with the voting options.
+        :param hidden_results: Are the results hidden until the poll is closed and then only the summary is published.
+        :param max_votes: The maximum amount of options a participant can vote for.
+        """
+        token = conversation.token if isinstance(conversation, Conversation) else conversation
+        params = {
+            "question": question,
+            "options": options,
+            "resultMode": int(hidden_results),
+            "maxVotes": max_votes,
+        }
+        return Poll(self._session.ocs("POST", self._ep_base + f"/api/v1/poll/{token}", json=params), token)
+
+    def get_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[Conversation, str] = "") -> Poll:
+        """Get state or result of a poll.
+
+        :param poll: Poll ID or :py:class:`~nc_py_api.talk.Poll`.
+        :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`.
+        """
+        if isinstance(poll, Poll):
+            poll_id = poll.poll_id
+            token = poll.conversation_token
+        else:
+            poll_id = poll
+            token = conversation.token if isinstance(conversation, Conversation) else conversation
+        return Poll(self._session.ocs("GET", self._ep_base + f"/api/v1/poll/{token}/{poll_id}"), token)
+
+    def vote_poll(
+        self,
+        options_ids: list[int],
+        poll: typing.Union[Poll, int],
+        conversation: typing.Union[Conversation, str] = "",
+    ) -> Poll:
+        """Vote on a poll.
+
+        :param options_ids: The option IDs the participant wants to vote for.
+        :param poll: Poll ID or :py:class:`~nc_py_api.talk.Poll`.
+        :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`.
+        """
+        if isinstance(poll, Poll):
+            poll_id = poll.poll_id
+            token = poll.conversation_token
+        else:
+            poll_id = poll
+            token = conversation.token if isinstance(conversation, Conversation) else conversation
+        r = self._session.ocs(
+            "POST", self._ep_base + f"/api/v1/poll/{token}/{poll_id}", json={"optionIds": options_ids}
+        )
+        return Poll(r, token)
+
+    def close_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[Conversation, str] = "") -> Poll:
+        """Close a poll.
+
+        :param poll: Poll ID or :py:class:`~nc_py_api.talk.Poll`.
+        :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`.
+        """
+        if isinstance(poll, Poll):
+            poll_id = poll.poll_id
+            token = poll.conversation_token
+        else:
+            poll_id = poll
+            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)
+
     @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/_version.py

@@ -1,3 +1,3 @@
 """Version of nc_py_api."""
 
-__version__ = "0.2.1"
+__version__ = "0.2.2.dev0"

nc_py_api/talk.py

@@ -625,3 +625,117 @@ def last_error_date(self) -> int:
     def last_error_message(self) -> typing.Optional[str]:
         """The last exception message or error response information when trying to reach the bot."""
         return self._raw_data["last_error_message"]
+
+
+@dataclasses.dataclass
+class PollDetail:
+    """Detail about who voted for option."""
+
+    def __init__(self, raw_data: dict):
+        self._raw_data = raw_data
+
+    @property
+    def actor_type(self) -> str:
+        """The actor type of the participant that voted: **users**, **groups**, **circles**, **guests**, **emails**."""
+        return self._raw_data["actorType"]
+
+    @property
+    def actor_id(self) -> str:
+        """The actor id of the participant that voted."""
+        return self._raw_data["actorId"]
+
+    @property
+    def actor_display_name(self) -> str:
+        """The display name of the participant that voted."""
+        return self._raw_data["actorDisplayName"]
+
+    @property
+    def option(self) -> int:
+        """The option that was voted for."""
+        return self._raw_data["optionId"]
+
+
+@dataclasses.dataclass
+class Poll:
+    """Conversation Poll information."""
+
+    def __init__(self, raw_data: dict, conversation_token: str):
+        self._raw_data = raw_data
+        self._conversation_token = conversation_token
+
+    @property
+    def conversation_token(self) -> str:
+        """Token identifier of the conversation to which poll belongs."""
+        return self._conversation_token
+
+    @property
+    def poll_id(self) -> int:
+        """ID of the poll."""
+        return self._raw_data["id"]
+
+    @property
+    def question(self) -> str:
+        """The question of the poll."""
+        return self._raw_data["question"]
+
+    @property
+    def options(self) -> list[str]:
+        """Options participants can vote for."""
+        return self._raw_data["options"]
+
+    @property
+    def votes(self) -> dict[str, int]:
+        """Map with 'option-' + optionId => number of votes.
+
+        .. note:: Only available for when the actor voted on the public poll or the poll is closed.
+        """
+        return self._raw_data.get("votes", {})
+
+    @property
+    def actor_type(self) -> str:
+        """Actor type of the poll author: **users**, **groups**, **circles**, **guests**, **emails**."""
+        return self._raw_data["actorType"]
+
+    @property
+    def actor_id(self) -> str:
+        """Actor ID identifying the poll author."""
+        return self._raw_data["actorId"]
+
+    @property
+    def actor_display_name(self) -> str:
+        """The display name of the poll author."""
+        return self._raw_data["actorDisplayName"]
+
+    @property
+    def closed(self) -> bool:
+        """Participants can no longer cast votes and the result is displayed."""
+        return bool(self._raw_data["status"] == 1)
+
+    @property
+    def hidden_results(self) -> bool:
+        """The results are hidden until the poll is closed."""
+        return bool(self._raw_data["resultMode"] == 1)
+
+    @property
+    def max_votes(self) -> int:
+        """The maximum amount of options a user can vote for, ``0`` means unlimited."""
+        return self._raw_data["maxVotes"]
+
+    @property
+    def voted_self(self) -> list[int]:
+        """Array of option ids the participant voted for."""
+        return self._raw_data["votedSelf"]
+
+    @property
+    def num_voters(self) -> int:
+        """The number of unique voters that voted.
+
+        .. note:: only available when the actor voted on the public poll or the
+            poll is closed unless for the creator and moderators.
+        """
+        return self._raw_data.get("numVoters", 0)
+
+    @property
+    def details(self) -> list[PollDetail]:
+        """Detailed list who voted for which option (only available for public closed polls)."""
+        return [PollDetail(i) for i in self._raw_data.get("details", [])]

tests/actual_tests/talk_test.py

@@ -253,3 +253,72 @@ def test_chat_bot_receive_message(nc_app):
     talk_bot_inst.callback_url = "invalid_url"
     with pytest.raises(RuntimeError):
         talk_bot_inst.send_message("message", 999999, token="sometoken")
+
+
+def test_create_close_poll(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:
+        poll = nc_any.talk.create_poll(conversation, "When was this test written?", ["2000", "2023", "2030"])
+
+        def check_poll(closed: bool):
+            assert isinstance(poll.poll_id, int)
+            assert poll.question == "When was this test written?"
+            assert poll.options == ["2000", "2023", "2030"]
+            assert poll.max_votes == 1
+            assert poll.num_voters == 0
+            assert poll.hidden_results is True
+            assert poll.details == []
+            assert poll.closed is closed
+            assert poll.conversation_token == conversation.token
+            assert poll.actor_type == "users"
+            assert poll.actor_id == nc_any.user
+            assert isinstance(poll.actor_display_name, str)
+            assert poll.voted_self == []
+            assert poll.votes == []
+
+        check_poll(False)
+        poll = nc_any.talk.get_poll(poll)
+        check_poll(False)
+        poll = nc_any.talk.get_poll(poll.poll_id, conversation.token)
+        check_poll(False)
+        poll = nc_any.talk.close_poll(poll.poll_id, conversation.token)
+        check_poll(True)
+    finally:
+        nc_any.talk.delete_conversation(conversation)
+
+
+def test_vote_poll(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:
+        poll = nc_any.talk.create_poll(
+            conversation, "what color is the grass", ["red", "green", "blue"], hidden_results=False, max_votes=3
+        )
+        assert poll.hidden_results is False
+        assert not poll.voted_self
+        poll = nc_any.talk.vote_poll([0, 2], poll)
+        assert poll.voted_self == [0, 2]
+        assert poll.votes == {
+            "option-0": 1,
+            "option-2": 1,
+        }
+        assert poll.num_voters == 1
+        poll = nc_any.talk.vote_poll([1], poll.poll_id, conversation)
+        assert poll.voted_self == [1]
+        assert poll.votes == {
+            "option-1": 1,
+        }
+        poll = nc_any.talk.close_poll(poll)
+        assert poll.closed is True
+        assert len(poll.details) == 1
+        assert poll.details[0].actor_id == nc_any.user
+        assert poll.details[0].actor_type == "users"
+        assert poll.details[0].option == 1
+        assert isinstance(poll.details[0].actor_display_name, str)
+    finally:
+        nc_any.talk.delete_conversation(conversation)