| 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 |