meerschaum.utils.prompt

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Functions for interacting with the user.
  7"""
  8
  9from __future__ import annotations
 10
 11import os
 12import meerschaum as mrsm
 13from meerschaum.utils.typing import Any, Union, Optional, Tuple, List
 14
 15
 16def prompt(
 17    question: str = '',
 18    *,
 19    icon: bool = True,
 20    default: Union[str, Tuple[str, str], None] = None,
 21    default_editable: Optional[str] = None,
 22    detect_password: bool = True,
 23    is_password: bool = False,
 24    wrap_lines: bool = True,
 25    noask: bool = False,
 26    silent: bool = False,
 27    **kw: Any
 28) -> str:
 29    """
 30    Ask the user a question and return the answer.
 31    Wrapper around `prompt_toolkit.prompt()` with modified behavior.
 32    For example, an empty string returns default instead of printing it for the user to delete
 33    (`prompt_toolkit` behavior).
 34
 35    Parameters
 36    ----------
 37    question: str
 38        The question to print to the user.
 39
 40    icon: bool, default True
 41        If True, prepend the configured icon.
 42
 43    default: Union[str, Tuple[str, str], None], default None
 44        If the response is '', return the default value.
 45
 46    default_editable: Optional[str], default None
 47        If provided, auto-type this user-editable string in the prompt.
 48
 49    detect_password: bool, default True
 50        If `True`, set the input method to a censored password box if the word `password`
 51        appears in the question.
 52
 53    is_password: default False
 54        If `True`, set the input method to a censored password box.
 55        May be overridden by `detect_password` unless `detect_password` is set to `False`.
 56
 57    wrap_lines: bool, default True
 58        If `True`, wrap the text across multiple lines.
 59        Flag is passed onto `prompt_toolkit`.
 60
 61    noask: bool, default False
 62        If `True`, only print the question and return the default answer.
 63
 64    silent: bool, default False
 65        If `True` do not print anything to the screen, but still block for input.
 66
 67    Returns
 68    -------
 69    A `str` of the input provided by the user.
 70
 71    """
 72    from meerschaum.utils.packages import attempt_import
 73    from meerschaum.utils.formatting import ANSI, CHARSET, highlight_pipes, fill_ansi
 74    from meerschaum.config import get_config
 75    from meerschaum.utils.misc import filter_keywords, remove_ansi
 76    from meerschaum.utils.daemon import running_in_daemon, get_current_daemon
 77    from meerschaum._internal.static import STATIC_CONFIG
 78
 79    original_kwargs = {
 80        'question': question,
 81        'icon': icon,
 82        'default': default,
 83        'default_editable': default_editable,
 84        'detect_password': detect_password,
 85        'is_password': is_password,
 86        'wrap_lines': wrap_lines,
 87        'noask': noask,
 88        'silent': silent,
 89        **kw
 90    }
 91
 92    noask = check_noask(noask)
 93    prompt_toolkit = attempt_import('prompt_toolkit')
 94    question_config = get_config('formatting', 'question', patch=True)
 95
 96    ### if a default is provided, append it to the question.
 97    default_answer = default
 98    if default is not None:
 99        question += " (default: "
100        if isinstance(default, tuple) and len(default) > 1:
101            question += f"{default[0]} [{default[1]}]"
102            default_answer = default[0]
103        else:
104            question += f"{default}"
105        question += ")"
106
107    ### detect password
108    if (detect_password and 'password' in question.lower()) or is_password:
109        kw['is_password'] = True
110
111    ### Add the icon and only color the first line.
112    lines = question.split('\n')
113    first_line = lines[0]
114    other_lines = '' if len(lines) <= 1 else '\n'.join(lines[1:])
115
116    if ANSI:
117        first_line = fill_ansi(highlight_pipes(first_line), **question_config['ansi']['rich'])
118        other_lines = highlight_pipes(other_lines)
119
120    _icon = question_config[CHARSET]['icon']
121    question = (' ' + _icon + ' ') if icon and len(_icon) > 0 else ''
122    question += first_line
123    if len(other_lines) > 0:
124        question += '\n' + other_lines
125
126    if not remove_ansi(question).endswith(' '):
127        question += ' '
128
129    prompt_kwargs = {
130        'message': prompt_toolkit.formatted_text.ANSI(question) if not silent else '',
131        'wrap_lines': wrap_lines,
132        'default': default_editable or '',
133        **filter_keywords(prompt_toolkit.prompt, **kw)
134    }
135
136    printed_question = False
137
138    if not running_in_daemon():
139        answer = prompt_toolkit.prompt(**prompt_kwargs) if not noask else ''
140        printed_question = True
141    else:
142        import json
143        daemon = get_current_daemon()
144        print(STATIC_CONFIG['jobs']['flush_token'], end='', flush=True)
145        wrote_file = False
146        try:
147            with open(daemon.prompt_kwargs_file_path, 'w+', encoding='utf-8') as f:
148                json.dump(original_kwargs, f, separators=(',', ':'))
149            wrote_file = True
150        except Exception:
151            pass
152
153        if not silent and not wrote_file:
154            print(question, end='', flush=True)
155            print(STATIC_CONFIG['jobs']['flush_token'], end='', flush=True)
156            printed_question = True
157
158        try:
159            answer = input() if not noask else ''
160        except EOFError:
161            answer = ''
162
163    if noask and not silent and not printed_question:
164        print(question)
165
166    if answer == '' and default is not None:
167        return default_answer
168
169    return answer
170
171
172def yes_no(
173    question: str = '',
174    options: Tuple[str, str] = ('y', 'n'),
175    default: str = 'y',
176    wrappers: Tuple[str, str] = ('[', ']'),
177    icon: bool = True,
178    yes: bool = False,
179    noask: bool = False,
180    **kw : Any
181) -> bool:
182    """
183    Print a question and prompt the user with a yes / no input.
184    Returns `True` for `'yes'`, False for `'no'`.
185
186    Parameters
187    ----------
188    question: str, default ''
189        The question to print to the user.
190
191    options: Tuple[str, str], default ('y', 'n')
192        The `y/n` options. The first is considered `True`, and all options must be lower case.
193
194    default: str, default y
195        The default option. Is represented with a capital to distinguish that it's the default.
196
197    wrappers: Tuple[str, str], default ('[', ']')
198        Text to print around the '[y/n]' options.
199
200    icon: bool, default True
201        If True, prepend the configured question icon.
202
203    Returns
204    -------
205    A bool indicating the user's choice.
206
207    Examples
208    --------
209    ```python-repl
210    >>> yes_no("Do you like me?", default='y')
211     ❓ Do you like me? [Y/n]
212    True
213    >>> yes_no("Cats or dogs?", options=('cats', 'dogs'))
214     ❓ Cats or dogs? [cats/dogs]
215     Please enter a valid response.
216     ❓ Cats or dogs? [cats/dogs] dogs
217    False
218    ```
219    """
220    from meerschaum.utils.warnings import error, warn
221
222    default = options[0] if yes else default
223    noask = yes or check_noask(noask)
224
225    ending = f" {wrappers[0]}" + "/".join(
226        [
227            o.upper() if o.lower() == default.lower()
228            else o.lower() for o in options
229        ]
230    ) + f"{wrappers[1]}"
231    while True:
232        try:
233            answer = prompt(question + ending, icon=icon, detect_password=False, noask=noask)
234            success = True
235        except KeyboardInterrupt:
236            success = False
237        
238        if not success:
239            error("Error getting response. Aborting...", stack=False)
240        if answer == "":
241            answer = default
242
243        if answer.lower() in options:
244            break
245        warn('Please enter a valid reponse.', stack=False)
246    
247    return answer.lower() == options[0].lower()
248
249
250def choose(
251    question: str,
252    choices: Union[List[str], List[Tuple[str, str]]],
253    default: Union[str, List[str], None] = None,
254    numeric: bool = True,
255    multiple: bool = False,
256    as_indices: bool = False,
257    delimiter: str = ',',
258    icon: bool = True,
259    warn: bool = True,
260    noask: bool = False,
261    **kw
262) -> Union[str, Tuple[str], None]:
263    """
264    Present a list of options and return the user's choice.
265
266    Parameters
267    ----------
268    question: str
269        The question to be printed.
270
271    choices: List[Union[str, Tuple[str, str]]
272        A list of options.
273        If an option is a tuple of two strings, the first string is treated as the index
274        and not displayed. In this case, set `as_indices` to `True` to return the index.
275
276    default: Union[str, List[str], None], default None
277        If the user declines to enter a choice, return this value.
278
279    numeric: bool, default True
280        If `True`, number the items in the list and ask for a number as input.
281        If `False`, require the user to type the complete string.
282
283    multiple: bool, default False
284        If `True`, allow the user to choose multiple answers separated by `delimiter`.
285
286    as_indices: bool, default False
287        If `True`, return the indices for the choices.
288        If a choice is a tuple of two strings, the first is assumed to be the index.
289        Otherwise the index in the list is returned.
290
291    delimiter: str, default ','
292        If `multiple`, separate answers by this string. Raise a warning if this string is contained
293        in any of the choices.
294
295    icon: bool, default True
296        If `True`, include the question icon.
297
298    warn: bool, default True
299        If `True`, raise warnings when invalid input is entered.
300
301    noask: bool, default False
302        If `True`, skip printing the question and return the default value.
303
304    Returns
305    -------
306    A string for a single answer or a tuple of strings if `multiple` is `True`.
307
308    """
309    from meerschaum.utils.warnings import warn as _warn
310    from meerschaum.utils.packages import attempt_import
311    from meerschaum.utils.misc import print_options
312    noask = check_noask(noask)
313
314    ### Handle empty choices.
315    if not choices:
316        if warn:
317            _warn(f"No available choices. Returning default value '{default}'.", stacklevel=3)
318        return default
319
320    ### If the default case is to include multiple answers, allow for multiple inputs.
321    if isinstance(default, list):
322        multiple = True
323
324    choices_indices = {}
325    for i, c in enumerate(choices, start=1):
326        if isinstance(c, tuple):
327            i, c = c
328        choices_indices[i] = c
329
330    choices_values_indices = {v: k for k, v in choices_indices.items()}
331    ordered_keys = list(choices_indices.keys())
332    numeric_map = {str(i): key for i, key in enumerate(ordered_keys, 1)}
333
334    def _enforce_default(d):
335        if d is None:
336            return True
337        if d not in choices_values_indices and d not in choices_indices:
338            if warn:
339                _warn(
340                    f"Default choice '{d}' is not contained in the choices. "
341                    + "Setting numeric = False.",
342                    stacklevel=3
343                )
344            return False
345        return True
346
347    ### Throw a warning if the default isn't a choice.
348    for d in (default if isinstance(default, list) else [default]):
349        if not _enforce_default(d):
350            numeric = False
351            break
352
353    _choices = (
354        [str(k) for k in choices_indices] if numeric
355        else list(choices_indices.values())
356    )
357    if multiple:
358        question += f"\n    Enter your choices, separated by '{delimiter}'.\n"
359
360    altered_choices = {}
361    if multiple and not numeric:
362        delim_replacement = '_' if delimiter != '_' else '-'
363        ### Check if the choices have the delimiter.
364        for i, c in choices_indices.items():
365            if delimiter not in c:
366                continue
367            if warn:
368                _warn(
369                    f"The delimiter '{delimiter}' is contained within choice '{c}'.\n"
370                    + f"Replacing the string '{delimiter}' with '{delim_replacement}' in "
371                    + "the choice for correctly parsing input (will be replaced upon returning the prompt).",
372                    stacklevel=3,
373                )
374            new_c = c.replace(delimiter, delim_replacement)
375            altered_choices[new_c] = c
376            choices_indices[i] = new_c
377
378    question_options = []
379    default_tuple = None
380    if numeric:
381        _default_prompt_str = ''
382        if default is not None:
383            default_list = default if isinstance(default, list) else [default]
384            if multiple and isinstance(default, str):
385                default_list = default.split(delimiter)
386
387            _default_indices = []
388            for d in default_list:
389                key = None
390                if d in choices_values_indices:  # is a value
391                    key = choices_values_indices[d]
392                elif d in choices_indices:  # is an index
393                    key = d
394
395                if key in ordered_keys:
396                    _default_indices.append(str(ordered_keys.index(key) + 1))
397
398            _default_prompt_str = delimiter.join(_default_indices)
399        
400        choices_digits = len(str(len(choices)))
401        for choice_ix, c in enumerate(choices_indices.values(), start=1):
402            question_options.append(
403                f"  {choice_ix}. "
404                + (" " * (choices_digits - len(str(choice_ix))))
405                + f"{c}\n"
406            )
407        default_tuple = (_default_prompt_str, default) if default is not None else None
408    else:
409        default_tuple = default
410        for c in choices_indices.values():
411            question_options.append(f"  • {c}\n")
412
413    if 'completer' not in kw:
414        WordCompleter = attempt_import('prompt_toolkit.completion', lazy=False).WordCompleter
415        kw['completer'] = WordCompleter(
416            [str(v) for v in choices_indices.values()] + [str(i) for i in choices_indices],
417            sentence=True,
418        )
419
420    answers = []
421    while not answers:
422        print_options(question_options, header='')
423        answer = prompt(
424            question,
425            icon=icon,
426            default=default_tuple,
427            noask=noask,
428            **kw
429        )
430        if not answer and default is not None:
431            answer = default if isinstance(default, str) else delimiter.join(default)
432
433        if not answer:
434            if warn:
435                _warn("Please pick a valid choice.", stack=False)
436            continue
437
438        _answers = [answer] if not multiple else [a.strip() for a in answer.split(delimiter)]
439        _answers = [a for a in _answers if a]
440
441        if numeric:
442            _raw_answers = list(_answers)
443            _answers = []
444            for _a in _raw_answers:
445                if _a in choices_values_indices:
446                    _answers.append(str(choices_values_indices[_a]))
447                elif _a in numeric_map:
448                    _answers.append(str(numeric_map[_a]))
449                else:
450                    _answers.append(_a)
451
452        _processed_answers = [altered_choices.get(a, a) for a in _answers]
453
454        valid_answers = []
455        for a in _processed_answers:
456            if a in _choices:
457                valid_answers.append(a)
458
459        if len(valid_answers) != len(_processed_answers):
460            if warn:
461                _warn("Please pick a valid choice.", stack=False)
462            continue
463        answers = valid_answers
464
465    def get_key(key_str):
466        try:
467            return int(key_str)
468        except (ValueError, TypeError):
469            return key_str
470
471    if not multiple:
472        answer = answers[0]
473        if not numeric:
474            return choices_values_indices.get(answer, answer) if as_indices else answer
475        
476        key = get_key(answer)
477        return key if as_indices else choices_indices[key]
478
479    if not numeric:
480        return [choices_values_indices.get(a, a) for a in answers] if as_indices else answers
481
482    final_answers = []
483    for a in answers:
484        key = get_key(a)
485        final_answers.append(key if as_indices else choices_indices[key])
486    return final_answers
487
488
489
490def get_password(
491    username: Optional[str] = None,
492    minimum_length: Optional[int] = None,
493    confirm: bool = True,
494    **kw: Any
495) -> str:
496    """
497    Prompt the user for a password.
498
499    Parameters
500    ----------
501    username: Optional[str], default None
502        If provided, print the username when asking for a password.
503
504    minimum_length: Optional[int], default None
505        If provided, enforce a password of at least this length.
506
507    confirm: bool, default True
508        If `True`, prompt the user for a password twice.
509
510    Returns
511    -------
512    The password string (censored from terminal output when typing).
513
514    Examples
515    --------
516    ```python-repl
517    >>> get_password()
518     ❓ Password: *******
519     ❓ Confirm password: *******
520    'hunter2'
521    ```
522
523    """
524    from meerschaum.utils.warnings import warn
525    while True:
526        password = prompt(
527            "Password" + (f" for user '{username}':" if username is not None else ":"),
528            is_password=True,
529            **kw
530        )
531        if minimum_length is not None and len(password) < minimum_length:
532            warn(
533                "Password is too short. " +
534                f"Please enter a password that is at least {minimum_length} characters.",
535                stack=False
536            )
537            continue
538
539        if not confirm:
540            return password
541
542        _password = prompt(
543            "Confirm password" + (f" for user '{username}':" if username is not None else ":"),
544            is_password=True,
545            **kw
546        )
547        if password != _password:
548            warn("Passwords do not match! Please try again.", stack=False)
549            continue
550        else:
551            return password
552
553
554def get_email(username: Optional[str] = None, allow_omit: bool = True, **kw: Any) -> str:
555    """
556    Prompt the user for an email and enforce that it's valid.
557
558    Parameters
559    ----------
560    username: Optional[str], default None
561        If provided, print the username in the prompt.
562
563    allow_omit: bool, default True
564        If `True`, allow the user to omit the email.
565
566    Returns
567    -------
568    The provided email string.
569
570    Examples
571    --------
572    ```python-repl
573    >>> get_email()
574     ❓ Email (empty to omit): foo@foo
575     Invalid email! Please try again.
576     ❓ Email (empty to omit): foo@foo.com
577    'foo@foo.com'
578    ```
579    """
580    from meerschaum.utils.warnings import warn
581    from meerschaum.utils.misc import is_valid_email
582    while True:
583        email = prompt(
584            "Email" + (f" for user '{username}'" if username is not None else "") +
585            (" (empty to omit):" if allow_omit else ": "),
586            **kw
587        )
588        if (allow_omit and email == '') or is_valid_email(email):
589            return email
590        warn("Invalid email! Please try again.", stack=False)
591
592
593def check_noask(noask: bool = False) -> bool:
594    """
595    Flip `noask` to `True` if `MRSM_NOASK` is set.
596    """
597    from meerschaum._internal.static import STATIC_CONFIG
598    NOASK = STATIC_CONFIG['environment']['noask']
599    if noask:
600        return True
601
602    return (
603        os.environ.get(NOASK, 'false').lower()
604        in ('1', 'true')
605    )
606
607
608def get_connectors_completer(*types: str):
609    """
610    Return a prompt-toolkit Completer object to pass into `prompt()`.
611    """
612    from meerschaum.utils.misc import get_connector_labels
613    prompt_toolkit_completion = mrsm.attempt_import('prompt_toolkit.completion', lazy=False)
614    Completer = prompt_toolkit_completion.Completer
615    Completion = prompt_toolkit_completion.Completion
616
617    class ConnectorCompleter(Completer):
618        def get_completions(self, document, complete_event):
619            for label in get_connector_labels(*types, search_term=document.text):
620                yield Completion(label, start_position=(-1 * len(document.text)))
621
622    return ConnectorCompleter()