Tools/c-analyzer/c_common/scriptutil.py - main - githubmirror/cpython

Tools/c-analyzer/c_common/scriptutil.py
627 строк19 KB
Victor Stinner
gh-105407: Remove unused imports in Tools/c-analyzer/ (#105410)
07 июн 2023, 00:08
07 июн 2023, 00:08d1b0015
import argparse
import contextlib
import logging
import os
import os.path
import shutil
import sys

from . import fsutil, strutil, iterutil, logging as loggingutil


_NOT_SET = object()


def get_prog(spec=None, *, absolute=False, allowsuffix=True):
    if spec is None:
        _, spec = _find_script()
        # This is more natural for prog than __file__ would be.
        filename = sys.argv[0]
    elif isinstance(spec, str):
        filename = os.path.normpath(spec)
        spec = None
    else:
        filename = spec.origin
    if _is_standalone(filename):
        # Check if "installed".
        if allowsuffix or not filename.endswith('.py'):
            basename = os.path.basename(filename)
            found = shutil.which(basename)
            if found:
                script = os.path.abspath(filename)
                found = os.path.abspath(found)
                if os.path.normcase(script) == os.path.normcase(found):
                    return basename
        # It is only "standalone".
        if absolute:
            filename = os.path.abspath(filename)
        return filename
    elif spec is not None:
        module = spec.name
        if module.endswith('.__main__'):
            module = module[:-9]
        return f'{sys.executable} -m {module}'
    else:
        if absolute:
            filename = os.path.abspath(filename)
        return f'{sys.executable} {filename}'


def _find_script():
    frame = sys._getframe(2)
    while frame.f_globals['__name__'] != '__main__':
        frame = frame.f_back

    # This should match sys.argv[0].
    filename = frame.f_globals['__file__']
    # This will be None if -m wasn't used..
    spec = frame.f_globals['__spec__']
    return filename, spec


def is_installed(filename, *, allowsuffix=True):
    if not allowsuffix and filename.endswith('.py'):
        return False
    filename = os.path.abspath(os.path.normalize(filename))
    found = shutil.which(os.path.basename(filename))
    if not found:
        return False
    if found != filename:
        return False
    return _is_standalone(filename)


def is_standalone(filename):
    filename = os.path.abspath(os.path.normalize(filename))
    return _is_standalone(filename)


def _is_standalone(filename):
    return fsutil.is_executable(filename)


##################################
# logging

VERBOSITY = 3

TRACEBACK = os.environ.get('SHOW_TRACEBACK', '').strip()
TRACEBACK = bool(TRACEBACK and TRACEBACK.upper() not in ('0', 'FALSE', 'NO'))


logger = logging.getLogger(__name__)


def configure_logger(verbosity, logger=None, **kwargs):
    if logger is None:
        # Configure the root logger.
        logger = logging.getLogger()
    loggingutil.configure_logger(logger, verbosity, **kwargs)


##################################
# selections

class UnsupportedSelectionError(Exception):
    def __init__(self, values, possible):
        self.values = tuple(values)
        self.possible = tuple(possible)
        super().__init__(f'unsupported selections {self.unique}')

    @property
    def unique(self):
        return tuple(sorted(set(self.values)))


def normalize_selection(selected: str, *, possible=None):
    if selected in (None, True, False):
        return selected
    elif isinstance(selected, str):
        selected = [selected]
    elif not selected:
        return ()

    unsupported = []
    _selected = set()
    for item in selected:
        if not item:
            continue
        for value in item.strip().replace(',', ' ').split():
            if not value:
                continue
            # XXX Handle subtraction (leading "-").
            if possible and value not in possible and value != 'all':
                unsupported.append(value)
            _selected.add(value)
    if unsupported:
        raise UnsupportedSelectionError(unsupported, tuple(possible))
    if 'all' in _selected:
        return True
    return frozenset(selected)


##################################
# CLI parsing helpers

class CLIArgSpec(tuple):
    def __new__(cls, *args, **kwargs):
        return super().__new__(cls, (args, kwargs))

    def __repr__(self):
        args, kwargs = self
        args = [repr(arg) for arg in args]
        for name, value in kwargs.items():
            args.append(f'{name}={value!r}')
        return f'{type(self).__name__}({", ".join(args)})'

    def __call__(self, parser, *, _noop=(lambda a: None)):
        self.apply(parser)
        return _noop

    def apply(self, parser):
        args, kwargs = self
        parser.add_argument(*args, **kwargs)


def apply_cli_argspecs(parser, specs):
    processors = []
    for spec in specs:
        if callable(spec):
            procs = spec(parser)
            _add_procs(processors, procs)
        else:
            args, kwargs = spec
            parser.add_argument(args, kwargs)
    return processors


def _add_procs(flattened, procs):
    # XXX Fail on non-empty, non-callable procs?
    if not procs:
        return
    if callable(procs):
        flattened.append(procs)
    else:
        #processors.extend(p for p in procs if callable(p))
        for proc in procs:
            _add_procs(flattened, proc)


def add_verbosity_cli(parser):
    parser.add_argument('-q', '--quiet', action='count', default=0)
    parser.add_argument('-v', '--verbose', action='count', default=0)

    def process_args(args, *, argv=None):
        ns = vars(args)
        key = 'verbosity'
        if key in ns:
            parser.error(f'duplicate arg {key!r}')
        ns[key] = max(0, VERBOSITY + ns.pop('verbose') - ns.pop('quiet'))
        return key
    return process_args


def add_traceback_cli(parser):
    parser.add_argument('--traceback', '--tb', action='store_true',
                        default=TRACEBACK)
    parser.add_argument('--no-traceback', '--no-tb', dest='traceback',
                        action='store_const', const=False)

    def process_args(args, *, argv=None):
        ns = vars(args)
        key = 'traceback_cm'
        if key in ns:
            parser.error(f'duplicate arg {key!r}')
        showtb = ns.pop('traceback')

        @contextlib.contextmanager
        def traceback_cm():
            restore = loggingutil.hide_emit_errors()
            try:
                yield
            except BrokenPipeError:
                # It was piped to "head" or something similar.
                pass
            except NotImplementedError:
                raise  # re-raise
            except Exception as exc:
                if not showtb:
                    sys.exit(f'ERROR: {exc}')
                raise  # re-raise
            except KeyboardInterrupt:
                if not showtb:
                    sys.exit('\nINTERRUPTED')
                raise  # re-raise
            except BaseException as exc:
                if not showtb:
                    sys.exit(f'{type(exc).__name__}: {exc}')
                raise  # re-raise
            finally:
                restore()
        ns[key] = traceback_cm()
        return key
    return process_args


def add_sepval_cli(parser, opt, dest, choices, *, sep=',', **kwargs):
#    if opt is True:
#        parser.add_argument(f'--{dest}', action='append', **kwargs)
#    elif isinstance(opt, str) and opt.startswith('-'):
#        parser.add_argument(opt, dest=dest, action='append', **kwargs)
#    else:
#        arg = dest if not opt else opt
#        kwargs.setdefault('nargs', '+')
#        parser.add_argument(arg, dest=dest, action='append', **kwargs)
    if not isinstance(opt, str):
        parser.error(f'opt must be a string, got {opt!r}')
    elif opt.startswith('-'):
        parser.add_argument(opt, dest=dest, action='append', **kwargs)
    else:
        kwargs.setdefault('nargs', '+')
        #kwargs.setdefault('metavar', opt.upper())
        parser.add_argument(opt, dest=dest, action='append', **kwargs)

    def process_args(args, *, argv=None):
        ns = vars(args)

        # XXX Use normalize_selection()?
        if isinstance(ns[dest], str):
            ns[dest] = [ns[dest]]
        selections = []
        for many in ns[dest] or ():
            for value in many.split(sep):
                if value not in choices:
                    parser.error(f'unknown {dest} {value!r}')
                selections.append(value)
        ns[dest] = selections
    return process_args


def add_files_cli(parser, *, excluded=None, nargs=None):
    process_files = add_file_filtering_cli(parser, excluded=excluded)
    parser.add_argument('filenames', nargs=nargs or '+', metavar='FILENAME')
    return [
        process_files,
    ]


def add_file_filtering_cli(parser, *, excluded=None):
    parser.add_argument('--start')
    parser.add_argument('--include', action='append')
    parser.add_argument('--exclude', action='append')

    excluded = tuple(excluded or ())

    def process_args(args, *, argv=None):
        ns = vars(args)
        key = 'iter_filenames'
        if key in ns:
            parser.error(f'duplicate arg {key!r}')

        _include = tuple(ns.pop('include') or ())
        _exclude = excluded + tuple(ns.pop('exclude') or ())
        kwargs = dict(
            start=ns.pop('start'),
            include=tuple(_parse_files(_include)),
            exclude=tuple(_parse_files(_exclude)),
            # We use the default for "show_header"
        )
        def process_filenames(filenames, relroot=None):
            return fsutil.process_filenames(filenames, relroot=relroot, **kwargs)
        ns[key] = process_filenames
    return process_args


def _parse_files(filenames):
    for filename, _ in strutil.parse_entries(filenames):
        yield filename.strip()


def add_progress_cli(parser, *, threshold=VERBOSITY, **kwargs):
    parser.add_argument('--progress', dest='track_progress', action='store_const', const=True)
    parser.add_argument('--no-progress', dest='track_progress', action='store_false')
    parser.set_defaults(track_progress=True)

    def process_args(args, *, argv=None):
        if args.track_progress:
            ns = vars(args)
            verbosity = ns.get('verbosity', VERBOSITY)
            if verbosity <= threshold:
                args.track_progress = track_progress_compact
            else:
                args.track_progress = track_progress_flat
    return process_args


def add_failure_filtering_cli(parser, pool, *, default=False):
    parser.add_argument('--fail', action='append',
                        metavar=f'"{{all|{"|".join(sorted(pool))}}},..."')
    parser.add_argument('--no-fail', dest='fail', action='store_const', const=())

    def process_args(args, *, argv=None):
        ns = vars(args)

        fail = ns.pop('fail')
        try:
            fail = normalize_selection(fail, possible=pool)
        except UnsupportedSelectionError as exc:
            parser.error(f'invalid --fail values: {", ".join(exc.unique)}')
        else:
            if fail is None:
                fail = default

            if fail is True:
                def ignore_exc(_exc):
                    return False
            elif fail is False:
                def ignore_exc(_exc):
                    return True
            else:
                def ignore_exc(exc):
                    for err in fail:
                        if type(exc) == pool[err]:
                            return False
                    else:
                        return True
            args.ignore_exc = ignore_exc
    return process_args


def add_kind_filtering_cli(parser, *, default=None):
    parser.add_argument('--kinds', action='append')

    def process_args(args, *, argv=None):
        ns = vars(args)

        kinds = []
        for kind in ns.pop('kinds') or default or ():
            kinds.extend(kind.strip().replace(',', ' ').split())

        if not kinds:
            match_kind = (lambda k: True)
        else:
            included = set()
            excluded = set()
            for kind in kinds:
                if kind.startswith('-'):
                    kind = kind[1:]
                    excluded.add(kind)
                    if kind in included:
                        included.remove(kind)
                else:
                    included.add(kind)
                    if kind in excluded:
                        excluded.remove(kind)
            if excluded:
                if included:
                    ...  # XXX fail?
                def match_kind(kind, *, _excluded=excluded):
                    return kind not in _excluded
            else:
                def match_kind(kind, *, _included=included):
                    return kind in _included
        args.match_kind = match_kind
    return process_args


COMMON_CLI = [
    add_verbosity_cli,
    add_traceback_cli,
    #add_dryrun_cli,
]


def add_commands_cli(parser, commands, *, commonspecs=COMMON_CLI, subset=None):
    arg_processors = {}
    if isinstance(subset, str):
        cmdname = subset
        try:
            _, argspecs, _ = commands[cmdname]
        except KeyError:
            raise ValueError(f'unsupported subset {subset!r}')
        parser.set_defaults(cmd=cmdname)
        arg_processors[cmdname] = _add_cmd_cli(parser, commonspecs, argspecs)
    else:
        if subset is None:
            cmdnames = subset = list(commands)
        elif not subset:
            raise NotImplementedError
        elif isinstance(subset, set):
            cmdnames = [k for k in commands if k in subset]
            subset = sorted(subset)
        else:
            cmdnames = [n for n in subset if n in commands]
        if len(cmdnames) < len(subset):
            bad = tuple(n for n in subset if n not in commands)
            raise ValueError(f'unsupported subset {bad}')

        common = argparse.ArgumentParser(add_help=False)
        common_processors = apply_cli_argspecs(common, commonspecs)
        subs = parser.add_subparsers(dest='cmd')
        for cmdname in cmdnames:
            description, argspecs, _ = commands[cmdname]
            sub = subs.add_parser(
                cmdname,
                description=description,
                parents=[common],
            )
            cmd_processors = _add_cmd_cli(sub, (), argspecs)
            arg_processors[cmdname] = common_processors + cmd_processors
    return arg_processors


def _add_cmd_cli(parser, commonspecs, argspecs):
    processors = []
    argspecs = list(commonspecs or ()) + list(argspecs or ())
    for argspec in argspecs:
        if callable(argspec):
            procs = argspec(parser)
            _add_procs(processors, procs)
        else:
            if not argspec:
                raise NotImplementedError
            args = list(argspec)
            if not isinstance(args[-1], str):
                kwargs = args.pop()
                if not isinstance(args[0], str):
                    try:
                        args, = args
                    except (TypeError, ValueError):
                        parser.error(f'invalid cmd args {argspec!r}')
            else:
                kwargs = {}
            parser.add_argument(*args, **kwargs)
            # There will be nothing to process.
    return processors


def _flatten_processors(processors):
    for proc in processors:
        if proc is None:
            continue
        if callable(proc):
            yield proc
        else:
            yield from _flatten_processors(proc)


def process_args(args, argv, processors, *, keys=None):
    processors = _flatten_processors(processors)
    ns = vars(args)
    extracted = {}
    if keys is None:
        for process_args in processors:
            for key in process_args(args, argv=argv):
                extracted[key] = ns.pop(key)
    else:
        remainder = set(keys)
        for process_args in processors:
            hanging = process_args(args, argv=argv)
            if isinstance(hanging, str):
                hanging = [hanging]
            for key in hanging or ():
                if key not in remainder:
                    raise NotImplementedError(key)
                extracted[key] = ns.pop(key)
                remainder.remove(key)
        if remainder:
            raise NotImplementedError(sorted(remainder))
    return extracted


def process_args_by_key(args, argv, processors, keys):
    extracted = process_args(args, argv, processors, keys=keys)
    return [extracted[key] for key in keys]


##################################
# commands

def set_command(name, add_cli):
    """A decorator factory to set CLI info."""
    def decorator(func):
        if hasattr(func, '__cli__'):
            raise Exception(f'already set')
        func.__cli__ = (name, add_cli)
        return func
    return decorator


##################################
# main() helpers

def filter_filenames(filenames, process_filenames=None, relroot=fsutil.USE_CWD):
    # We expect each filename to be a normalized, absolute path.
    for filename, _, check, _ in _iter_filenames(filenames, process_filenames, relroot):
        if (reason := check()):
            logger.debug(f'{filename}: {reason}')
            continue
        yield filename


def main_for_filenames(filenames, process_filenames=None, relroot=fsutil.USE_CWD):
    filenames, relroot = fsutil.fix_filenames(filenames, relroot=relroot)
    for filename, relfile, check, show in _iter_filenames(filenames, process_filenames, relroot):
        if show:
            print()
            print(relfile)
            print('-------------------------------------------')
        if (reason := check()):
            print(reason)
            continue
        yield filename, relfile


def _iter_filenames(filenames, process, relroot):
    if process is None:
        yield from fsutil.process_filenames(filenames, relroot=relroot)
        return

    onempty = Exception('no filenames provided')
    items = process(filenames, relroot=relroot)
    items, peeked = iterutil.peek_and_iter(items)
    if not items:
        raise onempty
    if isinstance(peeked, str):
        if relroot and relroot is not fsutil.USE_CWD:
            relroot = os.path.abspath(relroot)
        check = (lambda: True)
        for filename, ismany in iterutil.iter_many(items, onempty):
            relfile = fsutil.format_filename(filename, relroot, fixroot=False)
            yield filename, relfile, check, ismany
    elif len(peeked) == 4:
        yield from items
    else:
        raise NotImplementedError


def track_progress_compact(items, *, groups=5, **mark_kwargs):
    last = os.linesep
    marks = iter_marks(groups=groups, **mark_kwargs)
    for item in items:
        last = next(marks)
        print(last, end='', flush=True)
        yield item
    if not last.endswith(os.linesep):
        print()


def track_progress_flat(items, fmt='<{}>'):
    for item in items:
        print(fmt.format(item), flush=True)
        yield item


def iter_marks(mark='.', *, group=5, groups=2, lines=_NOT_SET, sep=' '):
    mark = mark or ''
    group = group if group and group > 1 else 1
    groups = groups if groups and groups > 1 else 1

    sep = f'{mark}{sep}' if sep else mark
    end = f'{mark}{os.linesep}'
    div = os.linesep
    perline = group * groups
    if lines is _NOT_SET:
        # By default we try to put about 100 in each line group.
        perlines = 100 // perline * perline
    elif not lines or lines < 0:
        perlines = None
    else:
        perlines = perline * lines

    if perline == 1:
        yield end
    elif group == 1:
        yield sep

    count = 1
    while True:
        if count % perline == 0:
            yield end
            if perlines and count % perlines == 0:
                yield div
        elif count % group == 0:
            yield sep
        else:
            yield mark
        count += 1