<commit_before>"""Generally useful utilities for AWS web services not specific to a service.

New things in this module should be of relevance to more than one of Amazon's
services.
"""

from base64 import b64encode
from hashlib import sha1, md5, sha256
import hmac
from urlparse import urlparse, urlunparse
import time

# Import XMLTreeBuilder from somewhere; here in one place to prevent
# duplication.
try:
    from xml.etree.ElementTree import XMLTreeBuilder
except ImportError:
    from elementtree.ElementTree import XMLTreeBuilder


__all__ = ["hmac_sha1", "hmac_sha256", "iso8601time", "calculate_md5", "XML"]


def calculate_md5(data):
    digest = md5(data).digest()
    return b64encode(digest)


def hmac_sha1(secret, data):
    digest = hmac.new(secret, data, sha1).digest()
    return b64encode(digest)


def hmac_sha256(secret, data):
    digest = hmac.new(secret, data, sha256).digest()
    return b64encode(digest)


def iso8601time(time_tuple):
    """Format time_tuple as a ISO8601 time string.

    :param time_tuple: Either None, to use the current time, or a tuple tuple.
    """
    if time_tuple:
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time_tuple)
    else:
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


class NamespaceFixXmlTreeBuilder(XMLTreeBuilder):

    def _fixname(self, key):
        if "}" in key:
            key = key.split("}", 1)[1]
        return key


def XML(text):
    parser = NamespaceFixXmlTreeBuilder()
    parser.feed(text)
    return parser.close()


def parse(url, defaultPort=None):
    """
    Split the given URL into the scheme, host, port, and path.

    @type url: C{str}
    @param url: An URL to parse.

    @type defaultPort: C{int} or C{None}
    @param defaultPort: An alternate value to use as the port if the URL does
    not include one.

    @return: A four-tuple of the scheme, host, port, and path of the URL.  All
    of these are C{str} instances except for port, which is an C{int}.
    """
    url = url.strip()
    parsed = urlparse(url)
    scheme = parsed[0]
    path = urlunparse(("", "") + parsed[2:])
    if defaultPort is None:
        if scheme == "https":
            defaultPort = 443
        else:
            defaultPort = 80
    host, port = parsed[1], defaultPort
    if ":" in host:
        host, port = host.split(":")
        try:
            port = int(port)
        except ValueError:
            port = defaultPort
    if path == "":
        path = "/"
    return (str(scheme), str(host), port, str(path))


def get_exitcode_reactor():
    """
    This is only neccesary until a fix like the one outlined here is implemented for Twisted:
        http://twistedmatrix.com/trac/ticket/2182
    """
    from twisted.internet.main import installReactor
    from twisted.internet.selectreactor import SelectReactor

    class ExitCodeReactor(SelectReactor):

        def stop(self, exitStatus=0):
            super(ExitCodeReactor, self).stop()
            self.exitStatus = exitStatus

        def run(self, *args, **kwargs):
            super(ExitCodeReactor, self).run(*args, **kwargs)
            return self.exitStatus

    reactor = ExitCodeReactor()
    installReactor(reactor)
    return reactor


reactor = get_exitcode_reactor()
<commit_msg>Wrapped the reactor variable definition in the util module in a try/except so that trial doesn't break.<commit_after>"""Generally useful utilities for AWS web services not specific to a service.

New things in this module should be of relevance to more than one of Amazon's
services.
"""

from base64 import b64encode
from hashlib import sha1, md5, sha256
import hmac
from urlparse import urlparse, urlunparse
import time

# Import XMLTreeBuilder from somewhere; here in one place to prevent
# duplication.
try:
    from xml.etree.ElementTree import XMLTreeBuilder
except ImportError:
    from elementtree.ElementTree import XMLTreeBuilder


__all__ = ["hmac_sha1", "hmac_sha256", "iso8601time", "calculate_md5", "XML"]


def calculate_md5(data):
    digest = md5(data).digest()
    return b64encode(digest)


def hmac_sha1(secret, data):
    digest = hmac.new(secret, data, sha1).digest()
    return b64encode(digest)


def hmac_sha256(secret, data):
    digest = hmac.new(secret, data, sha256).digest()
    return b64encode(digest)


def iso8601time(time_tuple):
    """Format time_tuple as a ISO8601 time string.

    :param time_tuple: Either None, to use the current time, or a tuple tuple.
    """
    if time_tuple:
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time_tuple)
    else:
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


class NamespaceFixXmlTreeBuilder(XMLTreeBuilder):

    def _fixname(self, key):
        if "}" in key:
            key = key.split("}", 1)[1]
        return key


def XML(text):
    parser = NamespaceFixXmlTreeBuilder()
    parser.feed(text)
    return parser.close()


def parse(url, defaultPort=None):
    """
    Split the given URL into the scheme, host, port, and path.

    @type url: C{str}
    @param url: An URL to parse.

    @type defaultPort: C{int} or C{None}
    @param defaultPort: An alternate value to use as the port if the URL does
    not include one.

    @return: A four-tuple of the scheme, host, port, and path of the URL.  All
    of these are C{str} instances except for port, which is an C{int}.
    """
    url = url.strip()
    parsed = urlparse(url)
    scheme = parsed[0]
    path = urlunparse(("", "") + parsed[2:])
    if defaultPort is None:
        if scheme == "https":
            defaultPort = 443
        else:
            defaultPort = 80
    host, port = parsed[1], defaultPort
    if ":" in host:
        host, port = host.split(":")
        try:
            port = int(port)
        except ValueError:
            port = defaultPort
    if path == "":
        path = "/"
    return (str(scheme), str(host), port, str(path))


def get_exitcode_reactor():
    """
    This is only neccesary until a fix like the one outlined here is
    implemented for Twisted:
        http://twistedmatrix.com/trac/ticket/2182
    """
    from twisted.internet.main import installReactor
    from twisted.internet.selectreactor import SelectReactor

    class ExitCodeReactor(SelectReactor):

        def stop(self, exitStatus=0):
            super(ExitCodeReactor, self).stop()
            self.exitStatus = exitStatus

        def run(self, *args, **kwargs):
            super(ExitCodeReactor, self).run(*args, **kwargs)
            return self.exitStatus

    reactor = ExitCodeReactor()
    installReactor(reactor)
    return reactor


try:
    reactor = get_exitcode_reactor()
except:
    from twisted.internet import reactor

{'fraction_non_alphanumeric': 0.10582959641255606, 'fraction_numerical': 0.017040358744394617, 'mean_word_length': 2.9591715976331363, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 3, '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': '5123726', 'n_tokens_mistral': 2160, 'n_tokens_neox': 2027, 'n_words': 660}