From 99ec73ebda7151679650de6787ff10e22b87f4c7 Mon Sep 17 00:00:00 2001 From: Alexander Piskun Date: Tue, 26 Sep 2023 00:46:17 +0300 Subject: [PATCH] Nextcloud Talk: Polls API Signed-off-by: Alexander Piskun --- CHANGELOG.md | 3 +- docs/reference/Talk.rst | 6 ++ nc_py_api/_talk_api.py | 77 +++++++++++++++++++++ nc_py_api/_version.py | 2 +- nc_py_api/talk.py | 114 ++++++++++++++++++++++++++++++++ tests/actual_tests/talk_test.py | 69 +++++++++++++++++++ 6 files changed, 269 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 14b3cc3b..7a99f540 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/reference/Talk.rst b/docs/reference/Talk.rst index 0b624d7e..60e4adf8 100644 --- a/docs/reference/Talk.rst +++ b/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: diff --git a/nc_py_api/_talk_api.py b/nc_py_api/_talk_api.py index 1d484baa..84192dd6 100644 --- a/nc_py_api/_talk_api.py +++ b/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): diff --git a/nc_py_api/_version.py b/nc_py_api/_version.py index c662d34b..93f4422a 100644 --- a/nc_py_api/_version.py +++ b/nc_py_api/_version.py @@ -1,3 +1,3 @@ """Version of nc_py_api.""" -__version__ = "0.2.1" +__version__ = "0.2.2.dev0" diff --git a/nc_py_api/talk.py b/nc_py_api/talk.py index b5124eb1..1bdec74c 100644 --- a/nc_py_api/talk.py +++ b/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", [])] diff --git a/tests/actual_tests/talk_test.py b/tests/actual_tests/talk_test.py index e6449147..767492f6 100644 --- a/tests/actual_tests/talk_test.py +++ b/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)