<commit_before>"""
:mod:`disco.util` --- Helper functions
======================================

This module provides utility functions that are mostly used by Disco
internally.

The :func:`external` function below comes in handy if you use the Disco
external interface.
"""
import os
import cPickle, marshal, sys, time, traceback

from collections import defaultdict
from itertools import chain, repeat
from types import CodeType, FunctionType
from urllib import urlencode

from disco.comm import download, open_remote
from disco.error import DiscoError, DataError
from disco.events import Message
from disco.settings import DiscoSettings

class DefaultDict(defaultdict):
    """Like a defaultdict, but calls the default_factory with the key argument."""
    def __missing__(self, key):
        return self.default_factory(key)

class MessageWriter(object):
    def write(self, string):
        Message(string)

class netloc(tuple):
    @classmethod
    def parse(cls, netlocstr):
        if ':' in netlocstr:
            return cls(netlocstr.split(':'))
        return cls((netlocstr, ''))

    def __str__((host, port)):
        return '%s:%s' % (host, port) if port else host

def flatten(iterable):
    for item in iterable:
        if isiterable(item):
            for subitem in flatten(item):
                yield subitem
        else:
            yield item

def isiterable(object):
    return hasattr(object, '__iter__')

def iskv(object):
    return isinstance(object, tuple) and len(object) is 2

def iterify(object):
    if hasattr(object, '__iter__'):
        return object
    return repeat(object, 1)

def ilen(iter):
    return sum(1 for _ in iter)

def partition(iterable, fn):
    t, f = [], []
    for item in iterable:
        (t if fn(item) else f).append(item)
    return t, f

def rapply(iterable, fn):
    for item in iterable:
        if isiterable(item):
            yield rapply(item, fn)
        else:
            yield fn(item)

def pack(object):
    if hasattr(object, 'func_code'):
        return marshal.dumps(object.func_code)
    return cPickle.dumps(object)

def unpack(string, globals={}):
    try:
        return cPickle.loads(string)
    except Exception:
        return FunctionType(marshal.loads(string), globals)

def pack_stack(stack):
    return pack([pack(object) for object in stack])

def unpack_stack(stackstring, globals={}):
    return [unpack(string, globals=globals) for string in unpack(stackstring)]

def schemesplit(url):
    return url.split('://', 1) if '://' in url else ('file', url)

def urlsplit(url):
    scheme, rest = schemesplit(url)
    locstr, path = rest.split('/', 1)  if '/'   in rest else (rest ,'')
    if scheme == 'disco':
        scheme = 'http'
        locstr = '%s:%s' % (locstr, DiscoSettings()['DISCO_PORT'])
    return scheme, netloc.parse(locstr), path

def urlresolve(url):
    return '%s://%s/%s' % urlsplit(url)

def urllist(url, partid=None, listdirs=True, ddfs=None):
    if isiterable(url):
        return [list(url)]
    scheme, netloc, path = urlsplit(url)
    if scheme == 'dir' and listdirs:
        return parse_dir(url, partid=partid)
    elif scheme == 'tag':
        from disco.ddfs import DDFS<commit_msg>added support for keyword arguments in job functions<commit_after>"""
:mod:`disco.util` --- Helper functions
======================================

This module provides utility functions that are mostly used by Disco
internally.

The :func:`external` function below comes in handy if you use the Disco
external interface.
"""
import os
import cPickle, marshal, sys, time, traceback

from collections import defaultdict
from itertools import chain, repeat
from types import CodeType, FunctionType
from urllib import urlencode

from disco.comm import download, open_remote
from disco.error import DiscoError, DataError
from disco.events import Message
from disco.settings import DiscoSettings

class DefaultDict(defaultdict):
    """Like a defaultdict, but calls the default_factory with the key argument."""
    def __missing__(self, key):
        return self.default_factory(key)

class MessageWriter(object):
    def write(self, string):
        Message(string)

class netloc(tuple):
    @classmethod
    def parse(cls, netlocstr):
        if ':' in netlocstr:
            return cls(netlocstr.split(':'))
        return cls((netlocstr, ''))

    def __str__((host, port)):
        return '%s:%s' % (host, port) if port else host

def flatten(iterable):
    for item in iterable:
        if isiterable(item):
            for subitem in flatten(item):
                yield subitem
        else:
            yield item

def isiterable(object):
    return hasattr(object, '__iter__')

def iskv(object):
    return isinstance(object, tuple) and len(object) is 2

def iterify(object):
    if hasattr(object, '__iter__'):
        return object
    return repeat(object, 1)

def ilen(iter):
    return sum(1 for _ in iter)

def partition(iterable, fn):
    t, f = [], []
    for item in iterable:
        (t if fn(item) else f).append(item)
    return t, f

def rapply(iterable, fn):
    for item in iterable:
        if isiterable(item):
            yield rapply(item, fn)
        else:
            yield fn(item)

def pack(object):
    if hasattr(object, 'func_code'):
        defs = [pack(x) for x in object.func_defaults]\
                    if object.func_defaults else None
        return marshal.dumps((object.func_code, defs))
    return cPickle.dumps(object)

def unpack(string, globals={}):
    try:
        return cPickle.loads(string)
    except Exception:
        code, defs = marshal.loads(string)
        defs = tuple([unpack(x) for x in defs]) if defs else None
        return FunctionType(code, globals, argdefs = defs)

def pack_stack(stack):
    return pack([pack(object) for object in stack])

def unpack_stack(stackstring, globals={}):
    return [unpack(string, globals=globals) for string in unpack(stackstring)]

def schemesplit(url):
    return url.split('://', 1) if '://' in url else ('file', url)

def urlsplit(url):
    scheme, rest = schemesplit(url)
    locstr, path = rest.split('/', 1)  if '/'   in rest else (rest ,'')
    if scheme == 'disco':
        scheme = 'http'
        locstr = '%s:%s' % (locstr, DiscoSettings()['DISCO_PORT'])
    return scheme, netloc.parse(locstr), path

def urlresolve(url):
    return '%s://%s/%s' % urlsplit(url)

def urllist(url, partid=None, listdirs=True, ddfs=None):
    if isiterable(url):
        return [list(url)]
    scheme, netloc, path = urlsplit(url)
    if scheme == 'dir' and listdirs:
        return parse_dir(url, partid=partid)
    elif scheme == 'tag':
        from disco.ddfs import DDFS
{'fraction_non_alphanumeric': 0.1244191275670814, 'fraction_numerical': 0.00149902563333833, 'mean_word_length': 3.2254591513616213, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7448144', 'n_tokens_mistral': 2068, 'n_tokens_neox': 2059, 'n_words': 654}