def execute_tools(config, path, progress=None):
    """
    Executes the suite of TidyPy tools upon the project and returns the
    issues that are found.

    :param config: the TidyPy configuration to use
    :type config: dict
    :param path: that path to the project to analyze
    :type path: str
    :param progress:
        the progress reporter object that will receive callbacks during the
        execution of the tool suite. If not specified, not progress
        notifications will occur.
    :type progress: tidypy.Progress
    :rtype: tidypy.Collector
    """

    progress = progress or QuietProgress()
    progress.on_start()

    with SyncManager() as manager:
        num_tools = 0
        tools = manager.Queue()
        for name, cls in get_tools().items():
            if config[name]['use'] and cls.can_be_used():
                num_tools += 1
                tools.put({
                    'name': name,
                    'config': config[name],
                })

        collector = Collector(config)
        if not num_tools:
            progress.on_finish()
            return collector

        notifications = manager.Queue()
        environment = manager.dict({
            'finder': Finder(path, config),
        })

        workers = []
        for _ in range(config['workers']):
            worker = Worker(
                args=(
                    tools,
                    notifications,
                    environment,
                ),
            )
            worker.start()
            workers.append(worker)

        while num_tools:
            try:
                notification = notifications.get(True, 0.25)
            except Empty:
                pass
            else:
                if notification['type'] == 'start':
                    progress.on_tool_start(notification['tool'])
                elif notification['type'] == 'complete':
                    collector.add_issues(notification['issues'])
                    progress.on_tool_finish(notification['tool'])
                    num_tools -= 1

    progress.on_finish()

    return collector def execute_reports(
        config,
        path,
        collector,
        on_report_finish=None,
        output_file=None):
    """
    Executes the configured suite of issue reports.

    :param config: the TidyPy configuration to use
    :type config: dict
    :param path: that path to the project that was analyzed
    :type path: str
    :param collector: the issues to report
    :type collector: tidypy.Collector
    """

    reports = get_reports()
    for report in config.get('requested_reports', []):
        if report.get('type') and report['type'] in reports:
            cfg = config.get('report', {}).get(report['type'], {})
            cfg.update(report)
            reporter = reports[report['type']](
                cfg,
                path,
                output_file=output_file,
            )
            reporter.produce(collector)
            if on_report_finish:
                on_report_finish(report) def get_tools():
    """
    Retrieves the TidyPy tools that are available in the current Python
    environment.

    The returned dictionary has keys that are the tool names and values are the
    tool classes.

    :rtype: dict
    """

    # pylint: disable=protected-access

    if not hasattr(get_tools, '_CACHE'):
        get_tools._CACHE = {}
        for entry in pkg_resources.iter_entry_points('tidypy.tools'):
            try:
                get_tools._CACHE[entry.name] = entry.load()
            except ImportError as exc:  # pragma: no cover
                output_error(
                    'Could not load tool "%s" defined by "%s": %s' % (
                        entry,
                        entry.dist,
                        exc,
                    ),
                )
    return get_tools._CACHE def get_reports():
    """
    Retrieves the TidyPy issue reports that are available in the current Python
    environment.

    The returned dictionary has keys are the report names and values are the
    report classes.

    :rtype: dict
    """

    # pylint: disable=protected-access

    if not hasattr(get_reports, '_CACHE'):
        get_reports._CACHE = {}
        for entry in pkg_resources.iter_entry_points('tidypy.reports'):
            try:
                get_reports._CACHE[entry.name] = entry.load()
            except ImportError as exc:  # pragma: no cover
                output_error(
                    'Could not load report "%s" defined by "%s": %s' % (
                        entry,
                        entry.dist,
                        exc,
                    ),
                )
    return get_reports._CACHE def get_extenders():
    """
    Retrieves the TidyPy configuration extenders that are available in the
    current Python environment.

    The returned dictionary has keys are the extender names and values are the
    extender classes.

    :rtype: dict
    """

    # pylint: disable=protected-access

    if not hasattr(get_extenders, '_CACHE'):
        get_extenders._CACHE = {}
        for entry in pkg_resources.iter_entry_points('tidypy.extenders'):
            try:
                get_extenders._CACHE[entry.name] = entry.load()
            except ImportError as exc:  # pragma: no cover
                output_error(
                    'Could not load extender "%s" defined by "%s": %s' % (
                        entry,
                        entry.dist,
                        exc,
                    ),
                )
    return get_extenders._CACHE def get_default_config():
    """
    Produces a stock/out-of-the-box TidyPy configuration.

    :rtype: dict
    """

    config = {}

    for name, cls in get_tools().items():
        config[name] = cls.get_default_config()

    try:
        workers = multiprocessing.cpu_count() - 1
    except NotImplementedError:  # pragma: no cover
        workers = 1
    workers = max(1, min(4, workers))

    config.update({
        'exclude': [],
        'merge-issues': True,
        'workers': workers,
        'disabled': [],
        'noqa': True,
        'extends': [],
        'ignore-missing-extends': False,
    })

    return config def get_specific_config(config_file_path, project_path, use_cache=True):
    """
    Produces a TidyPy configuration from the specified file.

    :param config_file_path: the file containing a TidyPy configuration
    :type config_file_path: str
    :param project_path: the path to the project that is going to be analyzed
    :type project_path: str
    :param use_cache:
        whether or not to use cached versions of any remote/referenced TidyPy
        configurations. If not specified, defaults to ``True``.
    :type use_cache: bool
    :rtype: dict
    """

    if os.path.exists(config_file_path):
        with open(config_file_path, 'r', encoding="utf8") as config_file:
            config = toml.load(config_file)
            if 'tidypy' in config:
                # Originally unintended, but we'll support configuration files
                # where everything in scoped in a "tidypy" table.
                config = config['tidypy']

        config = merge_dict(get_default_config(), config)
        config = process_extensions(config, project_path, use_cache=use_cache)
        return config

    return None def get_project_config(project_path, use_cache=True):
    """
    Produces the Tidypy configuration to use for the specified project.

    If a ``pyproject.toml`` exists, the configuration will be based on that. If
    not, the TidyPy configuration in the user's home directory will be used. If
    one does not exist, the default configuration will be used.

    :param project_path: the path to the project that is going to be analyzed
    :type project_path: str
    :param use_cache:
        whether or not to use cached versions of any remote/referenced TidyPy
        configurations. If not specified, defaults to ``True``.
    :type use_cache: bool
    :rtype: dict
    """

    return get_local_config(project_path, use_cache=use_cache) \
        or get_user_config(project_path, use_cache=use_cache) \
        or get_default_config() def purge_config_cache(location=None):
    """
    Clears out the cache of TidyPy configurations that were retrieved from
    outside the normal locations.
    """

    cache_path = get_cache_path(location)

    if location:
        os.remove(cache_path)
    else:
        shutil.rmtree(cache_path) class GitHook:
    def __init__(self):
        try:
            subprocess.call(['git'])  # noqa: bandit:B603,bandit:B607
        except OSError:
            self._git_available = False
        else:
            self._git_available = True

    def get_hook_dir(self, path, ensure_exists=False):
        git_dir = os.path.join(path, '.git')
        if not os.path.exists(git_dir):
            return None

        hook_dir = os.path.join(git_dir, 'hooks')
        if not os.path.exists(hook_dir):
            if ensure_exists:
                os.mkdir(hook_dir)
                return hook_dir
            return None
        return hook_dir

    def install(self, path, strict):
        hook_dir = self.get_hook_dir(path, ensure_exists=True)
        if not hook_dir:
            raise Exception(
                'Could not find Git configuration directory in: %s' % (
                    path,
                )
            )

        hook_filepath = os.path.join(hook_dir, 'pre-commit')
        if os.path.exists(hook_filepath):
            content = read_file(hook_filepath)
            if 'TIDYPY-INSTALLED-HOOK' not in content:
                raise Exception('A pre-commit hook already exists')

        executable = sys.executable or '/usr/bin/env python'

        with open(hook_filepath, 'w', encoding="utf8") as hook_file:
            hook_file.write(HOOK_TEMPLATE.format(
                executable=executable,
            ))

        os.chmod(hook_filepath, stat.S_IRWXU | stat.S_IRGRP | stat.S_IROTH)

        if self._git_available:
            git_config('--add', 'tidypy.strict', str(strict).lower())

    def remove(self, path):
        hook_dir = self.get_hook_dir(path)
        if not hook_dir:
            return

        hook_filepath = os.path.join(hook_dir, 'pre-commit')
        if not os.path.exists(hook_filepath):
            return

        content = read_file(hook_filepath)
        if 'TIDYPY-INSTALLED-HOOK' not in content:
            raise Exception(
                "The pre-commit hook in place is not TidyPy's",
            )

        os.remove(hook_filepath)

        if self._git_available:
            git_config('--remove-section', 'tidypy') class MercurialHook:
    def get_hgrc(self, path, ensure_exists=False):
        path = Path(path)
        if not path.exists():
            return None

        hg_dir = path / '.hg'
        if not hg_dir.exists():
            return None

        hgrc = hg_dir / 'hgrc'
        if not hgrc.exists():
            if ensure_exists:
                hgrc.touch()
                return hgrc
            return None
        return hgrc

    def install(self, path, strict):
        hgrc = self.get_hgrc(path, ensure_exists=True)
        if not hgrc:
            raise Exception(
                'Could not find/create Mercurial configuration in: %s' % (
                    path,
                )
            )

        config = configparser.ConfigParser()
        config.read(hgrc)

        if not config.has_section('hooks'):
            config.add_section('hooks')
        config.set(
            'hooks',
            'precommit.tidypy',
            'python:tidypy.plugin.mercurial.hook',
        )

        if not config.has_section('tidypy'):
            config.add_section('tidypy')
        config.set('tidypy', 'strict', str(strict))

        with open(hgrc, 'w', encoding="utf8") as config_file:
            config.write(config_file)

    def remove(self, path):
        hgrc = self.get_hgrc(path)
        if not hgrc:
            raise Exception(
                'Could not find Mercurial configuration in: %s' % (
                    path,
                )
            )

        config = configparser.ConfigParser()
        config.read(hgrc)

        if config.has_section('hooks'):
            config.remove_option('hooks', 'precommit.tidypy')

        if config.has_section('tidypy'):
            config.remove_section('tidypy')

        with open(hgrc, 'w', encoding="utf8") as config_file:
            config.write(config_file) class QuietProgress(Progress):
    """
    An implementation of ``tidypy.Progress`` that produces no output.
    """ class ConsoleProgress(Progress):
    """
    An implementation of ``tidypy.Progress`` that outputs a progress bar to the
    console.
    """

    def __init__(self, config):
        super().__init__()
        self._timer = None

        tools = [
            name
            for name in get_tools()
            if config[name]['use']
        ]

        tqdm.monitor_interval = 0
        self._bar = tqdm(
            total=len(tools),
            desc='Analyzing',
            unit='tool',
            dynamic_ncols=True,
            bar_format='Analyzing |{bar}|'
            ' {percentage:3.0f}% [{elapsed}{postfix}]',
        )

    def _refresh(self):
        self._bar.refresh()
        self._timer = Timer(1.0, self._refresh)
        self._timer.start()

    @property
    def currently_executing(self):
        if not self.current_tools:
            return ''
        return '[%s]' % (
            ', '.join(self.current_tools),
        )

    def on_start(self):
        super().on_start()
        self._timer = Timer(1.0, self._refresh)
        self._timer.start()

    def on_tool_start(self, tool):
        super().on_tool_start(tool)
        self._bar.set_postfix({
            'current': ','.join(self.current_tools),
        })

    def on_tool_finish(self, tool):
        do_update = tool in self.current_tools
        super().on_tool_finish(tool)
        if do_update:
            postfix = {
                'current': ','.join(self.current_tools),
            }
            if not postfix['current']:
                del postfix['current']
            self._bar.set_postfix(postfix, refresh=False)
            self._bar.update()

    def on_finish(self):
        super().on_finish()
        print('', file=sys.stderr)
        self._timer.cancel()
        self._bar.close() def output_error(msg):
    """
    Prints the specified string to ``stderr``.

    :param msg: the message to print
    :type msg: str
    """

    click.echo(click.style(msg, fg='red'), err=True)
