<commit_before>import os
import networkx as nx

from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.stage import Stage, Output, Dependency
from dvc.config import Config
from dvc.state import State
from dvc.lock import Lock
from dvc.scm import SCM
from dvc.cache import Cache
from dvc.data_cloud import DataCloud


class PipelineError(DvcException):
    pass


class StageNotInPipelineError(PipelineError):
    pass


class StageNotFoundError(DvcException):
    def __init__(self, path):
        msg = 'Stage file {} does not exist'.format(path)
        super(StageNotFoundError, self).__init__(msg)


class ReproductionError(DvcException):
    def __init__(self, dvc_file_name, msg):
        super(ReproductionError, self).__init__(u'Failed to reproduce \'{}\': {}'.format(dvc_file_name, msg))


class Pipeline(object):

    def __init__(self, project, G):
        self.project = project
        self.G = G

    def graph(self):
        return self.G

    def stages(self):
        return nx.get_node_attributes(self.G, 'stage')

    def changed(self, stage):
        for node in nx.dfs_postorder_nodes(G, stage.path.relative_to(self.project.root_dir)):
            if self.stages[node].changed():
                return True
        return False

    def reproduce(self, stage):
        if stage not in self.stages():
            raise StageNotInPipelineError()

        if not self.changed(stage):
            raise PipelineNotChangedError()

        for node in nx.dfs_postorder_nodes(G, stage.path.relative_to(self.project.root_dir)):
            self.stages[node].reproduce()

        stage.reproduce()


class Project(object):
    DVC_DIR = '.dvc'

    def __init__(self, root_dir):
        self.root_dir = os.path.abspath(os.path.realpath(root_dir))
        self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR)

        self.scm = SCM(self.root_dir)
        self.lock = Lock(self.dvc_dir)
        self.cache = Cache(self.dvc_dir)
        self.state = State(self.root_dir, self.dvc_dir)
        self.config = Config(self.dvc_dir)
        self.logger = Logger(self.config._config)
        self.cloud = DataCloud(self.config._config)

    @staticmethod
    def init(root_dir):
        """
        Initiate dvc project in directory.

        Args:
            root_dir: Path to project's root directory.

        Returns:
            Project instance.

        Raises:
            KeyError: Raises an exception.
        """
        root_dir = os.path.abspath(root_dir)
        dvc_dir = os.path.join(root_dir, Project.DVC_DIR)
        os.mkdir(dvc_dir)

        config = Config.init(dvc_dir)
        cache = Cache.init(dvc_dir)
        state = State.init(root_dir, dvc_dir)
        lock = Lock(dvc_dir)

        scm = SCM(root_dir)
        scm.ignore_list([cache.cache_dir,
                         state.state_file,
                         lock.lock_file])

        ignore_file = os.path.join(dvc_dir, scm.ignore_file())
        scm.add([config.config_file, ignore_file])
        scm.commit('DVC init')

        return Project(root_dir)

    def to_dvc_path(self, path):
        return os.path.relpath(path, self.root_dir)

    def add(self, fname):
        path = os.path.abspath(fname) + Stage.STAGE_FILE_SUFFIX
        cwd = os.path.dirname(path)
        outputs = [Output.loads(self, os.path.basename(fname), use_cache=True, cwd=cwd)]
        stage = Stage(project=self,
                      path=path,
                      cmd=None,
                      cwd=cwd,
                      outs=outputs,
                      deps=[],
                      locked=True)
        stage.save()
        stage.dump()
        return stage

    def remove(self, fname):
        stages = []
        output = Output.loads(self, fname)
        for out in self.outs():
            if out.path == output.path:
                stage = out.stage()
                stages.append(stage)

        if len(stages) == 0:
            raise StageNotFoundError(fname) 

        for stage in stages:
            stage.remove()

        return stages

    def run(self,
            cmd=None,
            deps=[],
            outs=[],
            outs_no_cache=[],
            locked=False,
            fname=Stage.STAGE_FILE,
            cwd=os.curdir,
            no_exec=False):
        cwd = os.path.abspath(cwd)
        path = os.path.join(cwd, fname)
        outputs = Output.loads_from(self, outs, use_cache=True, cwd=cwd)
        outputs += Output.loads_from(self, outs_no_cache, use_cache=False, cwd=cwd)
        deps = Dependency.loads_from(self, deps, cwd=cwd)

        stage = Stage(project=self,
                      path=path,
                      cmd=cmd,
                      cwd=cwd,
                      outs=outputs,
                      deps=deps,
                      locked=locked)
        if not no_exec:
            stage.run()
        stage.dump()
        return stage

    def reproduce(self, target, recursive=True, force=False):
        reproduced = []
        stages = nx.get_node_attributes(self.graph(), 'stage')
        node = os.path.relpath(os.path.abspath(target), self.root_dir)
        if node not in stages:
            raise StageNotFoundError(target)

        if recursive:
            for n in nx.dfs_postorder_nodes(self.graph(), node):
                try:
                    stages[n].reproduce(force=force)
                    stages[n].dump()
                    reproduced.append(stages[n])
                except Exception as ex:
                    raise ReproductionError(n, ex.message)

        stages[node].reproduce(force=force)
        stages[node].dump()
        reproduced.append(stages[node])

        return reproduced

    def checkout(self):
        for stage in self.stages():
            stage.checkout()

    def _used_cache(self):
        clist = []
        for stage in self.stages():
            for out in stage.outs:
                if not out.use_cache:
                    continue
                if out.cache not in clist:
                    clist.append(out.cache)
        return clist

    def gc(self):
        clist = self._used_cache()
        for cache in self.cache.all():
            if cache in clist:
                continue
            os.unlink(cache)
            self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache)))

    def push(self, jobs=1):
        self.cloud.push(self._used_cache(), jobs)

    def pull(self, jobs=1):
        self.cloud.pull(self._used_cache(), jobs)
        for stage in self.stages():
            for out in stage.outs:
                if out.use_cache:<commit_msg>Refactoring: extract method _reproduce_stages() and ReproductionError class changes<commit_after>import os
