favro-sync/favro_sync/favro_fuse.py

import dataclasses
import errno
import re
import stat
from collections.abc import Iterator
from logging import getLogger

import fuse

from .favro_client import FavroClient
from .favro_data_model import (
    Card,
    CustomField,
    CustomFieldInfo,
    CustomFieldItemId,
    SeqId,
)
from .favro_markdown import CardContents, CardFileFormatter

################################################################################
# Constants

logger = getLogger(__name__)

fuse.fuse_python_api = (0, 2)

OFFICIAL_URL = 'https://favro.com/organization/{org_id}?card=par-{seq_id}'
CARD_IDENTIFIER_FORMAT = 'PAR-{seq_id}'
CARD_FILENAME_FORMAT = CARD_IDENTIFIER_FORMAT + '.md'
CARD_FILENAME_REGEX = r'^PAR\-(\d+)\.md$'

################################################################################
# Formatting


def to_custom_field_value(
    custom_field: CustomField, field_def: CustomFieldInfo,
) -> str | None:
    value: CustomFieldItemId | list[CustomFieldItemId] = custom_field.value
    if field_def.type in {'Single select', 'Multiple select'}:
        items = [field_def.get_field_item(item_id) for item_id in value]
        items = [i for i in items if i]
        return items[0].name
    if field_def.type in {'Color'}:
        return custom_field.color
    return None


def to_custom_fields(card: Card, favro_client: FavroClient) -> dict[str, str]:
    custom_fields = {}
    for field_assignment in card.custom_fields:
        field_def = favro_client.get_custom_field(field_assignment.custom_field_id)
        str_value = to_custom_field_value(field_assignment, field_def)
        if str_value is not None:
            custom_fields[field_def.name] = str_value
        del field_assignment, str_value
    return custom_fields


def to_card_contents(card: Card, favro_client: FavroClient) -> CardContents:
    tags = [favro_client.get_tag(tag_id).name for tag_id in card.tags]
    assignments = [
        favro_client.get_user(assignment.user).name for assignment in card.assignments
    ]
    dependencies = [
        CARD_FILENAME_FORMAT.format(
            seq_id=favro_client.get_card_by_card_id(dep.card_id).seq_id.raw_id,
        )
        for dep in card.dependencies
        if dep.is_before
    ]
    return CardContents(
        CARD_IDENTIFIER_FORMAT.format(seq_id=card.seq_id.raw_id),
        card.name,
        card.detailed_description,
        tags,
        assignments,
        dependencies,
        url=OFFICIAL_URL.format(
            org_id=card.organization_id.raw_id,
            seq_id=card.seq_id.raw_id,
        ),
        todo_list_completed=card.todo_list_completed,
        is_archived=card.is_archived,
        start_date=card.start_date,
        due_date=card.due_date,
        custom_fields=to_custom_fields(card, favro_client),
    )


################################################################################
# FUSE


class FavroStat(fuse.Stat):
    def __init__(self):
        self.st_mode = 0
        self.st_ino = 0
        self.st_dev = 0
        self.st_nlink = 0
        self.st_uid = 0
        self.st_gid = 0
        self.st_size = 0
        self.st_atime = 0
        self.st_mtime = 0
        self.st_ctime = 0


@dataclasses.dataclass(frozen=True)
class FileSystemItem:
    pass


@dataclasses.dataclass(frozen=True)
class RootFileSystemItem(FileSystemItem):
    @staticmethod
    def from_path_segment(_segment: str) -> 'RootFileSystemItem':
        return RootFileSystemItem()

    def __str__(self):
        return '/'


@dataclasses.dataclass(frozen=True)
class CollectionFileSystemItem(FileSystemItem):
    collection_name: str

    def __post_init__(self):
        assert '/' not in self.collection_name

    @staticmethod
    def from_path_segment(segment: str) -> 'CollectionFileSystemItem':
        return CollectionFileSystemItem(segment)

    def __str__(self):
        return self.collection_name


@dataclasses.dataclass(frozen=True)
class CardFileSystemItem(FileSystemItem):
    seq_id: SeqId

    @staticmethod
    def from_path_segment(segment: str) -> 'CardFileSystemItem | None':
        if m := re.match(CARD_FILENAME_REGEX, segment):
            return CardFileSystemItem(SeqId(int(m.group(1))))
        return None

    def __str__(self):
        return CARD_FILENAME_FORMAT.format(seq_id=self.seq_id.raw_id)


def path_to_file_system_item(
    path_str: str, path_components: list[type[FileSystemItem]],
) -> FileSystemItem | None:
    path = re.findall(r'[^/]+', path_str)
    component = path_components[len(path)]
    return component.from_path_segment(path[-1] if path else None)


