<commit_before># Copyright 2014, <NAME> (see LICENSE.txt)
# Portions Copyright Python Software Foundation
# (see COPYRIGHT.txt and PYTHON-LICENSE.txt)

# main part
import code
import contextlib
import doctest
import getpass
import linecache
import pdb
import re
import sys
import traceback
import warnings

import six

from yalpt import ansi_helper as ansi
from yalpt import formatters
from yalpt import parsers


__all__ = ["LiterateInterpreter"]


@contextlib.contextmanager
def noop_mgr(writer):
    yield writer


class LiterateInterpreter(code.InteractiveConsole):
    def __init__(self, text_formatter=formatters.NoopFormatter(),
                 code_parser=parsers.DocTestParser(), use_ansi=True,
                 use_readline=True, env_driver=None, *args, **kwargs):
        code.InteractiveConsole.__init__(self, *args, **kwargs)

        self._output_checker = doctest.OutputChecker()
        self._fakeout = doctest._SpoofOut()
        self.chunks = None
        self.exc_msg = None
        self.name = 'literate program'
        self.text_formatter = text_formatter
        self.code_parser = code_parser
        self.use_ansi = use_ansi
        self.pause = True
        self.interactive = True

        if use_readline:
            self._add_readline()

        self._env_driver = env_driver

        self._correct_path()

    def runfunction(self, func):
        self.runcode(six.get_function_code(func))

    def _correct_path(self):
        def correct_path():
            import sys
            if '' not in sys.path:
                sys.path.insert(0, '')
            del sys

        self.runfunction(correct_path)

    def _add_readline(self):
        self.locals['__console_locals__'] = self.locals

        def add_readline():
            # add support for completion
            import atexit
            import os
            import readline
            import rlcompleter

            completer = rlcompleter.Completer(__console_locals__)  # noqa
            readline.set_completer(completer.complete)

            # tab completion
            readline.parse_and_bind('Control-space: complete')
            # history file
            histfile = os.path.join(os.environ['HOME'],
                                    '.literate-python-history')
            try:
                readline.read_history_file(histfile)
            except IOError:
                pass
            atexit.register(readline.write_history_file, histfile)
            del os, histfile, readline, rlcompleter

        self.runfunction(add_readline)

        del self.locals['__console_locals__']

    def _interact_once(self, more):
        try:
            if more:
                prompt = sys.ps2
            else:
                prompt = sys.ps1
            try:
                line = self.raw_input(prompt)
                # Can be None if sys.stdin was redefined
                encoding = getattr(sys.stdin, "encoding", None)
                if encoding and not isinstance(line, six.text_type):
                    line = line.decode(encoding)
            except EOFError:
                self.write("\n")
                return (False, None)
            else:
                if len(line) == 0:
                    blank = True
                else:
                    blank = False
                more = self.push(line)
        except KeyboardInterrupt:
            self.write("\nKeyboardInterrupt\n")
            self.resetbuffer()
            more = False
            blank = False

        return (blank, more)

    # BEGIN FROM PYTHON STD LIB (doctest)
    __LINECACHE_FILENAME_RE = re.compile(r'<literate '
                                         r'(?P<name>.+)'
                                         r'\[(?P<chunknum>\d+)\]>$')

    def _patched_linecache_getlines(self, filename, module_globals=None):
        m = self.__LINECACHE_FILENAME_RE.match(filename)
        if m and m.group('name') == self.name:
            chunk = self.chunks[int(m.group('chunknum'))]
            source = chunk.source
            if isinstance(source, six.text_type):
                source = source.encode('ascii', 'backslashreplace')
            return source.splitlines(True)
        else:
            return self.save_linecache_getlines(filename, module_globals)

    @contextlib.contextmanager
    def _capture_output(self):
        save_stdout = sys.stdout
        sys.stdout = self._fakeout

        # set up pdb to work properly
        save_set_trace = pdb.set_trace
        self.debugger = doctest._OutputRedirectingPdb(save_stdout)
        self.debugger.reset()
        pdb.set_trace = self.debugger.set_trace
        self.save_linecache_getlines = linecache.getlines
        linecache.getlines = self._patched_linecache_getlines
        save_displayhook = sys.displayhook
        sys.displayhook = sys.__displayhook__

        try:
            yield sys.stdout
        finally:
            sys.stdout = save_stdout
            pdb.set_trace = save_set_trace
            linecache.getlines = self.save_linecache_getlines
            sys.displayhook = save_displayhook
    # END FROM PYTHON STD LIB

    def _process_code_line(self, line, res, more):
        if more:
            self.write(sys.ps2)
        else:
            self.write(sys.ps1)

        if self.use_ansi:
            self.write(ansi.with_codes(line, 38, 5, 0xdf))
        else:
            self.write(line)

        self.write("\n")
        with self._capture_output() as output:
            more = self.push(line)

            res += output.getvalue()
            output.truncate(0)

        exc = self.exc_msg
        self.exc_msg = None
        return (res, exc, more)

    def _format_tb(self):
        try:
            extype, value, tb = sys.exc_info()
            sys.last_type = extype
            sys.last_value = value
            sys.last_traceback = tb
            res = traceback.format_exception_only(extype, value)
        finally:
            tb = None<commit_msg>Add lines from files to readline history<commit_after># Copyright 2014, <NAME> (see LICENSE.txt)
# Portions Copyright Python Software Foundation
# (see COPYRIGHT.txt and PYTHON-LICENSE.txt)