import networkx as nx

from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.stage import Stage, Output, Dependency
from dvc.config import Config
from dvc.state import State
from dvc.lock import Lock
from dvc.scm import SCM
from dvc.cache import Cache
from dvc.data_cloud import DataCloud


class PipelineError(DvcException):
    pass


class StageNotInPipelineError(PipelineError):
    pass


class StageNotFoundError(DvcException):
    def __init__(self, path):
        msg = 'Stage file {} does not exist'.format(path)
        super(StageNotFoundError, self).__init__(msg)


class ReproductionError(DvcException):
    def __init__(self, dvc_file_name, msg):
        self.dvc_file_name = dvc_file_name
        self.msg = msg
        super(ReproductionError, self).__init__(u'Failed to reproduce \'{}\': {}'.format(dvc_file_name, msg))


class Pipeline(object):

    def __init__(self, project, G):
        self.project = project
        self.G = G

    def graph(self):
        return self.G

    def stages(self):
        return nx.get_node_attributes(self.G, 'stage')

    def changed(self, stage):
        for node in nx.dfs_postorder_nodes(G, stage.path.relative_to(self.project.root_dir)):
            if self.stages[node].changed():
                return True
        return False

    def reproduce(self, stage):
        if stage not in self.stages():
            raise StageNotInPipelineError()

        if not self.changed(stage):
            raise PipelineNotChangedError()

        for node in nx.dfs_postorder_nodes(G, stage.path.relative_to(self.project.root_dir)):
            self.stages[node].reproduce()

        stage.reproduce()