class FavroFuse(fuse.Fuse):
    """Favro FileSystem in Userspace."""

    def __init__(
        self,
        favro_client: FavroClient,
        formatter: CardFileFormatter,
        **kwargs,
    ):
        self.favro_client = favro_client
        self.formatter = formatter
        self.wiped_cards = set()
        self.path_components = [
            RootFileSystemItem,
            CollectionFileSystemItem,
            CardFileSystemItem,
        ]
        super().__init__(**kwargs)

    def getattr(self, path: str) -> FavroStat | int:
        file_system_item = path_to_file_system_item(path, self.path_components)

        st = FavroStat()
        if isinstance(file_system_item, RootFileSystemItem | CollectionFileSystemItem):
            st.st_mode = stat.S_IFDIR | 0o755
            st.st_nlink = 2
        elif isinstance(file_system_item, CardFileSystemItem):
            card = self.favro_client.get_card(file_system_item.seq_id)

            st.st_mode = stat.S_IFREG | 0o666
            st.st_nlink = 1
            st.st_size = len(self._format_card_file(card))
            st.st_ctime = int(card.creation_date.timestamp())
            st.st_mtime = st.st_ctime  # TODO
        else:
            return -errno.ENOENT
        return st

    def readdir(self, path: str, offset: int) -> Iterator[fuse.Direntry]:
        file_system_item = path_to_file_system_item(path, self.path_components)

        yield fuse.Direntry('.')
        yield fuse.Direntry('..')

        if isinstance(file_system_item, RootFileSystemItem):
            for collection in self.favro_client.get_collections():
                collection_name = collection.name.replace('/', '')
                yield fuse.Direntry(str(CollectionFileSystemItem(collection_name)))
                del collection

        elif isinstance(file_system_item, CollectionFileSystemItem):
            # TODO: move into own function
            for collection in self.favro_client.get_collections():
                if collection.name.replace('/', '') == file_system_item.collection_name:
                    collection_id = collection.collection_id
                del collection

            for card in self.favro_client.get_cards(collection_id=collection_id):
                yield fuse.Direntry(str(CardFileSystemItem(card.seq_id)))
                del card

    def open(self, path: str, flags: int) -> int | None:
        file_system_item = path_to_file_system_item(path, self.path_components)
        if not isinstance(file_system_item, CardFileSystemItem):
            return -errno.ENOENT
        return None

    def read(self, path: str, size: int, offset: int) -> bytes | int:
        # Check that this is a card file_system_item.
        file_system_item = path_to_file_system_item(path, self.path_components)
        if not isinstance(file_system_item, CardFileSystemItem):
            return -errno.ENOENT

        card = self.favro_client.get_card(file_system_item.seq_id)

        contents_str = self._format_card_file(card)
        contents = bytes(contents_str, 'utf8')

        slen = len(contents)
        if offset < slen:
            if offset + size > slen:
                size = slen - offset
            buf = contents[offset : offset + size]
        else:
            buf = b''
        return buf

    def write(self, path: str, written_buffer: bytes, offset: int) -> int:
        # Check that this is a card file_system_item.
        file_system_item = path_to_file_system_item(path, self.path_components)
        if not isinstance(file_system_item, CardFileSystemItem):
            return -errno.ENOENT

        card = self.favro_client.get_card(file_system_item.seq_id)

        # Splice contents
        contents_str = self._format_card_file(card)
        contents = bytes(contents_str, 'utf8')
        contents = splice(contents, written_buffer, offset)
        contents_str = contents.decode('utf8')

        # Write to favro
        card_updated = self.formatter.parse_card_contents(contents_str)
        self.favro_client.update_card_contents(card.card_id, card_updated)

        self.wiped_cards.remove(file_system_item.seq_id)

        # Return amount written
        return len(written_buffer)

    def truncate(self, path: str, new_size: int):
        # Check that this is a card file_system_item.
        file_system_item = path_to_file_system_item(path, self.path_components)
        if not isinstance(file_system_item, CardFileSystemItem):
            return -errno.ENOENT

        card = self.favro_client.get_card(file_system_item.seq_id)

        # Splice contents
        contents_str = self._format_card_file(card)
        contents = bytes(contents_str, 'utf8')
        old_size = len(contents)
        contents = contents[0:new_size] + b' ' * (old_size - new_size)
        assert len(contents) == old_size
        contents_str = contents.decode('utf8')

        # Write to favro
        card_updated = self.formatter.parse_card_contents(contents_str)
        self.favro_client.update_card_contents_locally(card.card_id, card_updated)

        self.wiped_cards.add(file_system_item.seq_id)

        # Return amount written
        return 0

    def _format_card_file(self, card: Card) -> str:
        if card.seq_id in self.wiped_cards:
            return ''
        card_contents = to_card_contents(card, self.favro_client)
        return self.formatter.format_card_contents(card_contents)


def splice(original_buffer: bytes, input_buffer: bytes, offset: int) -> bytes:
    return (
        original_buffer[0 : offset - 1]
        + input_buffer
        + original_buffer[offset + len(input_buffer) + 1 : len(original_buffer)]
    )


HELP = (
    """
Userspace hello example

"""
    + fuse.Fuse.fusage
)


def start_favro_fuse(favro_client: FavroClient):
    # TODO:
    server = FavroFuse(
        favro_client=favro_client,
        formatter=CardFileFormatter(),
        version='%prog ' + fuse.__version__,
        usage=HELP,
        dash_s_do='setsingle',
    )
    server.parse(errex=1)
    server.main()