<commit_before>'''
Common utilities
'''
import threading as th
import httplib
import logging

logger = logging.getLogger(__name__)


class Drain(th.Thread):
    """
    Drain a generator to a destination that answers to put(), in a thread
    """

    def __init__(self, source, destination):
        super(Drain, self).__init__()
        self.source = source
        self.destination = destination
        self._finished = th.Event()
        self._interrupted = th.Event()

    def run(self):
        for item in self.source:
            self.destination.put(item)
            if self._interrupted.is_set():
                break
        self._finished.set()

    def wait(self, timeout=None):
        self._finished.wait(timeout=timeout)

    def close(self):
        self._interrupted.set()

### HTTP codes
HTTP = httplib.responses

### Extended list of HTTP status codes(WEBdav etc.)
### HTTP://en.wikipedia.org/wiki/List_of_HTTP_status_codes
WEBDAV = {
    102: 'Processing',
    103: 'Checkpoint',
    122: 'Request-URI too long',
    207: 'Multi-Status',
    226: 'IM Used',
    308: 'Resume Incomplete',
    418: 'I\'m a teapot',
    422: 'Unprocessable Entity',
    423: 'Locked',
    424: 'Failed Dependency',
    425: 'Unordered Collection',
    426: 'Upgrade Required',
    444: 'No Response',
    449: 'Retry With',
    450: 'Blocked by Windows Parental Controls',
    499: 'Client Closed Request',
    506: 'Variant Also Negotiates',
    507: 'Insufficient Storage',
    509: 'Bandwidth Limit Exceeded',
    510: 'Not Extended',
    598: 'network read timeout error',
    599: 'network connect timeout error',
    999: 'Common Failure',<commit_msg>securedshell class moved to ...core.util<commit_after>'''
Common utilities
'''
import threading as th
import httplib
import logging

logger = logging.getLogger(__name__)


class Drain(th.Thread):
    """
    Drain a generator to a destination that answers to put(), in a thread
    """

    def __init__(self, source, destination):
        super(Drain, self).__init__()
        self.source = source
        self.destination = destination
        self._finished = th.Event()
        self._interrupted = th.Event()

    def run(self):
        for item in self.source:
            self.destination.put(item)
            if self._interrupted.is_set():
                break
        self._finished.set()

    def wait(self, timeout=None):
        self._finished.wait(timeout=timeout)

    def close(self):
        self._interrupted.set()


class SecuredShell(object):
    def __init__(self, host, port, username, timeout):
        self.host = host
        self.port = port
        self.username = username
        self.timeout = timeout

    def connect(self):
        logger.debug(
            "Opening SSH connection to {host}:{port}".format(host=self.host,
                                                             port=self.port))
        client = SSHClient()
        client.load_system_host_keys()
        client.set_missing_host_key_policy(AutoAddPolicy())

        try:
            client.connect(self.host,
                           port=self.port,
                           username=self.username,
                           timeout=self.timeout, )
        except ValueError, e:
            logger.error(e)
            logger.warning("""
Patching Crypto.Cipher.AES.new and making another attempt.

See here for the details:
http://uucode.com/blog/2015/02/20/workaround-for-ctr-mode-needs-counter-parameter-not-iv/
            """)
            client.close()
            import Crypto.Cipher.AES
            orig_new = Crypto.Cipher.AES.new

            def fixed_AES_new(key, *ls):
                if Crypto.Cipher.AES.MODE_CTR == ls[0]:
                    ls = list(ls)
                    ls[1] = ''
                return orig_new(key, *ls)

            Crypto.Cipher.AES.new = fixed_AES_new
            client.connect(self.host,
                           port=self.port,
                           username=self.username,
                           timeout=self.timeout, )
        return client

    def execute(self, cmd):
        logger.info("Execute on %s: %s", self.host, cmd)
        with self.connect() as client:
            _, stdout, stderr = client.exec_command(cmd)
            output = stdout.read()
            errors = stderr.read()
            err_code = stdout.channel.recv_exit_status()
        return output, errors, err_code

    def rm(self, path):
        return self.execute("rm -f %s" % path)

    def rm_r(self, path):
        return self.execute("rm -rf %s" % path)

    def mkdir(self, path):
        return self.execute("mkdir -p %s" % path)

    def send_file(self, local_path, remote_path):
        logger.info("Sending [{local}] to {host}:[{remote}]".format(
            local=local_path,
            host=self.host,
            remote=remote_path))
        with self.connect() as client, client.open_sftp() as sftp:
            result = sftp.put(local_path, remote_path)
        return result

    def get_file(self, remote_path, local_path):
        logger.info("Receiving from {host}:[{remote}] to [{local}]".format(
            local=local_path,
            host=self.host,
            remote=remote_path))
        with self.connect() as client, client.open_sftp() as sftp:
            result = sftp.get(remote_path, local_path)
        return result

    def async_session(self, cmd):
        return AsyncSession(self, cmd)


class AsyncSession(object):
    def __init__(self, ssh, cmd):
        self.client = ssh.connect()
        self.session = self.client.get_transport().open_session()
        self.session.get_pty()
        self.session.exec_command(cmd)

    def send(self, data):
        self.session.send(data)

    def close(self):
        self.session.close()
        self.client.close()

    def finished(self):
        return self.session.exit_status_ready()

    def read_maybe(self):
        if self.session.recv_ready():
            return self.session.recv(4096)
        else:
            return None


### HTTP codes
HTTP = httplib.responses

### Extended list of HTTP status codes(WEBdav etc.)
### HTTP://en.wikipedia.org/wiki/List_of_HTTP_status_codes
WEBDAV = {
    102: 'Processing',
    103: 'Checkpoint',
    122: 'Request-URI too long',
    207: 'Multi-Status',
    226: 'IM Used',
    308: 'Resume Incomplete',
    418: 'I\'m a teapot',
    422: 'Unprocessable Entity',
    423: 'Locked',
    424: 'Failed Dependency',
    425: 'Unordered Collection',
    426: 'Upgrade Required',
    444: 'No Response',
    449: 'Retry With',
    450: 'Blocked by Windows Parental Controls',
    499: 'Client Closed Request',
    506: 'Variant Also Negotiates',
    507: 'Insufficient Storage',
    509: 'Bandwidth Limit Exceeded',
    510: 'Not Extended',
    598: 'network read timeout error',
    599: 'network connect timeout error',
    999: 'Common Failure',
{'fraction_non_alphanumeric': 0.11704908510020332, 'fraction_numerical': 0.02207377287249492, 'mean_word_length': 2.182532347504621, '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': '5796763', 'n_tokens_mistral': 2166, 'n_tokens_neox': 1994, 'n_words': 515}