moment/src/backend/user_files.py

# SPDX-License-Identifier: LGPL-3.0-or-later

"""User data and configuration files definitions."""

import asyncio
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional

import aiofiles

from .theme_parser import convert_to_qml
from .utils import dict_update_recursive

if TYPE_CHECKING:
    from .backend import Backend

JsonData = Dict[str, Any]

WRITE_LOCK = asyncio.Lock()


@dataclass
class DataFile:
    """Base class representing a user data file."""

    is_config:      ClassVar[bool] = False
    create_missing: ClassVar[bool] = True

    backend:  "Backend" = field(repr=False)
    filename: str       = field()

    _to_write: Optional[str] = field(init=False, default=None)


    def __post_init__(self) -> None:
        asyncio.ensure_future(self._write_loop())


    @property
    def path(self) -> Path:
        """Full path of the file, even if it doesn't exist yet."""

        if self.is_config:
            return Path(self.backend.appdirs.user_config_dir) / self.filename

        return Path(self.backend.appdirs.user_data_dir) / self.filename


    async def default_data(self):
        """Default content if the file doesn't exist."""

        return ""


    async def read(self):
        """Return content of the existing file on disk, or default content."""

        try:
            return self.path.read_text()
        except FileNotFoundError:
            default = await self.default_data()

            if self.create_missing:
                await self.write(default)

            return default


    async def write(self, data) -> None:
        """Request for the file to be written/updated with data."""

        self._to_write = data


    async def _write_loop(self) -> None:
        """Write/update file on disk with a 1 second cooldown."""

        self.path.parent.mkdir(parents=True, exist_ok=True)

        while True:
            await asyncio.sleep(1)

            if self._to_write is None:
                continue

            if not self.create_missing and not self.path.exists():
                continue

            async with aiofiles.open(self.path, "w") as new:
                await new.write(self._to_write)

            self._to_write = None



@dataclass
class JSONDataFile(DataFile):
    """Represent a user data file in the JSON format."""

    async def default_data(self) -> JsonData:
        return {}


    async def read(self) -> JsonData:
        """Return content of the existing file on disk, or default content.

        If the file has missing keys, the missing data will be merged and
        written to disk before returning.

        If `create_missing` is `True` and the file doesn't exist, it will be
        created.
        """

        try:
            data = json.loads(await super().read())
        except FileNotFoundError:
            if not self.create_missing:
                return await self.default_data()

            data = {}
        except json.JSONDecodeError:
            data = {}

        all_data = await self.default_data()
        dict_update_recursive(all_data, data)

        if data != all_data:
            await self.write(all_data)

        return all_data


    async def write(self, data: JsonData) -> None:
        js = json.dumps(data, indent=4, ensure_ascii=False, sort_keys=True)
        await super().write(js)


@dataclass
class Accounts(JSONDataFile):
    """Config file for saved matrix accounts: user ID, access tokens, etc."""

    is_config = True

    filename: str = "accounts.json"


    async def any_saved(self) -> bool:
        """Return whether there are any accounts saved on disk."""
        return bool(await self.read())


    async def add(self, user_id: str) -> None:
        """Add an account to the config and write it on disk.

        The account's details such as its access token are retrieved from
        the corresponding `MatrixClient` in `backend.clients`.
        """

        client = self.backend.clients[user_id]

        await self.write({
            **await self.read(),
            client.user_id: {
                "homeserver": client.homeserver,
                "token":      client.access_token,
                "device_id":  client.device_id,
            },
        })


    async def delete(self, user_id: str) -> None:
        """Delete an account from the config and write it on disk."""

        await self.write({
            uid: info
            for uid, info in (await self.read()).items() if uid != user_id
        })


@dataclass
class UISettings(JSONDataFile):
    """Config file for QML interface settings and keybindings."""

    is_config = True

    filename: str = "settings.json"


    async def default_data(self) -> JsonData:
        return {
            "alertOnMessageForMsec": 4000,
            "clearRoomFilterOnEnter": True,
            "clearRoomFilterOnEscape": True,
            "theme": "Midnight.qpl",
            "writeAliases": {},
            "media": {
                "autoLoad": True,
                "autoPlay": False,
                "autoPlayGIF": True,
                "autoHideOSDAfterMsec": 3000,
                "defaultVolume": 100,
                "startMuted": False,
            },
            "keys": {
                "startPythonDebugger": ["Alt+Shift+D"],
                "toggleDebugConsole":  ["Alt+Shift+C", "F1"],
                "reloadConfig":        ["Alt+Shift+R"],

                "zoomIn":    ["Ctrl++"],
                "zoomOut":   ["Ctrl+-"],
                "zoomReset": ["Ctrl+="],

                "scrollUp":       ["Alt+Up", "Alt+K"],
                "scrollDown":     ["Alt+Down", "Alt+J"],
                "scrollPageUp":   ["Alt+Ctrl+Up", "Alt+Ctrl+K", "PgUp"],
                "scrollPageDown": ["Alt+Ctrl+Down", "Alt+Ctrl+J", "PgDown"],
                "scrollToTop":
                    ["Alt+Ctrl+Shift+Up", "Alt+Ctrl+Shift+K", "Home"],
                "scrollToBottom":
                    ["Alt+Ctrl+Shift+Down", "Alt+Ctrl+Shift+J", "End"],

                "previousTab": ["Alt+Shift+Left", "Alt+Shift+H"],
                "nextTab":     ["Alt+Shift+Right", "Alt+Shift+L"],

                "focusMainPane":   ["Alt+S"],
                "clearRoomFilter": ["Alt+Shift+S"],
                "accountSettings": ["Alt+A"],
                "addNewChat":      ["Alt+N"],
                "addNewAccount":   ["Alt+Shift+N"],

                "goToLastPage":          ["Ctrl+Tab"],
                "goToPreviousRoom":      ["Alt+Shift+Up", "Alt+Shift+K"],
                "goToNextRoom":          ["Alt+Shift+Down", "Alt+Shift+J"],
                "toggleCollapseAccount": [ "Alt+O"],

                "clearRoomMessages":           ["Ctrl+L"],
                "sendFile":                    ["Alt+F"],
                "sendFileFromPathInClipboard": ["Alt+Shift+F"],
            },
        }


@dataclass
class UIState(JSONDataFile):
    """File to save and restore the state of the QML interface."""

    filename: str = "state.json"


    async def default_data(self) -> JsonData:
        return {
            "collapseAccounts": {},
            "page":             "Pages/Default.qml",
            "pageProperties":   {},
        }


@dataclass
class History(JSONDataFile):
    """File to save and restore lines typed by the user in QML components."""

    filename: str = "history.json"


    async def default_data(self) -> JsonData:
        return {"console": []}


@dataclass
class Theme(DataFile):
    """A theme file defining the look of QML components."""

    # Since it currently breaks at every update and the file format will be
    # changed later, don't copy the theme to user data dir if it doesn't exist.
    create_missing = False


    @property
    def path(self) -> Path:
        data_dir  = Path(self.backend.appdirs.user_data_dir)
        user_path = data_dir / "themes" / self.filename

        if not user_path.exists():
            return Path("src") / "themes" / self.filename

        return user_path


    async def default_data(self) -> str:
        async with aiofiles.open("src/themes/Midnight.qpl") as file:
            return await file.read()


    async def read(self) -> str:
        return convert_to_qml(await super().read())