<commit_before>"""Subclasses for the Click commands which implement custom functionality."""
import click
import os
import imp
import logging
import warnings
from click.decorators import _make_command
import importlib
from timeit import default_timer as timer

import pkgutil


# Code to make ipython launch whenever an exception occurs  TODO: add this into the unsync commands
# import sys
# from IPython.core import ultratb
# sys.excepthook = ultratb.FormattedTB(mode='Verbose', color_scheme='Linux', call_pdb=1)


# TODO: Figure out how to use the default_map context setting to read variables out of a config file or something to that effect.
class UnsyncCommands(click.MultiCommand):
    """A Custom MultiCommand which loads commands from individual files in the command_dir."""

    def __init__(self, command_dir=None, *args, **kwargs):
        """Ensure that an UnsyncData object is initialized in the root context. Replaces @pass_data for this object."""
        # kwargs['context_settings'] = {'obj': UnsyncData()}
        self.logger = logging.getLogger('unsync.commands')
        self.command_dir = command_dir
        super(UnsyncCommands, self).__init__(*args, **kwargs)

    def _generate_command_mappings(self):
        """Using the currently configured sys.path setting look for packages that hold unsync commands and generate a mapping of names to actual commands."""
        # 
        commands = {}

        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=Warning)
            for module_info in pkgutil.walk_packages():
                # Only look in packages that start with unsync_
                if module_info.name.startswith('unsync_'):
                    a = importlib.import_module(module_info.name)
                    
                    # Get the prefix for the Usync command.
                    try:
                        prefix = a.unsync_prefix
                    except AttributeError:
                        prefix = module_info.name

                    module_commands = getattr(a, 'unsync_commands', [])
                    for command_path in module_commands:
                        pkg_path = [module_info.name]
                        pkg_path = pkg_path + command_path.split('.')[:-1]
                        pkg_path = '.'.join(pkg_path)
                        command_name = command_path.split('.')[-1:][0]

                        try:
                            package = importlib.import_module(pkg_path)
                        except NameError as e:
                            click.secho(f'Unable to import command package {pkg_path}', fg='red')
                            click.secho(f'Encountered Exception: {e}')
                            continue

                        try:
                            command = getattr(package, command_name)
                        except AttributeError as e:
                            click.secho(f'Unable to import command {command_name} from {pkg_path}', fg='red')
                            click.secho(f'Encountered Exception: {e}')
                            continue

                        command_display_name = getattr(command, 'display_name', command_name)
                        commands[(f'{prefix}.{command_display_name}')] = command
        return commands


    def list_commands(self, ctx):
        """List available commands by trying to import files from the plugins directory."""
        command_mappings = self._generate_command_mappings()
        return command_mappings.keys()

    def get_command(self, ctx, name):
        """Retrieve and return the command to run."""
        command_mappings = self._generate_command_mappings()
        if name in command_mappings:
            return command_mappings[name]
        else:
            click.secho(f'Unable to find command: {name}', fg='red')


class UnsyncCommand(click.Command):
    """A custom command used in Unsync."""

    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger('unsync.command')
        super(UnsyncCommand, self).__init__(*args, **kwargs)

    def invoke(self, ctx):
        start = timer()<commit_msg>Added a quiet keyword to _generate_command_mappings. Now it only generates errors on the first run, after that it is quiet.<commit_after>"""Subclasses for the Click commands which implement custom functionality."""
import click
import os
import imp
import logging
import warnings
from click.decorators import _make_command
import importlib
from timeit import default_timer as timer

import pkgutil


# Code to make ipython launch whenever an exception occurs  TODO: add this into the unsync commands
# import sys
# from IPython.core import ultratb
# sys.excepthook = ultratb.FormattedTB(mode='Verbose', color_scheme='Linux', call_pdb=1)


# TODO: Figure out how to use the default_map context setting to read variables out of a config file or something to that effect.
class UnsyncCommands(click.MultiCommand):
    """A Custom MultiCommand which loads commands from individual files in the command_dir."""

    def __init__(self, command_dir=None, *args, **kwargs):
        """Ensure that an UnsyncData object is initialized in the root context. Replaces @pass_data for this object."""
        # kwargs['context_settings'] = {'obj': UnsyncData()}
        self.logger = logging.getLogger('unsync.commands')
        self.command_dir = command_dir
        super(UnsyncCommands, self).__init__(*args, **kwargs)

    def _generate_command_mappings(self, quiet=False):
        """Using the currently configured sys.path setting look for packages that hold unsync commands and generate a mapping of names to actual commands."""
        commands = {}

        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=Warning)
            for module_info in pkgutil.walk_packages():
                # Only look in packages that start with unsync_
                if module_info.name.startswith('unsync_'):
                    a = importlib.import_module(module_info.name)
                    
                    # Get the prefix for the Usync command.
                    try:
                        prefix = a.unsync_prefix
                    except AttributeError:
                        prefix = module_info.name

                    module_commands = getattr(a, 'unsync_commands', [])
                    for command_path in module_commands:
                        pkg_path = [module_info.name]
                        pkg_path = pkg_path + command_path.split('.')[:-1]
                        pkg_path = '.'.join(pkg_path)
                        command_name = command_path.split('.')[-1:][0]

                        try:
                            package = importlib.import_module(pkg_path)
                        except (NameError, ModuleNotFoundError) as e:
                            if not quiet:
                                click.secho(f'Unable to import command package {pkg_path}', fg='red')
                                click.secho(f'Encountered Exception: {e}')
                            continue

                        try:
                            command = getattr(package, command_name)
                        except AttributeError as e:
                            if not quiet:
                                click.secho(f'Unable to import command {command_name} from {pkg_path}', fg='red')
                                click.secho(f'Encountered Exception: {e}')
                            continue

                        command_display_name = getattr(command, 'display_name', command_name)
                        commands[(f'{prefix}.{command_display_name}')] = command
        return commands


    def list_commands(self, ctx):
        """List available commands by trying to import files from the plugins directory."""
        command_mappings = self._generate_command_mappings()
        return command_mappings.keys()

    def get_command(self, ctx, name):
        """Retrieve and return the command to run."""
        command_mappings = self._generate_command_mappings(quiet=True)  # We dont need to see errors more than once.
        if name in command_mappings:
            return command_mappings[name]
        else:
            click.secho(f'Unable to find command: {name}', fg='red')


class UnsyncCommand(click.Command):
    """A custom command used in Unsync."""

    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger('unsync.command')
        super(UnsyncCommand, self).__init__(*args, **kwargs)

    def invoke(self, ctx):
        start = timer()
{'fraction_non_alphanumeric': 0.08450052089362195, 'fraction_numerical': 0.0009260331056835282, 'mean_word_length': 2.0879199428162973, '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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '112956', 'n_tokens_mistral': 2195, 'n_tokens_neox': 2116, 'n_words': 665}