<commit_before>import glob
import multiprocessing
import os
import random
import re
import string
import subprocess
import xml.dom.minidom

from contextlib import contextmanager
from functools import partial, wraps

import Chirp as chirp

logger = multiprocessing.get_logger()

class StorageElement(object):
    """Weird class to handle all needs of storage implementations.

    This class can be used for file system operations after at least one of
    its subclasses has been instantiated.

    "One size fits nobody." (<NAME>)
    """
    _defaults = []
    _systems = []

    def __init__(self, pfnprefix=None):
        """Create or use a StorageElement abstraction.

        As a user, use with no parameters to access various storage
        elements transparently.

        Subclasses should call the constructor with appropriate arguments,
        which should also be made available to the user.

        Parameters
        ----------
        pfnprefix : string, optional
            The path prefix under which relative file names can be
            accessed.
        """
        self.__master = False

        if pfnprefix is not None:
            self._pfnprefix = pfnprefix
            StorageElement._systems.append(self)
        else:
            self.__master = True

    def __getattr__(self, attr):
        if attr in self.__dict__ or not self.__master:
            return self.__dict__[attr]

        def switch(path, *args):
            for imp in StorageElement._systems:
                try:
                    return imp.fixresult(getattr(imp, attr)(imp.lfn2pfn(path), *args))
                except (IOError, OSError):
                    pass
            raise AttributeError("no path resolution found for '{0}'".format(path))
        return switch

    def lfn2pfn(self, path):
        if path.startswith('/'):
            return os.path.join(self._pfnprefix, path[1:])
        return os.path.join(self._pfnprefix, path)

    def fixresult(self, res):
        def pfn2lfn(p):
            return p.replace(self._pfnprefix, '', 1)

        if isinstance(res, str):
            return pfn2lfn(res)

        try:
            return map(pfn2lfn, res)
        except TypeError:
            return res

    def makedirs(self, path):
        parent = os.path.dirname(path)
        if parent != '' and not self.exists(parent):
            self.makedirs(parent)
        mode = self.permissions(parent)
        self.mkdir(path, mode)

    @classmethod
    def reset(cls):
        cls._systems = []

    @classmethod
    def store(cls):
        cls._defaults = list(cls._systems)

    @contextmanager
    def default(self):
        tmp = StorageElement._systems
        StorageElement._systems = self._defaults
        try:
            yield
        finally:
            StorageElement._systems = tmp

class Local(StorageElement):
    def __init__(self, pfnprefix=''):
        super(Local, self).__init__(pfnprefix)
        self.exists = os.path.exists
        self.getsize = os.path.getsize
        self.isdir = self._guard(os.path.isdir)
        self.isfile = self._guard(os.path.isfile)
        self.remove = os.remove

    def _guard(self, method):
        """Protect method against non-existent paths.
        """
        def guarded(path):
            if not os.path.exists(path):
                raise IOError()
            return method(path)
        return guarded

    def ls(self, path):
        for fn in os.listdir(path):
            yield os.path.join(path, fn)

    def mkdir(self, path, mode=None):
        os.mkdir(path)
        if mode:
            os.chmod(path, mode)

    def permissions(self, path):
        return os.stat(path).st_mode & 0777

class Hadoop(StorageElement):
    def __init__(self, pfnprefix='/hadoop'):
        super(Hadoop, self).__init__(pfnprefix)

    def __execute(self, cmd, path, safe=False):
        cmds = cmd.split()
        args = ['hadoop', 'fs', '-' + cmds[0]] + cmds[1:] + [path]
        try:
            p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            p.wait()
            if p.returncode != 0 and not safe:
                msg = "Failed to execute '{0}':\n{1}\n{2}".format(' '.join(args), p.stderr.read(), p.stdout.read())
                raise IOError(msg)
        except OSError:
            raise AttributeError("hadoop filesystem utilities not available")
        return p.stdout.read()

        self.mkdir = hadoopy.mkdir
        self.remove = hadoopy.rmr

    def exists(self, path):
        try:
            self.__execute('stat', path)
            return True
        except IOError:
            return False

    def getsize(self, path):
        return int(self.__execute('stat %b', path))

    def isdir(self, path):
        return self.__execute('stat %F', path) == 'directory'

    def isfile(self, path):
        return self.__execute('stat %F', path) == 'regular file'

    def ls(self, path):
        for line in self.__execute('ls', path).splitlines()[1:]:
            yield line.split(None, 7)[-1]

    def mkdir(self, path, mode):
        self.__execute('mkdir', path)
        self.__execute('chmod ' + oct(mode), path)

    def permissions(self, path):
        output = self.__execute('ls -d', path).splitlines()[1].split(None, 1)[0]
        tr = string.maketrans('rwx-', '1110')
        # first character of output is either 'd' or '-' and needs to be
        # removed
        return int(output[1:].translate(tr), 2)

class Chirp(StorageElement):
    def __init__(self, server, pfnprefix):
        super(Chirp, self).__init__(pfnprefix)

        self.__c = chirp.Client(server, timeout=10)
<commit_msg>Fix hadoop backend.<commit_after>import glob
import multiprocessing
import os
import random
import re
import string
import subprocess
import xml.dom.minidom