import csv
import inspect
import os
import click
import basicserial
from collections import OrderedDict
from .core import execute_tools, execute_reports
from .config import (
    get_tools,
    get_reports,
    get_extenders,
    get_default_config,
    get_specific_config,
    get_project_config,
    purge_config_cache,
)
from .plugin.git import GitHook
from .plugin.mercurial import MercurialHook
from .progress import QuietProgress, ConsoleProgress
from .util import output_error
    elif fmt == 'json':
        click.echo(basicserial.to_json(codes, pretty=True))
    elif fmt == 'yaml':
        click.echo(basicserial.to_yaml(codes, pretty=True))
    elif fmt == 'csv':
        writer = csv.writer(click.get_text_stream('stdout'))
        writer.writerow(['tool', 'code', 'message'])
        for tool in codes:
            for code, message in codes[tool].items():
                writer.writerow([tool, code, message])


@main.command(
    'default-config',
    short_help='Outputs a default configuration that can be used to bootstrap'
    ' your own configuration file.',
    help='Outputs a default configuration that can be used to bootstrap'
    ' your own configuration file.',
)
@click.option(
    '--pyproject',
    is_flag=True,
    help='Output the config so that it can be used in a pyproject.toml file.',
)
def default_config(pyproject):
    config = get_default_config()

    # Don't include empty option dicts
    for tool in get_tools():
        if not config[tool]['options']:
            del config[tool]['options']

    if pyproject:
        config = {'tool': {'tidypy': config}}

    click.echo(basicserial.to_toml(config, pretty=True))


