<commit_before>"""File-system mock cloud wrapper."""
import os
import re

from cloud_browser.cloud import errors, base
from cloud_browser.common import SEP


NO_DOT_RE = re.compile("^[^.]+")


def not_dot(path):
    """Check if non-dot."""
    return NO_DOT_RE.match(os.path.basename(path))


def is_dir(path):
    """Check if non-dot and directory."""
    return not_dot(path) and os.path.isdir(path)


class FilesystemContainerWrapper(errors.CloudExceptionWrapper):
    """Exception translator."""
    translations = {
        OSError: errors.NoContainerException,
    }
wrap_fs_cont_errors = FilesystemContainerWrapper()  # pylint: disable=C0103


class FilesystemObjectWrapper(errors.CloudExceptionWrapper):
    """Exception translator."""
    translations = {
        OSError: errors.NoObjectException,
    }
wrap_fs_obj_errors = FilesystemObjectWrapper()  # pylint: disable=C0103


# TODO: Handle empty directory (with non non-dot files).
class FilesystemObject(base.CloudObject):
    """Filesystem object wrapper."""

    def _get_object(self):
        """Return native storage object."""
        return object()

    def _read(self):
        """Return contents of object."""
        return "TODO"

    @property
    def base_path(self):
        """Base absolute path of container."""
        return os.path.join(self.container.base_path, self.name)

    @classmethod
    def from_path(cls, container, path):
        """Create object from path."""
        from datetime import datetime

        path = path.strip(SEP)
        full_path = os.path.join(container.base_path, path)
        last_modified = datetime.fromtimestamp(os.path.getmtime(full_path))
        obj_type = cls.type_cls.SUBDIR if is_dir(full_path)\
            else cls.type_cls.FILE

        return cls(container,
                   name=path,
                   size=os.path.getsize(full_path),
                   content_type=None,
                   last_modified=last_modified,
                   obj_type=obj_type)


class FilesystemContainer(base.CloudContainer):
    """Filesystem container wrapper."""
    obj_cls = FilesystemObject

    def _get_container(self):
        """Return native container object."""
        return object()

    # TODO: get_objects - set limit to DEFAULT_LIMIT
    @wrap_fs_obj_errors
    def get_objects(self, path, marker=None, limit=20):
        """Get objects."""
        def _filter(name):
            """Filter."""
            return (not_dot(name) and
                    (marker is None or
                     os.path.join(path, name).strip(SEP) > marker.strip(SEP)))

        search_path = os.path.join(self.base_path, path)
        all = [self.obj_cls.from_path(self, os.path.join(path, o))
               for o in os.listdir(search_path) if _filter(o)]
        all = sorted(all, key=lambda x: x.base_path)
        return all[:limit]

    @wrap_fs_obj_errors
    def get_object(self, path):
        """Get single object."""
        return self.obj_cls.from_path(self, path)

    @property
    def base_path(self):
        """Base absolute path of container."""
        return os.path.join(self.conn.abs_root, self.name)

    @classmethod
    def from_path(cls, conn, path):
        """Create container from path."""
        path = path.strip(SEP)
        full_path = os.path.join(conn.abs_root, path)
        return cls(conn, path, 0, os.path.getsize(full_path))


class FilesystemConnection(base.CloudConnection):
    """Filesystem connection wrapper."""
    cont_cls = FilesystemContainer

    def __init__(self, root):
        """Initializer."""
        super(FilesystemConnection, self).__init__(None, None)<commit_msg>pylint<commit_after>"""File-system mock cloud wrapper."""
import os
import re

from cloud_browser.cloud import errors, base
from cloud_browser.common import SEP


NO_DOT_RE = re.compile("^[^.]+")


def not_dot(path):
    """Check if non-dot."""
    return NO_DOT_RE.match(os.path.basename(path))


def is_dir(path):
    """Check if non-dot and directory."""
    return not_dot(path) and os.path.isdir(path)


class FilesystemContainerWrapper(errors.CloudExceptionWrapper):
    """Exception translator."""
    translations = {
        OSError: errors.NoContainerException,
    }
wrap_fs_cont_errors = FilesystemContainerWrapper()  # pylint: disable=C0103


class FilesystemObjectWrapper(errors.CloudExceptionWrapper):
    """Exception translator."""
    translations = {
        OSError: errors.NoObjectException,
    }
wrap_fs_obj_errors = FilesystemObjectWrapper()  # pylint: disable=C0103


# TODO: Handle empty directory (with non non-dot files).
class FilesystemObject(base.CloudObject):
    """Filesystem object wrapper."""

    def _get_object(self):
        """Return native storage object."""
        return object()

    def _read(self):
        """Return contents of object."""
        return "TODO"

    @property
    def base_path(self):
        """Base absolute path of container."""
        return os.path.join(self.container.base_path, self.name)

    @classmethod
    def from_path(cls, container, path):
        """Create object from path."""
        from datetime import datetime

        path = path.strip(SEP)
        full_path = os.path.join(container.base_path, path)
        last_modified = datetime.fromtimestamp(os.path.getmtime(full_path))
        obj_type = cls.type_cls.SUBDIR if is_dir(full_path)\
            else cls.type_cls.FILE

        return cls(container,
                   name=path,
                   size=os.path.getsize(full_path),
                   content_type=None,
                   last_modified=last_modified,
                   obj_type=obj_type)


class FilesystemContainer(base.CloudContainer):
    """Filesystem container wrapper."""
    obj_cls = FilesystemObject

    def _get_container(self):
        """Return native container object."""
        return object()

    # TODO: get_objects - set limit to DEFAULT_LIMIT
    @wrap_fs_obj_errors
    def get_objects(self, path, marker=None, limit=20):
        """Get objects."""
        def _filter(name):
            """Filter."""
            return (not_dot(name) and
                    (marker is None or
                     os.path.join(path, name).strip(SEP) > marker.strip(SEP)))

        search_path = os.path.join(self.base_path, path)
        objs = [self.obj_cls.from_path(self, os.path.join(path, o))
                for o in os.listdir(search_path) if _filter(o)]
        objs = sorted(objs, key=lambda x: x.base_path)
        return objs[:limit]

    @wrap_fs_obj_errors
    def get_object(self, path):
        """Get single object."""
        return self.obj_cls.from_path(self, path)

    @property
    def base_path(self):
        """Base absolute path of container."""
        return os.path.join(self.conn.abs_root, self.name)

    @classmethod
    def from_path(cls, conn, path):
        """Create container from path."""
        path = path.strip(SEP)
        full_path = os.path.join(conn.abs_root, path)
        return cls(conn, path, 0, os.path.getsize(full_path))


class FilesystemConnection(base.CloudConnection):
    """Filesystem connection wrapper."""
    cont_cls = FilesystemContainer

    def __init__(self, root):
        """Initializer."""
        super(FilesystemConnection, self).__init__(None, None)
{'fraction_non_alphanumeric': 0.11751784733662822, 'fraction_numerical': 0.0030203185063152112, 'mean_word_length': 2.7474279835390947, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '483558', 'n_tokens_mistral': 2183, 'n_tokens_neox': 2111, 'n_words': 495}