<commit_before># helpers to move files/dirs to and from ZK and also among ZK clusters

from __future__ import print_function

from collections import defaultdict, namedtuple
import json
import os
import re
import urlparse

from kazoo.client import KazooClient


DEFAULT_ZK_PORT = 2181


def zk_client(host, username, password):
    if not re.match(":\d+$", host):
        hostname = "%s:%d" % (host, DEFAULT_ZK_PORT)

    client = KazooClient(hosts=host)
    client.start()

    # TODO: handle more than just digest
    # TODO: add_auth() isn't truly synchronous!
    if username != "":
        client.add_auth("digest", "%s:%s" % (username, password))

    return client


class Netloc(namedtuple("Netloc", "username password host")):
    @classmethod
    def from_string(cls, netloc_string):
        username = password = host = ""
        if not "@" in netloc_string:
            host = netloc_string
        else:
            username_passwd, host =  netloc_string.rsplit("@", 1)
            if ":" in username_passwd:
                username, password = username_<PASSWORD>.split(":", 1)
            else:
                username = username_passwd

        return cls(username, password, host)


class CopyError(Exception): pass


class PathValue(object):
    def __init__(self, value):
        self._value = value

    @property
    def value(self):
        return self._value


class ProxyType(type):
    TYPES = {}
    SCHEME = ""

    def __new__(cls, clsname, bases, dct):
        obj = super(ProxyType, cls).__new__(cls, clsname, bases, dct)
        if obj.SCHEME in cls.TYPES:
            raise ValueError("Duplicate scheme handler: %s" % obj.SCHEME)

        if obj.SCHEME != "": cls.TYPES[obj.SCHEME] = obj
        return obj


class Proxy(ProxyType("ProxyBase", (object,), {})):
    SCHEME = ""

    def __init__(self, parse_result, exists):
        self.parse_result = parse_result
        self.netloc = Netloc.from_string(parse_result.netloc)
        self.exists = exists

    @property
    def scheme(self):
        return self.parse_result.scheme

    @property
    def url(self):
        return self.parse_result.geturl()

    @property
    def path(self):
        return self.parse_result.path

    @property
    def host(self):
        return self.netloc.host

    @property
    def username(self):
        return self.netloc.username

    @property
    def password(self):
        return self.netloc.password

    def set_url(self, string):
        """ useful for recycling a stateful proxy """
        self.parse_result = Proxy.parse(string)

    @classmethod
    def from_string(cls, string, exists):
        """
        if exists is bool, then check it either exists or it doesn't.
        if exists is None, we don't care.
        """
        result = cls.parse(string)

        if result.scheme not in cls.TYPES:
            raise CopyError("Invalid scheme: %s" % (result.scheme))

        return cls.TYPES[result.scheme](result, exists)

    @classmethod
    def parse(cls, url_string):
        """ default to file:// """
        if not re.match("^\w+://", url_string):
            url_string = "file://%s" % (url_string)

        return urlparse.urlparse(url_string)

    def __enter__(self):
        pass

    def __exit__(self, type, value, traceback):
        pass

    def check_path(self):
        raise NotImplementedError, "check_path must be implemented"

    def read_path(self):
        raise NotImplementedError, "read_path must be implemented"

    def write_path(self, path_value):
        raise NotImplementedError, "write_path must be implemented"

    def children_of(self):
        raise NotImplementedError, "children_of must be implemented"


class ZKProxy(Proxy):
    """ read/write ZooKeeper paths """

    SCHEME = "zk"

    class ZKPathValue(PathValue):
        def __init__(self, value, acl=None):
            PathValue.__init__(self, value)
            self._acl = acl

        @property
        def acl(self):
            return self._acl

    def __init__(self, parse_result, exists):
        super(ZKProxy, self).__init__(parse_result, exists)

        self.client = zk_client(self.host, self.username, self.password)

        if exists is not None:
            self.check_path()

    def check_path(self):
        retval = True if self.client.exists(self.path) else False
        if retval is not self.exists:
            if self.exists:
                m = "znode %s in %s doesn't exist" % \
                    (self.path, self.host)
            else:<commit_msg>ZKProxy, which is a cxt mgr, should connect/stop when entering/leaving<commit_after># helpers to move files/dirs to and from ZK and also among ZK clusters

from __future__ import print_function

from collections import defaultdict, namedtuple
import json
import os
import re
import urlparse

from kazoo.client import KazooClient