@main.command(
    'purge-config-cache',
    short_help='Deletes the cache of configurations retrieved from outside the'
    ' primary configuration.',
    help='Deletes the cache of configurations retrieved from outside the'
    ' primary configuration.',
)
def purge_cache():
    purge_config_cache()


@main.command(
    'install-vcs',
    short_help='Installs TidyPy as a pre-commit hook into the specified VCS.',
    help='''Installs TidyPy as a pre-commit hook into the specified VCS.

Accepts two arguments:

  VCS: The version control system to install the hook into. Choose from: git,
hg

  PATH: The path to the base of the repository to install the hook into. If not
specified, defaults to the current working directory.
''',
)
@click.option(
    '--strict',
    is_flag=True,
    help='Whether or not the hook should prevent the commit if TidyPy finds'
    ' issues.',
)
@click.argument(
    'vcs',
    type=click.Choice(['git', 'hg']),
)
@click.argument(
    'path',
    type=click.Path(exists=True),
    default=os.getcwd(),
)
@click.pass_context
def install_vcs(ctx, vcs, path, strict):
    if vcs == 'hg':
        hook = MercurialHook()
    elif vcs == 'git':
        hook = GitHook()

    try:
        hook.install(path, strict)
    except Exception as exc:  # pylint: disable=broad-except
        output_error('VCS hook installation failed: %s' % (exc,))
        ctx.exit(1)


@main.command(
    'remove-vcs',
    short_help='Removes the TidyPy pre-commit hook from the specified VCS.',
    help='''Removes the TidyPy pre-commit hook from the specified VCS.

Accepts two arguments:

  VCS: The version control system to remove the hook from. Choose from: git,
hg

  PATH: The path to the base of the repository to remove the hook from. If not
specified, defaults to the current working directory.
''',
)
@click.argument(
    'vcs',
    type=click.Choice(['git', 'hg']),
)
@click.argument(
    'path',
    type=click.Path(exists=True),
    default=os.getcwd(),
)
@click.pass_context
def remove_vcs(ctx, vcs, path):
    if vcs == 'hg':
        hook = MercurialHook()
    elif vcs == 'git':

        hook = GitHook()