<commit_before># This file was originally part of Paste, but <NAME> pulled it out as a
# separate file in 2011 and made some modifications. The original license text
# was:
#     (c) 2005 <NAME> and contributors; written for Paste (http://pythonpaste.org)
#     Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php

"""
Application that runs a CGI script.
"""
import os
import sys
import subprocess
import urllib
try:
    import select
except ImportError:
    select = None

__all__ = ['CGIError', 'CGIApplication']

class CGIError(Exception):
    """
    Raised when the CGI script can't be found or doesn't
    act like a proper CGI script.
    """

class CGIApplication(object):

    def __init__(self,
                 script,
                 include_os_environ=True,
                 query_string=None):
        if '?' in script:
            assert query_string is None, (
                "You cannot have '?' in your script name (%r) and also "
                "give a query_string (%r)" % (script, query_string))
            script, query_string = script.split('?', 1)
        assert os.access(script, os.X_OK)
        self.script = os.path.abspath(script)
        self.include_os_environ = include_os_environ
        self.query_string = query_string

    def __call__(self, environ, start_response):
        if 'REQUEST_URI' not in environ:
            environ['REQUEST_URI'] = (
                urllib.quote(environ.get('SCRIPT_NAME', ''))
                + urllib.quote(environ.get('PATH_INFO', '')))
        if self.include_os_environ:
            cgi_environ = os.environ.copy()
        else:
            cgi_environ = {}
        for name in environ:
            # Should unicode values be encoded?
            if (name.upper() == name
                and isinstance(environ[name], str)):
                cgi_environ[name] = environ[name]
        if self.query_string is not None:
            old = cgi_environ.get('QUERY_STRING', '')
            if old:
                old += '&'
            cgi_environ['QUERY_STRING'] = old + self.query_string
        cgi_environ['SCRIPT_FILENAME'] = self.script
        proc = subprocess.Popen(
            [self.script],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=cgi_environ,
            cwd=os.path.dirname(self.script),
            )
        writer = CGIWriter(environ, start_response)
        if select and sys.platform != 'win32':
            for block in proc_communicate_yielding_stdout(
                    proc,
                    stdin=StdinReader.from_environ(environ),
                    stderr=environ['wsgi.errors']):
                for block2 in writer.write(block):
                    yield block2
        else:
            stdout, stderr = proc.communicate(StdinReader.from_environ(environ).read())
            if stderr:
                environ['wsgi.errors'].write(stderr)
            for block in writer.write(stdout):
                yield block
        yield ""
        if not writer.headers_finished:
            start_response(writer.status, writer.headers)