# main part
import code
import contextlib
import doctest
import getpass
import linecache
import pdb
import re
import sys
import traceback
import warnings

import six

from yalpt import ansi_helper as ansi
from yalpt import formatters
from yalpt import parsers


__all__ = ["LiterateInterpreter"]


@contextlib.contextmanager
def noop_mgr(writer):
    yield writer


class LiterateInterpreter(code.InteractiveConsole):
    def __init__(self, text_formatter=formatters.NoopFormatter(),
                 code_parser=parsers.DocTestParser(), use_ansi=True,
                 use_readline=True, env_driver=None, *args, **kwargs):
        code.InteractiveConsole.__init__(self, *args, **kwargs)

        self._output_checker = doctest.OutputChecker()
        self._fakeout = doctest._SpoofOut()
        self.chunks = None
        self.exc_msg = None
        self.name = 'literate program'
        self.text_formatter = text_formatter
        self.code_parser = code_parser
        self.use_ansi = use_ansi
        self.pause = True
        self.interactive = True

        if use_readline:
            self._readline = __import__('readline')
            self._add_readline()
        else:
            self._readline = None

        self._env_driver = env_driver

        self._correct_path()

    def runfunction(self, func):
        self.runcode(six.get_function_code(func))

    def _correct_path(self):
        def correct_path():
            import sys
            if '' not in sys.path:
                sys.path.insert(0, '')
            del sys

        self.runfunction(correct_path)

    def _add_readline(self):
        self.locals['__console_locals__'] = self.locals
        self.locals['readline'] = self._readline

        def add_readline():
            # add support for completion
            import atexit
            import os
            import readline
            import rlcompleter

            completer = rlcompleter.Completer(__console_locals__)  # noqa
            readline.set_completer(completer.complete)

            # tab completion
            readline.parse_and_bind('Control-space: complete')
            # history file
            histfile = os.path.join(os.environ['HOME'],
                                    '.literate-python-history')
            try:
                readline.read_history_file(histfile)
            except IOError:
                pass
            atexit.register(readline.write_history_file, histfile)
            del os, histfile, readline, rlcompleter

        self.runfunction(add_readline)

        del self.locals['__console_locals__']
        del self.locals['readline']

    def _interact_once(self, more):
        try:
            if more:
                prompt = sys.ps2
            else:
                prompt = sys.ps1
            try:
                line = self.raw_input(prompt)
                # Can be None if sys.stdin was redefined
                encoding = getattr(sys.stdin, "encoding", None)
                if encoding and not isinstance(line, six.text_type):
                    line = line.decode(encoding)
            except EOFError:
                self.write("\n")
                return (False, None)
            else:
                if len(line) == 0:
                    blank = True
                else:
                    blank = False
                more = self.push(line)
        except KeyboardInterrupt:
            self.write("\nKeyboardInterrupt\n")
            self.resetbuffer()
            more = False
            blank = False

        return (blank, more)

    # BEGIN FROM PYTHON STD LIB (doctest)
    __LINECACHE_FILENAME_RE = re.compile(r'<literate '
                                         r'(?P<name>.+)'
                                         r'\[(?P<chunknum>\d+)\]>$')

    def _patched_linecache_getlines(self, filename, module_globals=None):
        m = self.__LINECACHE_FILENAME_RE.match(filename)
        if m and m.group('name') == self.name:
            chunk = self.chunks[int(m.group('chunknum'))]
            source = chunk.source
            if isinstance(source, six.text_type):
                source = source.encode('ascii', 'backslashreplace')
            return source.splitlines(True)
        else:
            return self.save_linecache_getlines(filename, module_globals)

    @contextlib.contextmanager
    def _capture_output(self):
        save_stdout = sys.stdout
        sys.stdout = self._fakeout

        # set up pdb to work properly
        save_set_trace = pdb.set_trace
        self.debugger = doctest._OutputRedirectingPdb(save_stdout)
        self.debugger.reset()
        pdb.set_trace = self.debugger.set_trace
        self.save_linecache_getlines = linecache.getlines
        linecache.getlines = self._patched_linecache_getlines
        save_displayhook = sys.displayhook
        sys.displayhook = sys.__displayhook__

        try:
            yield sys.stdout
        finally:
            sys.stdout = save_stdout
            pdb.set_trace = save_set_trace
            linecache.getlines = self.save_linecache_getlines
            sys.displayhook = save_displayhook
    # END FROM PYTHON STD LIB

    def _process_code_line(self, line, res, more):
        if self._readline is not None:
            self._readline.add_history(line)

        if more:
            self.write(sys.ps2)
        else:
            self.write(sys.ps1)

        if self.use_ansi:
            self.write(ansi.with_codes(line, 38, 5, 0xdf))
        else:
            self.write(line)

        self.write("\n")
        with self._capture_output() as output:
            more = self.push(line)

            res += output.getvalue()
            output.truncate(0)

        exc = self.exc_msg
        self.exc_msg = None
        return (res, exc, more)

    def _format_tb(self):
        try:
            extype, value, tb = sys.exc_info()
            sys.last_type = extype
            sys.last_value = value
            sys.last_traceback = tb
            res = traceback.format_exception_only(extype, value)
        finally:
            tb = None
{'fraction_non_alphanumeric': 0.08373577801424245, 'fraction_numerical': 0.0024555946631742652, 'mean_word_length': 1.8139106402579457, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5849441', 'n_tokens_mistral': 3551, 'n_tokens_neox': 3413, 'n_words': 746}