Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def _ask_if_quit(conversation_id: Text, endpoint: EndpointConfig) -> bool:
"""Display the exit menu.
Return `True` if the previous question should be retried."""
answer = questionary.select(
message="Do you want to stop?",
choices=[
Choice("Continue", "continue"),
Choice("Undo Last", "undo"),
Choice("Fork", "fork"),
Choice("Start Fresh", "restart"),
Choice("Export & Quit", "quit"),
],
).ask()
if not answer or answer == "quit":
# this is also the default answer if the user presses Ctrl-C
await _write_data_to_file(conversation_id, endpoint)
raise Abort()
elif answer == "continue":
# in this case we will just return, and the original
if version in PRERELEASE_FLAVORS and not current_version.pre:
# at this stage it's hard to guess the kind of version bump the
# releaser wants, so we ask them
if version == "alpha":
choices = [
str(current_version.next_alpha("minor")),
str(current_version.next_alpha("micro")),
str(current_version.next_alpha("major")),
]
else:
choices = [
str(current_version.next_release_candidate("minor")),
str(current_version.next_release_candidate("micro")),
str(current_version.next_release_candidate("major")),
]
version = questionary.select(
f"Which {version} do you want to release?", choices=choices,
).ask()
if version:
return version
else:
print("Aborting.")
sys.exit(1)
"default": "/api"
},
{
"type": "text",
"name": "uri",
"message": "Database URI:"
},
{
"type": "select",
"name": "allow_drop_all",
"message": "Allowed to drop all tables: ",
"choices": ["True", "False"]
}
])
res = q.select("Which level of logging would you like?",
choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "NOTSET"]
).ask()
config["logging"] = {
"level": res,
"file": f"{instance_name}.log",
"use_console": True,
"backup_count":5,
"max_size": 1024,
"format": "%(asctime)s - %(name)-14s - %(levelname)-8s - %(message)s",
"datefmt": "%H:%M:%S"
}
return config
.skip_if(PASS_SAVED, default=False)
.ask()
)
else:
github_password = retrieve_pass(github_username)
password_save = (
questionary.select(
"Do you want to save your password?", choices=["Yes", "No"], style=style
)
.skip_if(PASS_SAVED, default=False)
.ask()
)
repo_name = questionary.text("Repository Name:", style=style).ask()
repo_type = questionary.select(
"Repository Type:", choices=["Public", "Private"], style=style
).ask()
gitignore = questionary.text(
""" Gitignore(Optional):
Please enter the desired language name to create
gitignore file, press enter if you don't want to:
""",
style=style,
).ask()
return (
github_username,
github_password,
password_save,
repo_name,
def _request_action_from_user(
predictions: List[Dict[Text, Any]],
sender_id: Text, endpoint: EndpointConfig
) -> (Text, bool):
"""Ask the user to correct an action prediction."""
_print_history(sender_id, endpoint)
choices = [{"name": "{:03.2f} {:40}".format(a.get("score"),
a.get("action")),
"value": a.get("action")}
for a in predictions]
choices = ([{"name": "", "value": OTHER_ACTION}] +
choices)
question = questionary.select("What is the next action of the bot?",
choices)
action_name = _ask_or_abort(question, sender_id, endpoint)
is_new_action = action_name == OTHER_ACTION
if is_new_action:
action_name = _request_free_text_action(sender_id, endpoint)
print("Thanks! The bot will now run {}.\n".format(action_name))
return action_name, is_new_action
def _get_word_embedding_type():
embedding_type = questionary.select(
message="Chose one of the embeddings available: ",
choices=[
Choice(title="random", value="random"),
Choice(title="parscit", value="parscit"),
Choice(title="glove_6B_100", value="glove_6B_100"),
Choice(title="glove_6B_200", value="glove_6B_200"),
Choice(title="glove_6B_300", value="glove_6B_300"),
],
).ask()
return embedding_type
tracker = await retrieve_tracker(endpoint, sender_id)
events = tracker.get("events", [])
session_actions_all = [a["name"] for a in _collect_actions(events)]
session_actions_unique = list(set(session_actions_all))
old_actions = [action["value"] for action in choices]
new_actions = [
{"name": action, "value": OTHER_ACTION + action}
for action in session_actions_unique
if action not in old_actions
]
choices = (
[{"name": "", "value": NEW_ACTION}] + new_actions + choices
)
question = questionary.select("What is the next action of the bot?", choices)
action_name = await _ask_questions(question, sender_id, endpoint)
is_new_action = action_name == NEW_ACTION
if is_new_action:
# create new action
action_name = await _request_free_text_action(sender_id, endpoint)
if action_name.startswith(UTTER_PREFIX):
utter_message = await _request_free_text_utterance(
sender_id, endpoint, action_name
)
NEW_TEMPLATES[action_name] = {utter_message: ""}
elif action_name[:32] == OTHER_ACTION:
# action was newly created in the session, but not this turn
is_new_action = True
"message": "Database URI:",
"default": "sqlite:///default.sqlite"
},
{
"type": "select",
"name": "allow_drop_all",
"message": "Allowed to drop all tables: ",
"choices": ["True", "False"]
}
])
constant_jwt_secret = q.confirm("Do you want a constant JWT secret?").ask()
if constant_jwt_secret:
config["jwt_secret_key"] = str(uuid.uuid1())
res = q.select("Which level of logging would you like?",
choices=["DEBUG", "INFO", "WARNING", "ERROR",
"CRITICAL", "NOTSET"]).ask()
config["logging"] = {
"level": res,
"file": f"{instance_name}.log",
"use_console": True,
"backup_count": 5,
"max_size": 1024,
"format": "%(asctime)s - %(name)-14s - %(levelname)-8s - %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S"
}
return config