class CGIWriter(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start_response = start_response
        self.status = '200 OK'
        self.headers = []
        self.headers_finished = False
        self.buffer = ''

    def write(self, data):
        if self.headers_finished:
            return [data]
        self.buffer += data
        while '\n' in self.buffer:
            if '\r\n' in self.buffer and self.buffer.find('\r\n') < self.buffer.find('\n'):
                line1, self.buffer = self.buffer.split('\r\n', 1)
            else:
                line1, self.buffer = self.buffer.split('\n', 1)
            if not line1:
                self.headers_finished = True
                self.start_response(
                    self.status, self.headers)
                buffer = self.buffer
                del self.buffer
                del self.headers
                del self.status
                return [buffer]
            elif ':' not in line1:
                raise CGIError(
                    "Bad header line: %r" % line1)
            else:
                name, value = line1.split(':', 1)
                value = value.lstrip()
                name = name.strip()
                if name.lower() == 'status':
                    if ' ' not in value:
                        # WSGI requires this space, sometimes CGI scripts don't set it:
                        value = '%s General' % value
                    self.status = value
                else:
                    self.headers.append((name, value))
        return []

class StdinReader(object):

    def __init__(self, stdin, content_length):
        self.stdin = stdin
        self.content_length = content_length

    def from_environ(cls, environ):
        length = environ.get('CONTENT_LENGTH')
        if length:
            length = int(length)
        else:
            length = 0
        return cls(environ['wsgi.input'], length)

    from_environ = classmethod(from_environ)

    def read(self, size=None):
        if not self.content_length:
            return ''
        if size is None:
            text = self.stdin.read(self.content_length)
        else:
            text = self.stdin.read(min(self.content_length, size))
        self.content_length -= len(text)
        return text

def proc_communicate_yielding_stdout(proc, stdin=None, stderr=None):
    """
    Run the given process, piping input and errors to/from the given
    file-like objects (which need not be actual file objects, unlike
    the arguments passed to Popen). Yield data written to stdout as a
    series of strings. Wait for process to terminate.

    Note: this is taken from the posix version of
    subprocess.Popen.communicate, but made more general through the
    use of file-like objects.<commit_msg>Make cgi_as_wsgi.py use subprocess32, not subprocess.<commit_after># This file was originally part of Paste, but <NAME> pulled it out as a
# separate file in 2011 and made some modifications. The original license text
# was:
#     (c) 2005 <NAME> and contributors; written for Paste (http://pythonpaste.org)
#     Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php

"""
Application that runs a CGI script.
"""
import os
import sys
import subprocess32
import urllib
try:
    import select
except ImportError:
    select = None

__all__ = ['CGIError', 'CGIApplication']

class CGIError(Exception):
    """
    Raised when the CGI script can't be found or doesn't
    act like a proper CGI script.
    """

class CGIApplication(object):

    def __init__(self,
                 script,
                 include_os_environ=True,
                 query_string=None):
        if '?' in script:
            assert query_string is None, (
                "You cannot have '?' in your script name (%r) and also "
                "give a query_string (%r)" % (script, query_string))
            script, query_string = script.split('?', 1)
        assert os.access(script, os.X_OK)
        self.script = os.path.abspath(script)
        self.include_os_environ = include_os_environ
        self.query_string = query_string

    def __call__(self, environ, start_response):
        if 'REQUEST_URI' not in environ:
            environ['REQUEST_URI'] = (
                urllib.quote(environ.get('SCRIPT_NAME', ''))
                + urllib.quote(environ.get('PATH_INFO', '')))
        if self.include_os_environ:
            cgi_environ = os.environ.copy()
        else:
            cgi_environ = {}
        for name in environ:
            # Should unicode values be encoded?
            if (name.upper() == name
                and isinstance(environ[name], str)):
                cgi_environ[name] = environ[name]
        if self.query_string is not None:
            old = cgi_environ.get('QUERY_STRING', '')
            if old:
                old += '&'
            cgi_environ['QUERY_STRING'] = old + self.query_string
        cgi_environ['SCRIPT_FILENAME'] = self.script
        proc = subprocess32.Popen(
            [self.script],
            stdin=subprocess32.PIPE,
            stdout=subprocess32.PIPE,
            stderr=subprocess32.PIPE,
            env=cgi_environ,
            cwd=os.path.dirname(self.script),
            )
        writer = CGIWriter(environ, start_response)
        if select and sys.platform != 'win32':
            for block in proc_communicate_yielding_stdout(
                    proc,
                    stdin=StdinReader.from_environ(environ),
                    stderr=environ['wsgi.errors']):
                for block2 in writer.write(block):
                    yield block2
        else:
            stdout, stderr = proc.communicate(StdinReader.from_environ(environ).read())
            if stderr:
                environ['wsgi.errors'].write(stderr)
            for block in writer.write(stdout):
                yield block
        yield ""
        if not writer.headers_finished:
            start_response(writer.status, writer.headers)

class CGIWriter(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start_response = start_response
        self.status = '200 OK'
        self.headers = []
        self.headers_finished = False
        self.buffer = ''

    def write(self, data):
        if self.headers_finished:
            return [data]
        self.buffer += data
        while '\n' in self.buffer:
            if '\r\n' in self.buffer and self.buffer.find('\r\n') < self.buffer.find('\n'):
                line1, self.buffer = self.buffer.split('\r\n', 1)
            else:
                line1, self.buffer = self.buffer.split('\n', 1)
            if not line1:
                self.headers_finished = True
                self.start_response(
                    self.status, self.headers)
                buffer = self.buffer
                del self.buffer
                del self.headers
                del self.status
                return [buffer]
            elif ':' not in line1:
                raise CGIError(
                    "Bad header line: %r" % line1)
            else:
                name, value = line1.split(':', 1)
                value = value.lstrip()
                name = name.strip()
                if name.lower() == 'status':
                    if ' ' not in value:
                        # WSGI requires this space, sometimes CGI scripts don't set it:
                        value = '%s General' % value
                    self.status = value
                else:
                    self.headers.append((name, value))
        return []

class StdinReader(object):

    def __init__(self, stdin, content_length):
        self.stdin = stdin
        self.content_length = content_length

    def from_environ(cls, environ):
        length = environ.get('CONTENT_LENGTH')
        if length:
            length = int(length)
        else:
            length = 0
        return cls(environ['wsgi.input'], length)

    from_environ = classmethod(from_environ)

    def read(self, size=None):
        if not self.content_length:
            return ''
        if size is None:
            text = self.stdin.read(self.content_length)
        else:
            text = self.stdin.read(min(self.content_length, size))
        self.content_length -= len(text)
        return text

def proc_communicate_yielding_stdout(proc, stdin=None, stderr=None):
    """
    Run the given process, piping input and errors to/from the given
    file-like objects (which need not be actual file objects, unlike
    the arguments passed to Popen). Yield data written to stdout as a
    series of strings. Wait for process to terminate.

    Note: this is taken from the posix version of
    subprocess32.Popen.communicate, but made more general through the
    use of file-like objects.
{'fraction_non_alphanumeric': 0.09539254823685961, 'fraction_numerical': 0.0054890219560878245, 'mean_word_length': 1.8017241379310345, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '950490', 'n_tokens_mistral': 3389, 'n_tokens_neox': 3286, 'n_words': 906}