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, SeqId, CustomFieldInfo, CustomFieldItemId, CustomField
from .favro_markdown import CardContents, CardFileFormatter

logger = getLogger(__name__)

fuse.fuse_python_api = (0, 2)


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


CARD_IDENTIFIER_FORMAT = 'PAR-{seq_id}'
CARD_FILENAME_FORMAT = CARD_IDENTIFIER_FORMAT + '.md'
CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$'

OFFICIAL_URL = 'https://favro.com/organization/{org_id}?card=par-{seq_id}'


@dataclasses.dataclass(frozen=True)
class FileSystemItem:
    @staticmethod
    def from_path(path_str: str) -> 'FileSystemItem | None':
        if path_str == '/':
            return RootFileSystemItem()
        if m := re.match(CARD_FILENAME_REGEX, path_str):
            return CardFileSystemItem(SeqId(int(m.group(1))))
        return None


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


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


def to_custom_field_value(custom_field: CustomField, field_def: CustomFieldInfo) -> str:
    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
    assert False, 'Unknown type: ' + field_def.type

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)
        custom_fields[field_def.name] = to_custom_field_value(field_assignment, field_def)
        del field_assignment
    return custom_fields

def to_card_contents(card: Card, favro_client: FavroClient) -> str:
    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),
    )


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()
        super().__init__(**kwargs)

    def getattr(self, path: str) -> FavroStat | int:
        thing = FileSystemItem.from_path(path)

        st = FavroStat()
        if isinstance(thing, RootFileSystemItem):
            st.st_mode = stat.S_IFDIR | 0o755
            st.st_nlink = 2
        elif isinstance(thing, CardFileSystemItem):
            card = self.favro_client.get_card(thing.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]:
        logger.warning('readdir(path=%s, offset=%s)', path, offset)
        yield fuse.Direntry('.')
        yield fuse.Direntry('..')

        for card in self.favro_client.get_todo_list_cards():
            yield fuse.Direntry(CARD_FILENAME_FORMAT.format(seq_id=card.seq_id.raw_id))

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

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

        card = self.favro_client.get_card(thing.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 thing.
        thing = FileSystemItem.from_path(path)
        if not isinstance(thing, CardFileSystemItem):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.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(thing.seq_id)

        # Return amount written
        return len(written_buffer)

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

        card = self.favro_client.get_card(thing.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(thing.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()