<commit_before>import subprocess, sys
import StringIO
import inspect
from threading import Thread

from doit.exceptions import TaskFailed, TaskError

# Exceptions
class InvalidTask(Exception):
    """Invalid task instance. User error on specifying the task."""
    pass



# Actions
class BaseAction(object):
    """Base class for all actions"""

    # must implement:
    # def execute(self, out=None, err=None)

    pass



class CmdAction(BaseAction):
    """
    Command line action. Spawns a new process.

    @ivar action(str): Command to be passed to the shell subprocess.
         It may contain python mapping strings with the keys: dependencies,
         changed and targets. ie. "zip %(targets)s %(changed)s"
    @ivar task(Task): reference to task that contains this action
    """

    def __init__(self, action):
        assert isinstance(action,str), "CmdAction must be a string."
        self.action = action
        self.task = None
        self.out = None
        self.err = None
        self.result = None
        self.values = {}

    def execute(self, out=None, err=None):
        """
        Execute command action

        both stdout and stderr from the command are captured and saved
        on self.out/err. Real time output is controlled by parameters
        @param out: None - no real time output
                    a file like object (has write method)
        @param err: idem
        @return failure:
            - None: if successful
            - TaskError: If subprocess return code is greater than 125
            - TaskFailed: If subprocess return code isn't zero (and
        not greater than 125)
        """
        action = self.expand_action()

        # spawn task process
        process = subprocess.Popen(action, shell=True,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        def print_process_output(process, input, capture, realtime):
            while True:
                # line buffered
                line = input.readline()
                # unbuffered ? process.stdout.read(1)
                if line:
                    capture.write(line)
                    if realtime:
                        realtime.write(line)
                if not line and process.poll() != None:
                    break

        output = StringIO.StringIO()
        errput = StringIO.StringIO()
        t_out = Thread(target=print_process_output,
                       args=(process, process.stdout, output, out))
        t_err = Thread(target=print_process_output,
                       args=(process, process.stderr, errput, err))
        t_out.start()
        t_err.start()
        t_out.join()
        t_err.join()

        self.out = output.getvalue()
        self.err = errput.getvalue()
        self.result = self.out + self.err

        # task error - based on:
        # http://www.gnu.org/software/bash/manual/bashref.html#Exit-Status
        # it doesnt make so much difference to return as Error or Failed anyway
        if process.returncode > 125:
            return TaskError("Command error: '%s' returned %s" %
                            (action,process.returncode))

        # task failure
        if process.returncode != 0:
            return TaskFailed("Command failed: '%s' returned %s" %
                             (action,process.returncode))


    def expand_action(self):
        """expand action string using task meta informations
        @returns (string) - expanded string after substitution
        """
        if not self.task:
            return self.action

        subs_dict = {'targets' : " ".join(self.task.targets),
                     'dependencies': " ".join(self.task.file_dep)}
        # just included changed if it is set
        if self.task.dep_changed is not None:
            subs_dict['changed'] = " ".join(self.task.dep_changed)
        # task option parameters
        subs_dict.update(self.task.options)
        return self.action % subs_dict


    def __str__(self):
        return "Cmd: %s" % self.expand_action()

    def __repr__(self):
        return "<CmdAction: '%s'>"  % self.expand_action()




class Writer(object):
    """write to many streams"""
    def __init__(self, *writers) :
        self.writers = writers

    def write(self, text) :
        for w in self.writers :
                w.write(text)


class PythonAction(BaseAction):
    """Python action. Execute a python callable.<commit_msg>Python-action output streams support 'flush' method.<commit_after>import subprocess, sys
import StringIO
import inspect
from threading import Thread

from doit.exceptions import TaskFailed, TaskError

# Exceptions
class InvalidTask(Exception):
    """Invalid task instance. User error on specifying the task."""
    pass



# Actions
class BaseAction(object):
    """Base class for all actions"""

    # must implement:
    # def execute(self, out=None, err=None)

    pass



class CmdAction(BaseAction):
    """
    Command line action. Spawns a new process.

    @ivar action(str): Command to be passed to the shell subprocess.
         It may contain python mapping strings with the keys: dependencies,
         changed and targets. ie. "zip %(targets)s %(changed)s"
    @ivar task(Task): reference to task that contains this action
    """

    def __init__(self, action):
        assert isinstance(action,str), "CmdAction must be a string."
        self.action = action
        self.task = None
        self.out = None
        self.err = None
        self.result = None
        self.values = {}

    def execute(self, out=None, err=None):
        """
        Execute command action

        both stdout and stderr from the command are captured and saved
        on self.out/err. Real time output is controlled by parameters
        @param out: None - no real time output
                    a file like object (has write method)
        @param err: idem
        @return failure:
            - None: if successful
            - TaskError: If subprocess return code is greater than 125
            - TaskFailed: If subprocess return code isn't zero (and
        not greater than 125)
        """
        action = self.expand_action()

        # spawn task process
        process = subprocess.Popen(action, shell=True,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        def print_process_output(process, input, capture, realtime):
            while True:
                # line buffered
                line = input.readline()
                # unbuffered ? process.stdout.read(1)
                if line:
                    capture.write(line)
                    if realtime:
                        realtime.write(line)
                if not line and process.poll() != None:
                    break

        output = StringIO.StringIO()
        errput = StringIO.StringIO()
        t_out = Thread(target=print_process_output,
                       args=(process, process.stdout, output, out))
        t_err = Thread(target=print_process_output,
                       args=(process, process.stderr, errput, err))
        t_out.start()
        t_err.start()
        t_out.join()
        t_err.join()

        self.out = output.getvalue()
        self.err = errput.getvalue()
        self.result = self.out + self.err

        # task error - based on:
        # http://www.gnu.org/software/bash/manual/bashref.html#Exit-Status
        # it doesnt make so much difference to return as Error or Failed anyway
        if process.returncode > 125:
            return TaskError("Command error: '%s' returned %s" %
                            (action,process.returncode))

        # task failure
        if process.returncode != 0:
            return TaskFailed("Command failed: '%s' returned %s" %
                             (action,process.returncode))


    def expand_action(self):
        """expand action string using task meta informations
        @returns (string) - expanded string after substitution
        """
        if not self.task:
            return self.action

        subs_dict = {'targets' : " ".join(self.task.targets),
                     'dependencies': " ".join(self.task.file_dep)}
        # just included changed if it is set
        if self.task.dep_changed is not None:
            subs_dict['changed'] = " ".join(self.task.dep_changed)
        # task option parameters
        subs_dict.update(self.task.options)
        return self.action % subs_dict


    def __str__(self):
        return "Cmd: %s" % self.expand_action()

    def __repr__(self):
        return "<CmdAction: '%s'>"  % self.expand_action()




class Writer(object):
    """write to many streams"""
    def __init__(self, *writers) :
        self.writers = writers

    def write(self, text) :
        for w in self.writers :
                w.write(text)

    def flush(self) :
        for w in self.writers :
                w.flush()


class PythonAction(BaseAction):
    """Python action. Execute a python callable.
{'fraction_non_alphanumeric': 0.09290953545232274, 'fraction_numerical': 0.0024449877750611247, 'mean_word_length': 1.944698952879581, 'pattern_counts': {'":': 0, '<': 5, '<?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': '6386027', 'n_tokens_mistral': 2470, 'n_tokens_neox': 2327, 'n_words': 747}