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
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


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 ''

        tags = [self.favro_client.get_tag(tag_id).name for tag_id in card.tags]
        assignments = [
            self.favro_client.get_user(assignment.user).name
            for assignment in card.assignments
        ]
        dependencies = [
            CARD_FILENAME_FORMAT.format(
                seq_id=self.favro_client.get_card_by_card_id(dep.card_id).seq_id.raw_id,
            )
            for dep in card.dependencies
            if dep.is_before
        ]
        card_contents = 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,
            ),
        )
        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()