<commit_before># -*- coding: utf-8 -*-
"Main module defining filesystem and file classes"
import os
import ctypes
import sys
import subprocess
import warnings
import fnmatch
PY3 = sys.version_info.major > 2
here = os.path.dirname(os.path.abspath(__file__))
_lib = ctypes.cdll.LoadLibrary(os.sep.join([here, 'libhdfs3.so']))


def get_default_host():
    "Try to guess the namenode by looking in this machine's hadoop conf."
    confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                           '') + '/hadoop/conf')
    host = open(os.sep.join([confd, 'masters'])).readlines()[1][:-1]
    return host


def ensure_byte(s):
    "Give strings that ctypes is guaranteed to handle"
    if PY3:
        if isinstance(s, str):
            return s.encode('ascii')
        elif isinstance(s, bytes):
            return s
        else:
            raise ValueError('Could not convert %s to bytes' % s)
    else:  # in PY2, strings are fine for ctypes
        return s


def init_kerb():
    """Uses system kinit to find credentials. Set up by editing
    krb5.conf"""
    raise NotImplementedError("Please set your credentials manually")
    out1 = subprocess.check_call(['kinit'])
    out2 = subprocess.check_call(['klist'])
    HDFileSystem.ticket_cache = None
    HDFileSystem.token = None


class HDFileSystem():
    "A connection to an HDFS namenode"
    _handle = None
    host = get_default_host()
    port = 9000
    user = None
    ticket_cache = None
    token = None
    pars = {}
    _token = None  # Delegation token (generated)
    autoconnect = True

    def __init__(self, **kwargs):
        """
        Parameters
        ----------

        host : str (default from config files)
            namenode (name or IP)

        port : int (9000)
            connection port

        user, ticket_cache, token : str
            kerberos things

        pars : {str: str}
            other parameters for hadoop
        """
        for arg in kwargs:
            setattr(self, arg, kwargs[arg])
        # self.__dict__.update(kwargs)
        if self.autoconnect:
            self.connect()

    def connect(self):
        assert self._handle is None, "Already connected"
        o = _lib.hdfsNewBuilder()
        _lib.hdfsBuilderSetNameNodePort(o, self.port)
        _lib.hdfsBuilderSetNameNode(o, ensure_byte(self.host))
        if self.user:
            _lib.hdfsBuilderSetUserName(o, ensure_byte(self.user))
        if self.ticket_cache:
            _lib.hdfsBuilderSetKerbTicketCachePath(o, ensure_byte(self.ticket_cache))
        if self.token:
            _lib.hdfsBuilderSetToken(o, ensure_byte(self.token))
        if self.pars:
            for par in self.pars:
                try:
                    assert _lib.hdfsBuilderConfSetStr(o, ensure_byte(par),
                                          ensure_byte(self.pars(par))) == 0
                except AssertionError:
                    warnings.warn('Setting conf parameter %s failed' % par)
        fs = _lib.hdfsBuilderConnect(o)
        if fs:
            self._handle = fs
            #if self.token:   # TODO: find out what a delegation token is
            #    self._token = _lib.hdfsGetDelegationToken(self._handle,
            #                                             ensure_byte(self.user))
        else:
            raise RuntimeError('Connection Failed')

    def disconnect(self):
        if self._handle:
            _lib.hdfsDisconnect(self._handle)
        self._handle = None

    def open(self, path, mode='r', **kwargs):
        assert self._handle, "Filesystem not connected"
        return HDFile(self, path, mode, **kwargs)

    def du(self, path, total=False, deep=False):
        if isinstance(total, str):
            total = total=='True'
        if isinstance(deep, str):
            deep = deep=='True'
        fi = self.ls(path)
        if len(fi) == 0:
            raise IOError('Not Found')
        if deep:
            for apath in fi:
                if apath['kind'] == 68:  # directory
                    fi.extend(self.ls(apath['name']))
        if total:
            return {path: sum(f['size'] for f in fi)}
        return {p['name']: p['size'] for p in fi}

    def df(self):
        cap = _lib.hdfsGetCapacity(self._handle)
        used = _lib.hdfsGetUsed(self._handle)
        return {'capacity': cap, 'used': used, 'free%': 100*(cap-used)/cap}

    def get_block_locations(self, path, start=0, length=0):
        "Fetch physical locations of blocks"
        assert self._handle, "Filesystem not connected"
        start = int(start) or 0
        length = int(length) or self.info(path)['size']
        nblocks = ctypes.c_int(0)
        out = _lib.hdfsGetFileBlockLocations(self._handle, ensure_byte(path),
                                ctypes.c_int64(start), ctypes.c_int64(length),
                                ctypes.byref(nblocks))
        locs = []
        for i in range(nblocks.value):
            block = out[i]
            hosts = [block.hosts[i] for i in
                     range(block.numOfNodes)]
            locs.append({'hosts': hosts, 'length': block.length,
                         'offset': block.offset})
        _lib.hdfsFreeFileBlockLocations(out, nblocks)
        return locs

    def info(self, path):
        "File information"
        fi = _lib.hdfsGetPathInfo(self._handle, ensure_byte(path)).contents
        out = struct_to_dict(fi)
        _lib.hdfsFreeFileInfo(ctypes.byref(fi), 1)
        return out

    def glob(self, path):
        allfiles = self.du('/', False, True).keys()
        return fnmatch.filter(allfiles, path)

    def ls(self, path):
        num = ctypes.c_int(0)
        fi = _lib.hdfsListDirectory(self._handle, ensure_byte(path), ctypes.byref(num))
        out = [struct_to_dict(fi[i]) for i in range(num.value)]
        _lib.hdfsFreeFileInfo(fi, num.value)
        return out

    def __repr__(self):
        state = ['Disconnected', 'Connected'][self._handle is not None]
        return 'hdfs://%s:%s, %s' % (self.host, self.port, state)<commit_msg>Avoid bytes when matching filenames<commit_after># -*- coding: utf-8 -*-
