<commit_before>import logging
import os
import typing as tb

from satella.coding import rethrow_as
from .base import BaseSource
from .derivative import MergingSource
from .format import FORMAT_SOURCES, FormatSource
from ...exceptions import ConfigurationError

logger = logging.getLogger(__name__)

__all__ = ['FileSource', 'DirectorySource']

FormatSourceType = tb.Union[tb.Type[FormatSource], str]


class FileSource(BaseSource):
    """
    Try to read a file and parse it with a known format.
    """

    def __init__(self, path: str,
                 encoding: str = 'utf-8',
                 interpret_as: tb.List[FormatSourceType] = FORMAT_SOURCES):
        """
        :param interpret_as: names or classes of format sources to parse with
        """
        super(FileSource, self).__init__()
        from .. import sources
        self.source_classes = [  # list of tp.Type[FormatSource]
            (p if not isinstance(p, str) else getattr(sources, p)) for p in
            interpret_as]
        self.path = path
        self.encoding = encoding

    @rethrow_as((IOError, OSError), ConfigurationError)
    def provide(self) -> dict:

        with open(self.path, 'rb') as fin:
            data = fin.read()

        for source_class in self.source_classes:
            try:
                s = source_class(data, encoding=self.encoding).provide()
                if not isinstance(s, dict):
                    raise ConfigurationError('%s is not a dict instance' % (s,))
                return s
            except ConfigurationError:
                pass
        else:
            raise ConfigurationError('no reader could parse the file')


class DirectorySource(FileSource):
    """
    Load all files from given directory and merge them
    """

    def __init__(self, path, encoding: str = 'utf-8',
                 interpret_as=FORMAT_SOURCES,
                 fname_filter: tb.Callable[[str], bool] = lambda fullpath: True,
                 scan_subdirectories: bool = True):
        """
        :param filter: callable that tells whether to use this file (or subdirectory if scan_subdirectories is enabled)
        """
        super(DirectorySource, self).__init__(path, encoding, interpret_as)
        self.filter = lambda files: filter(fname_filter,
                                           files)  # tp.Callable[[tp.List[str]], tp.List[str]]
        self.scan_subdirectories = scan_subdirectories

    def get_sources_from_directory(self, directory: str) -> tb.List[FileSource]:

        sources = []  # list of  FileSource

        try:
            files = self.filter(os.path.join(directory, x) for x in os.listdir(directory))
        except OSError as e:
            logger.warning(
                'OSError %s while accessing configuration directory %s, skipping files' % (
                    e, directory))
            return []

        for file_name in files:

            fullname = os.path.join(directory, file_name)
            if os.path.isfile(fullname):
                sources.append(FileSource(fullname, encoding=self.encoding,
                                          interpret_as=self.source_classes))
            elif os.path.isdir(fullname) and self.scan_subdirectories:
                sources.extend(self.get_sources_from_directory(fullname))
            else:
                pass  # FIFOs or sockets or something else

        return sources

    def provide(self) -> dict:
        return MergingSource(
            *self.get_sources_from_directory(self.path)).provide()


try:
    import requests
except ModuleNotFoundError:
    pass
else:
    class HTTPJSONSource(BaseSource):
        """Call somwhere, count on a 200-esque code and return a JSON!"""
<commit_msg>#debugbyci<commit_after>import logging
import os
import typing as tb

from satella.coding import rethrow_as
from .base import BaseSource
from .derivative import MergingSource
from .format import FORMAT_SOURCES, FormatSource
from ...exceptions import ConfigurationError

logger = logging.getLogger(__name__)

__all__ = ['FileSource', 'DirectorySource']

FormatSourceType = tb.Union[tb.Type[FormatSource], str]


class FileSource(BaseSource):
    """
    Try to read a file and parse it with a known format.
    """

    def __init__(self, path: str,
                 encoding: str = 'utf-8',
                 interpret_as: tb.List[FormatSourceType] = FORMAT_SOURCES):
        """
        :param interpret_as: names or classes of format sources to parse with
        """
        super(FileSource, self).__init__()
        from .. import sources
        self.source_classes = [  # list of tp.Type[FormatSource]
            (p if not isinstance(p, str) else getattr(sources, p)) for p in
            interpret_as]
        self.path = path
        self.encoding = encoding

    @rethrow_as((IOError, OSError), ConfigurationError)
    def provide(self) -> dict:

        with open(self.path, 'rb') as fin:
            data = fin.read()

        for source_class in self.source_classes:
            try:
                s = source_class(data, encoding=self.encoding).provide()
                if not isinstance(s, dict):
                    raise ConfigurationError('%s is not a dict instance' % (s,))
                logger.warning('Processed %s, returned %s', self.path, s)
                return s
            except ConfigurationError:
                pass
        else:
            raise ConfigurationError('no reader could parse the file')


class DirectorySource(FileSource):
    """
    Load all files from given directory and merge them
    """

    def __init__(self, path, encoding: str = 'utf-8',
                 interpret_as=FORMAT_SOURCES,
                 fname_filter: tb.Callable[[str], bool] = lambda fullpath: True,
                 scan_subdirectories: bool = True):
        """
        :param filter: callable that tells whether to use this file (or subdirectory if scan_subdirectories is enabled)
        """
        super(DirectorySource, self).__init__(path, encoding, interpret_as)
        self.filter = lambda files: filter(fname_filter,
                                           files)  # tp.Callable[[tp.List[str]], tp.List[str]]
        self.scan_subdirectories = scan_subdirectories

    def get_sources_from_directory(self, directory: str) -> tb.List[FileSource]:

        sources = []  # list of  FileSource

        try:
            files = self.filter(os.path.join(directory, x) for x in os.listdir(directory))
        except OSError as e:
            logger.warning(
                'OSError %s while accessing configuration directory %s, skipping files' % (
                    e, directory))
            return []

        for file_name in files:

            fullname = os.path.join(directory, file_name)
            if os.path.isfile(fullname):
                logger.warning('Appending file source %s', fullname)
                sources.append(FileSource(fullname, encoding=self.encoding,
                                          interpret_as=self.source_classes))
            elif os.path.isdir(fullname) and self.scan_subdirectories:
                sources.extend(self.get_sources_from_directory(fullname))
            else:
                pass  # FIFOs or sockets or something else

        return sources

    def provide(self) -> dict:
        return MergingSource(
            *self.get_sources_from_directory(self.path)).provide()


try:
    import requests
except ModuleNotFoundError:
    pass
else:
    class HTTPJSONSource(BaseSource):
        """Call somwhere, count on a 200-esque code and return a JSON!"""

{'fraction_non_alphanumeric': 0.08962513199577614, 'fraction_numerical': 0.0013199577613516368, 'mean_word_length': 2.176939203354298, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '114136', 'n_tokens_mistral': 2063, 'n_tokens_neox': 1974, 'n_words': 583}