DEFAULT_ZK_PORT = 2181


def zk_client(host, username, password):
    if not re.match(":\d+$", host):
        hostname = "%s:%d" % (host, DEFAULT_ZK_PORT)

    client = KazooClient(hosts=host)
    client.start()

    # TODO: handle more than just digest
    # TODO: add_auth() isn't truly synchronous!
    if username != "":
        client.add_auth("digest", "%s:%s" % (username, password))

    return client


class Netloc(namedtuple("Netloc", "username password host")):
    @classmethod
    def from_string(cls, netloc_string):
        username = password = host = ""
        if not "@" in netloc_string:
            host = netloc_string
        else:
            username_passwd, host =  netloc_string.rsplit("@", 1)
            if ":" in username_passwd:
                username, password = username_passwd.split(":", 1)
            else:
                username = username_passwd

        return cls(username, password, host)


class CopyError(Exception): pass


class PathValue(object):
    def __init__(self, value):
        self._value = value

    @property
    def value(self):
        return self._value


class ProxyType(type):
    TYPES = {}
    SCHEME = ""

    def __new__(cls, clsname, bases, dct):
        obj = super(ProxyType, cls).__new__(cls, clsname, bases, dct)
        if obj.SCHEME in cls.TYPES:
            raise ValueError("Duplicate scheme handler: %s" % obj.SCHEME)

        if obj.SCHEME != "": cls.TYPES[obj.SCHEME] = obj
        return obj


class Proxy(ProxyType("ProxyBase", (object,), {})):
    SCHEME = ""

    def __init__(self, parse_result, exists):
        self.parse_result = parse_result
        self.netloc = Netloc.from_string(parse_result.netloc)
        self.exists = exists

    @property
    def scheme(self):
        return self.parse_result.scheme

    @property
    def url(self):
        return self.parse_result.geturl()

    @property
    def path(self):
        return self.parse_result.path

    @property
    def host(self):
        return self.netloc.host

    @property
    def username(self):
        return self.netloc.username

    @property
    def password(self):
        return <PASSWORD>

    def set_url(self, string):
        """ useful for recycling a stateful proxy """
        self.parse_result = Proxy.parse(string)

    @classmethod
    def from_string(cls, string, exists):
        """
        if exists is bool, then check it either exists or it doesn't.
        if exists is None, we don't care.
        """
        result = cls.parse(string)

        if result.scheme not in cls.TYPES:
            raise CopyError("Invalid scheme: %s" % (result.scheme))

        return cls.TYPES[result.scheme](result, exists)

    @classmethod
    def parse(cls, url_string):
        """ default to file:// """
        if not re.match("^\w+://", url_string):
            url_string = "file://%s" % (url_string)

        return urlparse.urlparse(url_string)

    def __enter__(self):
        pass

    def __exit__(self, type, value, traceback):
        pass

    def check_path(self):
        raise NotImplementedError, "check_path must be implemented"

    def read_path(self):
        raise NotImplementedError, "read_path must be implemented"

    def write_path(self, path_value):
        raise NotImplementedError, "write_path must be implemented"

    def children_of(self):
        raise NotImplementedError, "children_of must be implemented"


class ZKProxy(Proxy):
    """ read/write ZooKeeper paths """

    SCHEME = "zk"

    class ZKPathValue(PathValue):
        def __init__(self, value, acl=None):
            PathValue.__init__(self, value)
            self._acl = acl

        @property
        def acl(self):
            return self._acl

    def __init__(self, parse_result, exists):
        super(ZKProxy, self).__init__(parse_result, exists)
        self.client = None

    def connect(self):
        self.client = zk_client(self.host, self.username, self.password)

        if self.exists is not None:
            self.check_path()

    def __enter__(self):
        self.connect()

    def __exit__(self, type, value, traceback):
        if self.client:
            self.client.stop()

    def check_path(self):
        retval = True if self.client.exists(self.path) else False
        if retval is not self.exists:
            if self.exists:
                m = "znode %s in %s doesn't exist" % \
                    (self.path, self.host)
            else:
{'fraction_non_alphanumeric': 0.10491239385144577, 'fraction_numerical': 0.0012899064817800709, 'mean_word_length': 2.398100803506209, 'pattern_counts': {'":': 10, '<': 5, '<?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': '3989478', 'n_tokens_mistral': 2844, 'n_tokens_neox': 2755, 'n_words': 763}