class Project(object):
    DVC_DIR = '.dvc'

    def __init__(self, root_dir):
        self.root_dir = os.path.abspath(os.path.realpath(root_dir))
        self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR)

        self.scm = SCM(self.root_dir)
        self.lock = Lock(self.dvc_dir)
        self.cache = Cache(self.dvc_dir)
        self.state = State(self.root_dir, self.dvc_dir)
        self.config = Config(self.dvc_dir)
        self.logger = Logger(self.config._config)
        self.cloud = DataCloud(self.config._config)

    @staticmethod
    def init(root_dir):
        """
        Initiate dvc project in directory.

        Args:
            root_dir: Path to project's root directory.

        Returns:
            Project instance.

        Raises:
            KeyError: Raises an exception.
        """
        root_dir = os.path.abspath(root_dir)
        dvc_dir = os.path.join(root_dir, Project.DVC_DIR)
        os.mkdir(dvc_dir)

        config = Config.init(dvc_dir)
        cache = Cache.init(dvc_dir)
        state = State.init(root_dir, dvc_dir)
        lock = Lock(dvc_dir)

        scm = SCM(root_dir)
        scm.ignore_list([cache.cache_dir,
                         state.state_file,
                         lock.lock_file])

        ignore_file = os.path.join(dvc_dir, scm.ignore_file())
        scm.add([config.config_file, ignore_file])
        scm.commit('DVC init')

        return Project(root_dir)

    def to_dvc_path(self, path):
        return os.path.relpath(path, self.root_dir)

    def add(self, fname):
        path = os.path.abspath(fname) + Stage.STAGE_FILE_SUFFIX
        cwd = os.path.dirname(path)
        outputs = [Output.loads(self, os.path.basename(fname), use_cache=True, cwd=cwd)]
        stage = Stage(project=self,
                      path=path,
                      cmd=None,
                      cwd=cwd,
                      outs=outputs,
                      deps=[],
                      locked=True)
        stage.save()
        stage.dump()
        return stage

    def remove(self, fname):
        stages = []
        output = Output.loads(self, fname)
        for out in self.outs():
            if out.path == output.path:
                stage = out.stage()
                stages.append(stage)

        if len(stages) == 0:
            raise StageNotFoundError(fname) 

        for stage in stages:
            stage.remove()

        return stages

    def run(self,
            cmd=None,
            deps=[],
            outs=[],
            outs_no_cache=[],
            locked=False,
            fname=Stage.STAGE_FILE,
            cwd=os.curdir,
            no_exec=False):
        cwd = os.path.abspath(cwd)
        path = os.path.join(cwd, fname)
        outputs = Output.loads_from(self, outs, use_cache=True, cwd=cwd)
        outputs += Output.loads_from(self, outs_no_cache, use_cache=False, cwd=cwd)
        deps = Dependency.loads_from(self, deps, cwd=cwd)

        stage = Stage(project=self,
                      path=path,
                      cmd=cmd,
                      cwd=cwd,
                      outs=outputs,
                      deps=deps,
                      locked=locked)
        if not no_exec:
            stage.run()
        stage.dump()
        return stage

    def reproduce(self, target, recursive=True, force=False):
        stages = nx.get_node_attributes(self.graph(), 'stage')
        node = os.path.relpath(os.path.abspath(target), self.root_dir)
        if node not in stages:
            raise StageNotFoundError(target)

        reproduced = self._reproduce_stages(force, node, stages) if recursive else []

        stages[node].reproduce(force=force)
        stages[node].dump()
        reproduced.append(stages[node])

        return reproduced

    def _reproduce_stages(self, force, node, stages):
        result = []
        for n in nx.dfs_postorder_nodes(self.graph(), node):
            try:
                stages[n].reproduce(force=force)
                stages[n].dump()
                result.append(stages[n])
            except Exception as ex:
                raise ReproductionError(n, ex.message)
        return result

    def checkout(self):
        for stage in self.stages():
            stage.checkout()

    def _used_cache(self):
        clist = []
        for stage in self.stages():
            for out in stage.outs:
                if not out.use_cache:
                    continue
                if out.cache not in clist:
                    clist.append(out.cache)
        return clist

    def gc(self):
        clist = self._used_cache()
        for cache in self.cache.all():
            if cache in clist:
                continue
            os.unlink(cache)
            self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache)))

    def push(self, jobs=1):
        self.cloud.push(self._used_cache(), jobs)

    def pull(self, jobs=1):
        self.cloud.pull(self._used_cache(), jobs)
        for stage in self.stages():
            for out in stage.outs:
                if out.use_cache:
{'fraction_non_alphanumeric': 0.10257552141319676, 'fraction_numerical': 0.00044533511467379205, 'mean_word_length': 1.8704729441840648, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1946217', 'n_tokens_mistral': 4183, 'n_tokens_neox': 4041, 'n_words': 866}