OpenSlides/openslides/utils/consumers.py

from collections import defaultdict
from typing import Any, Dict, List
from urllib.parse import parse_qs

from .auth import async_anonymous_is_enabled
from .autoupdate import AutoupdateFormat
from .cache import element_cache, split_element_id
from .websocket import ProtocollAsyncJsonWebsocketConsumer, get_element_data


class SiteConsumer(ProtocollAsyncJsonWebsocketConsumer):
    """
    Websocket Consumer for the site.
    """

    groups = ["site"]

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        self.projector_hash: Dict[int, int] = {}
        super().__init__(*args, **kwargs)

    async def connect(self) -> None:
        """
        A user connects to the site.

        If it is an anonymous user and anonymous is disabled, the connection is closed.

        Sends the startup data to the user.
        """
        # self.scope['user'] is the full_data dict of the user. For an
        # anonymous user is it the dict {'id': 0}
        change_id = None
        if not await async_anonymous_is_enabled() and not self.scope["user"]["id"]:
            await self.close()
            return

        query_string = parse_qs(self.scope["query_string"])
        if b"change_id" in query_string:
            try:
                change_id = int(query_string[b"change_id"][0])
            except ValueError:
                await self.close()  # TODO: Find a way to send an error code
                return

        if b"autoupdate" in query_string and query_string[b"autoupdate"][
            0
        ].lower() not in [b"0", b"off", b"false"]:
            # a positive value in autoupdate. Start autoupdate
            await self.channel_layer.group_add("autoupdate", self.channel_name)

        await self.accept()

        if change_id is not None:
            try:
                data = await get_element_data(self.scope["user"]["id"], change_id)
            except ValueError:
                # When the change_id is to big, do nothing
                pass
            else:
                await self.send_json(type="autoupdate", content=data)

    async def disconnect(self, close_code: int) -> None:
        """
        A user disconnects. Remove it from autoupdate.
        """
        await self.channel_layer.group_discard("autoupdate", self.channel_name)

    async def send_notify(self, event: Dict[str, Any]) -> None:
        """
        Send a notify message to the user.
        """
        user_id = self.scope["user"]["id"]
        item = event["incomming"]

        users = item.get("users")
        reply_channels = item.get("replyChannels")
        if (
            (isinstance(users, bool) and users)
            or (isinstance(users, list) and user_id in users)
            or (
                isinstance(reply_channels, list) and self.channel_name in reply_channels
            )
            or (users is None and reply_channels is None)
        ):
            item["senderChannelName"] = event["senderChannelName"]
            item["senderUserId"] = event["senderUserId"]
            await self.send_json(type="notify", content=item)

    async def send_data(self, event: Dict[str, Any]) -> None:
        """
        Send changed or deleted elements to the user.
        """
        change_id = event["change_id"]
        changed_elements, deleted_elements_ids = await element_cache.get_restricted_data(
            self.scope["user"]["id"], change_id, max_change_id=change_id
        )

        deleted_elements: Dict[str, List[int]] = defaultdict(list)
        for element_id in deleted_elements_ids:
            collection_string, id = split_element_id(element_id)
            deleted_elements[collection_string].append(id)
        await self.send_json(
            type="autoupdate",
            content=AutoupdateFormat(
                changed=changed_elements,
                deleted=deleted_elements,
                from_change_id=change_id,
                to_change_id=change_id,
                all_data=False,
            ),
        )

    async def projector_changed(self, event: Dict[str, Any]) -> None:
        """
        The projector has changed.
        """
        all_projector_data = event["data"]
        projector_data: Dict[int, Dict[str, Any]] = {}
        for projector_id in self.listen_projector_ids:
            data = all_projector_data.get(
                projector_id, {"error": f"No data for projector {projector_id}"}
            )
            new_hash = hash(str(data))
            if new_hash != self.projector_hash[projector_id]:
                projector_data[projector_id] = data
                self.projector_hash[projector_id] = new_hash

        if projector_data:
            await self.send_json(type="projector", content=projector_data)