"Main module defining filesystem and file classes"
import os
import ctypes
import sys
import subprocess
import warnings
import fnmatch
PY3 = sys.version_info.major > 2
here = os.path.dirname(os.path.abspath(__file__))
_lib = ctypes.cdll.LoadLibrary(os.sep.join([here, 'libhdfs3.so']))


def get_default_host():
    "Try to guess the namenode by looking in this machine's hadoop conf."
    confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                           '') + '/hadoop/conf')
    host = open(os.sep.join([confd, 'masters'])).readlines()[1][:-1]
    return host


def ensure_byte(s):
    "Give strings that ctypes is guaranteed to handle"
    if PY3:
        if isinstance(s, str):
            return s.encode('ascii')
        elif isinstance(s, bytes):
            return s
        else:
            raise ValueError('Could not convert %s to bytes' % s)
    else:  # in PY2, strings are fine for ctypes
        return s


def init_kerb():
    """Uses system kinit to find credentials. Set up by editing
    krb5.conf"""
    raise NotImplementedError("Please set your credentials manually")
    out1 = subprocess.check_call(['kinit'])
    out2 = subprocess.check_call(['klist'])
    HDFileSystem.ticket_cache = None
    HDFileSystem.token = None


class HDFileSystem():
    "A connection to an HDFS namenode"
    _handle = None
    host = get_default_host()
    port = 9000
    user = None
    ticket_cache = None
    token = None
    pars = {}
    _token = None  # Delegation token (generated)
    autoconnect = True

    def __init__(self, **kwargs):
        """
        Parameters
        ----------

        host : str (default from config files)
            namenode (name or IP)

        port : int (9000)
            connection port

        user, ticket_cache, token : str
            kerberos things

        pars : {str: str}
            other parameters for hadoop
        """
        for arg in kwargs:
            setattr(self, arg, kwargs[arg])
        # self.__dict__.update(kwargs)
        if self.autoconnect:
            self.connect()

    def connect(self):
        assert self._handle is None, "Already connected"
        o = _lib.hdfsNewBuilder()
        _lib.hdfsBuilderSetNameNodePort(o, self.port)
        _lib.hdfsBuilderSetNameNode(o, ensure_byte(self.host))
        if self.user:
            _lib.hdfsBuilderSetUserName(o, ensure_byte(self.user))
        if self.ticket_cache:
            _lib.hdfsBuilderSetKerbTicketCachePath(o, ensure_byte(self.ticket_cache))
        if self.token:
            _lib.hdfsBuilderSetToken(o, ensure_byte(self.token))
        if self.pars:
            for par in self.pars:
                try:
                    assert _lib.hdfsBuilderConfSetStr(o, ensure_byte(par),
                                          ensure_byte(self.pars(par))) == 0
                except AssertionError:
                    warnings.warn('Setting conf parameter %s failed' % par)
        fs = _lib.hdfsBuilderConnect(o)
        if fs:
            self._handle = fs
            #if self.token:   # TODO: find out what a delegation token is
            #    self._token = _lib.hdfsGetDelegationToken(self._handle,
            #                                             ensure_byte(self.user))
        else:
            raise RuntimeError('Connection Failed')

    def disconnect(self):
        if self._handle:
            _lib.hdfsDisconnect(self._handle)
        self._handle = None

    def open(self, path, mode='r', **kwargs):
        assert self._handle, "Filesystem not connected"
        return HDFile(self, path, mode, **kwargs)

    def du(self, path, total=False, deep=False):
        if isinstance(total, str):
            total = total=='True'
        if isinstance(deep, str):
            deep = deep=='True'
        fi = self.ls(path)
        if len(fi) == 0:
            raise IOError('Not Found')
        if deep:
            for apath in fi:
                if apath['kind'] == 68:  # directory
                    fi.extend(self.ls(apath['name']))
        if total:
            return {path: sum(f['size'] for f in fi)}
        return {p['name']: p['size'] for p in fi}

    def df(self):
        cap = _lib.hdfsGetCapacity(self._handle)
        used = _lib.hdfsGetUsed(self._handle)
        return {'capacity': cap, 'used': used, 'free%': 100*(cap-used)/cap}

    def get_block_locations(self, path, start=0, length=0):
        "Fetch physical locations of blocks"
        assert self._handle, "Filesystem not connected"
        start = int(start) or 0
        length = int(length) or self.info(path)['size']
        nblocks = ctypes.c_int(0)
        out = _lib.hdfsGetFileBlockLocations(self._handle, ensure_byte(path),
                                ctypes.c_int64(start), ctypes.c_int64(length),
                                ctypes.byref(nblocks))
        locs = []
        for i in range(nblocks.value):
            block = out[i]
            hosts = [block.hosts[i] for i in
                     range(block.numOfNodes)]
            locs.append({'hosts': hosts, 'length': block.length,
                         'offset': block.offset})
        _lib.hdfsFreeFileBlockLocations(out, nblocks)
        return locs

    def info(self, path):
        "File information"
        fi = _lib.hdfsGetPathInfo(self._handle, ensure_byte(path)).contents
        out = struct_to_dict(fi)
        _lib.hdfsFreeFileInfo(ctypes.byref(fi), 1)
        return out

    def glob(self, path):
        allfiles = self.du('/', False, True).keys()
        out = [f for f in allfiles if fnmatch.fnmatch(f.decode(), path)]
        return out

    def ls(self, path):
        num = ctypes.c_int(0)
        fi = _lib.hdfsListDirectory(self._handle, ensure_byte(path), ctypes.byref(num))
        out = [struct_to_dict(fi[i]) for i in range(num.value)]
        _lib.hdfsFreeFileInfo(fi, num.value)
        return out

    def __repr__(self):
        state = ['Disconnected', 'Connected'][self._handle is not None]
        return 'hdfs://%s:%s, %s' % (self.host, self.port, state)
{'fraction_non_alphanumeric': 0.11289793239251723, 'fraction_numerical': 0.00590744995077125, 'mean_word_length': 2.0480120030007503, '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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5583876', 'n_tokens_mistral': 3731, 'n_tokens_neox': 3617, 'n_words': 925}