from contextlib import contextmanager
from functools import partial, wraps

import Chirp as chirp

logger = multiprocessing.get_logger()

class StorageElement(object):
    """Weird class to handle all needs of storage implementations.

    This class can be used for file system operations after at least one of
    its subclasses has been instantiated.

    "One size fits nobody." (<NAME>)
    """
    _defaults = []
    _systems = []

    def __init__(self, pfnprefix=None):
        """Create or use a StorageElement abstraction.

        As a user, use with no parameters to access various storage
        elements transparently.

        Subclasses should call the constructor with appropriate arguments,
        which should also be made available to the user.

        Parameters
        ----------
        pfnprefix : string, optional
            The path prefix under which relative file names can be
            accessed.
        """
        self.__master = False

        if pfnprefix is not None:
            self._pfnprefix = pfnprefix
            StorageElement._systems.append(self)
        else:
            self.__master = True

    def __getattr__(self, attr):
        if attr in self.__dict__ or not self.__master:
            return self.__dict__[attr]

        def switch(path, *args):
            for imp in StorageElement._systems:
                try:
                    return imp.fixresult(getattr(imp, attr)(imp.lfn2pfn(path), *args))
                except (IOError, OSError):
                    pass
            raise AttributeError("no path resolution found for '{0}'".format(path))
        return switch

    def lfn2pfn(self, path):
        if path.startswith('/'):
            return os.path.join(self._pfnprefix, path[1:])
        return os.path.join(self._pfnprefix, path)

    def fixresult(self, res):
        def pfn2lfn(p):
            return p.replace(self._pfnprefix, '', 1)

        if isinstance(res, str):
            return pfn2lfn(res)

        try:
            return map(pfn2lfn, res)
        except TypeError:
            return res

    def makedirs(self, path):
        parent = os.path.dirname(path)
        if parent != '' and not self.exists(parent):
            self.makedirs(parent)
        mode = self.permissions(parent)
        self.mkdir(path, mode)

    @classmethod
    def reset(cls):
        cls._systems = []

    @classmethod
    def store(cls):
        cls._defaults = list(cls._systems)

    @contextmanager
    def default(self):
        tmp = StorageElement._systems
        StorageElement._systems = self._defaults
        try:
            yield
        finally:
            StorageElement._systems = tmp

class Local(StorageElement):
    def __init__(self, pfnprefix=''):
        super(Local, self).__init__(pfnprefix)
        self.exists = os.path.exists
        self.getsize = os.path.getsize
        self.isdir = self._guard(os.path.isdir)
        self.isfile = self._guard(os.path.isfile)
        self.remove = os.remove

    def _guard(self, method):
        """Protect method against non-existent paths.
        """
        def guarded(path):
            if not os.path.exists(path):
                raise IOError()
            return method(path)
        return guarded

    def ls(self, path):
        for fn in os.listdir(path):
            yield os.path.join(path, fn)

    def mkdir(self, path, mode=None):
        os.mkdir(path)
        if mode:
            os.chmod(path, mode)

    def permissions(self, path):
        return os.stat(path).st_mode & 0777

class Hadoop(StorageElement):
    def __init__(self, pfnprefix='/hadoop'):
        super(Hadoop, self).__init__(pfnprefix)

    def __execute(self, cmd, path, safe=False):
        cmds = cmd.split()
        args = ['hadoop', 'fs', '-' + cmds[0]] + cmds[1:] + [path]
        try:
            p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            p.wait()
            if p.returncode != 0 and not safe:
                msg = "Failed to execute '{0}':\n{1}\n{2}".format(' '.join(args), p.stderr.read(), p.stdout.read())
                raise IOError(msg)
        except OSError:
            raise AttributeError("hadoop filesystem utilities not available")
        return p.stdout.read()

        self.mkdir = hadoopy.mkdir
        self.remove = hadoopy.rmr

    def exists(self, path):
        try:
            self.__execute('stat', path)
            return True
        except IOError:
            return False

    def getsize(self, path):
        return int(self.__execute('stat %b', path))

    def isdir(self, path):
        return self.__execute('stat %F', path).strip() == 'directory'

    def isfile(self, path):
        return self.__execute('stat %F', path).strip() == 'regular file'

    def ls(self, path):
        for line in self.__execute('ls', path).splitlines()[1:]:
            yield line.split(None, 7)[-1]

    def mkdir(self, path, mode):
        self.__execute('mkdir', path)
        self.__execute('chmod ' + oct(mode), path)

    def permissions(self, path):
        output = self.__execute('ls -d', path).splitlines()[1].split(None, 1)[0]
        tr = string.maketrans('rwx-', '1110')
        # first character of output is either 'd' or '-' and needs to be
        # removed
        return int(output[1:].translate(tr), 2)

class Chirp(StorageElement):
    def __init__(self, server, pfnprefix):
        super(Chirp, self).__init__(pfnprefix)

        self.__c = chirp.Client(server, timeout=10)

{'fraction_non_alphanumeric': 0.11009743135518157, 'fraction_numerical': 0.00566873339238264, 'mean_word_length': 2.1329078801331853, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '565024', 'n_tokens_mistral': 3340, 'n_tokens_neox': 3197, 'n_words': 904}