from __future__ import annotations
import getpass
import os
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Any
from typing import Callable
from cleo.formatters.style import Style
from cleo.io.outputs.stream_output import StreamOutput
if TYPE_CHECKING:
from cleo.io.io import IO
Validator = Callable[[str], Any]
Normalizer = Callable[[str], Any]
class Question:
"""
A question that will be asked in a Console.
"""
def __init__(self, question: str, default: Any = None) -> None:
self._question = question
self._default = default
self._attempts: int | None = None
self._hidden = False
self._hidden_fallback = True
self._autocomplete_values: list[str] = []
self._validator: Validator = lambda s: s
self._normalizer: Normalizer = lambda s: s
self._error_message = 'Value "{}" is invalid'
@property
def question(self) -> str:
return self._question
@property
def default(self) -> Any:
return self._default
@property
def autocomplete_values(self) -> list[str]:
return self._autocomplete_values
@property
def max_attempts(self) -> int | None:
return self._attempts
def is_hidden(self) -> bool:
return self._hidden
def hide(self, hidden: bool = True) -> None:
if hidden is True and self._autocomplete_values:
raise RuntimeError("A hidden question cannot use the autocompleter.")
self._hidden = hidden
def set_autocomplete_values(self, autocomplete_values: list[str]) -> None:
if self.is_hidden():
raise RuntimeError("A hidden question cannot use the autocompleter.")
self._autocomplete_values = autocomplete_values
def set_max_attempts(self, attempts: int | None) -> None:
self._attempts = attempts
def set_validator(self, validator: Validator) -> None:
self._validator = validator
def ask(self, io: IO) -> Any:
"""
Asks the question to the user.
"""
if not io.is_interactive():
return self.default
return self._validate_attempts(lambda: self._do_ask(io), io)
def _do_ask(self, io: IO) -> Any:
"""
Asks the question to the user.
"""
self._write_prompt(io)
if not (self._autocomplete_values and self._has_stty_available()):
ret: str | None = None
if self.is_hidden():
try:
ret = self._get_hidden_response(io)
except RuntimeError:
if not self._hidden_fallback:
raise
if not ret:
ret = self._read_from_input(io)
else:
ret = self._autocomplete(io)
if len(ret) <= 0:
ret = self._default
return self._normalizer(ret) # type: ignore[arg-type]
def _write_prompt(self, io: IO) -> None:
"""
Outputs the question prompt.
"""
io.write_error(f"{self._question} ")
def _write_error(self, io: IO, error: Exception) -> None:
"""
Outputs an error message.
"""
io.write_error_line(f"{error!s}")
def _autocomplete(self, io: IO) -> str:
"""
Autocomplete a question.
"""
autocomplete = self._autocomplete_values
ret = ""
i = 0
ofs = -1
matches = list(autocomplete)
num_matches = len(matches)
# Add highlighted text style
style = Style(options=["reverse"])
io.error_output.formatter.set_style("hl", style)
stty_mode = subprocess.check_output(["stty", "-g"]).decode().rstrip("\n")
# Disable icanon (so we can read each keypress) and
# echo (we'll do echoing here instead)
subprocess.check_output(["stty", "-icanon", "-echo"])
try:
# Read a keypress
while True:
c = io.read(1)
# Backspace character
if c == "\177":
if num_matches == 0 and i != 0:
i -= 1
# Move cursor backwards
io.write_error("\033[1D")
if i == 0:
ofs = -1
matches = list(autocomplete)
num_matches = len(matches)
else:
num_matches = 0
# Pop the last character off the end of our string
ret = ret[:i]
# Did we read an escape sequence
elif c == "\033":
c += io.read(2)
# A = Up Arrow. B = Down Arrow
if c[2] == "A" or c[2] == "B":
if c[2] == "A" and ofs == -1:
ofs = 0
if num_matches == 0:
continue
ofs += -1 if c[2] == "A" else 1
ofs = (num_matches + ofs) % num_matches
elif ord(c) < 32:
if c in ["\t", "\n"]:
if num_matches > 0 and ofs != -1:
ret = matches[ofs]
# Echo out remaining chars for current match
io.write_error(ret[i:])
i = len(ret)
if c == "\n":
io.write_error(c)
break
num_matches = 0
continue
else:
io.write_error(c)
ret += c
i += 1
num_matches = 0
ofs = 0
for value in autocomplete:
# If typed characters match the beginning
# chunk of value (e.g. [AcmeDe]moBundle)
if value.startswith(ret) and i != len(value):
num_matches += 1
matches[num_matches - 1] = value
# Erase characters from cursor to end of line
io.write_error("\033[K")
if num_matches > 0 and ofs != -1:
# Save cursor position
io.write_error("\0337")
# Write highlighted text
io.write_error("" + matches[ofs][i:] + "")
# Restore cursor position
io.write_error("\0338")
finally:
subprocess.call(["stty", f"{stty_mode}"])
return ret
def _get_hidden_response(self, io: IO) -> str:
"""
Gets a hidden response from user.
"""
stream = None
if isinstance(io.error_output, StreamOutput):
stream = io.error_output.stream
return getpass.getpass("", stream=stream)
def _validate_attempts(self, interviewer: Callable[[], Any], io: IO) -> Any:
"""
Validates an attempt.
"""
error = None
attempts = self._attempts
while attempts is None or attempts:
if error is not None:
self._write_error(io, error)
try:
return self._validator(interviewer())
except Exception as e:
error = e
if attempts is not None:
attempts -= 1
assert error
raise error
def _read_from_input(self, io: IO) -> str:
"""
Read user input.
"""
ret = io.read_line(4096)
if not ret:
raise RuntimeError("Aborted")
return ret.strip()
def _has_stty_available(self) -> bool:
with Path(os.devnull).open("w") as devnull:
try:
exit_code = subprocess.call(["stty"], stdout=devnull, stderr=devnull)
except Exception:
exit_code = 2
return exit_code == 0