def main():
"""
Start the Slack Client
"""
os.system("clear; figlet 'Slack Gitsin' | lolcat")
history = FileHistory(os.path.expanduser("~/.slackHistory"))
while True:
text = prompt("slack> ", history=history,
auto_suggest=AutoSuggestFromHistory(),
on_abort=AbortAction.RETRY,
style=DocumentStyle,
completer=Completer(fuzzy_match=False,
text_utils=TextUtils()),
complete_while_typing=Always(),
get_bottom_toolbar_tokens=get_bottom_toolbar_tokens,
key_bindings_registry=manager.registry,
accept_action=AcceptAction.RETURN_DOCUMENT
)
slack = Slack(text)
slack.run_command()
python类FileHistory()的实例源码
def __init__(self, parser, engine, options=None):
self.parser = parser
self.engine = engine
self.options = options if options is not None else {}
util.ensure_data_dir_exists()
application = create_prompt_application(
message='> ',
lexer=PygmentsLexer(SqlLexer),
history=FileHistory(os.path.expanduser('~/.aq/history')),
completer=AqCompleter(schemas=engine.available_schemas, tables=engine.available_tables),
auto_suggest=AutoSuggestFromHistory(),
validator=QueryValidator(parser),
on_abort=AbortAction.RETRY,
)
loop = create_eventloop()
self.cli = CommandLineInterface(application=application, eventloop=loop)
self.patch_context = self.cli.patch_stdout_context()
def tagger_repl(tagger, **kwargs):
mxlen = int(kwargs.get('mxlen', 100))
maxw = int(kwargs.get('maxw', 100))
#zeropad = int(kwargs.get('zeropad', 0))
prompt_name = kwargs.get('prompt', 'class> ')
history_file = kwargs.get('history_file', '.history')
history = FileHistory(history_file)
while True:
text = prompt(prompt_name, history=history)
text = text.strip()
if text == 'quit':
break
try:
tokens = text.split(' ')
best = tagger.predict_text(tokens, mxlen=mxlen, maxw=maxw)
print(best)
except Exception as e:
logging.exception('Error')
def test(self):
self._load_auth()
history = FileHistory('.history')
project_id = self._get_current_project_id()
bot = self._load_bot()
while True:
try:
line = prompt('BotHub> ', history=history)
if not line:
continue
event = make_event(line)
context = {}
bot.handle_message(event, context)
except (EOFError, KeyboardInterrupt):
break
except Exception:
traceback.print_exc()
def __init__(self, client, multiline, metadata, *args, **kwargs):
@Condition
def is_multiline():
if not multiline:
return False
text = self.document.text
return not query_is_finished(text, multiline)
super(CLIBuffer, self).__init__(
*args,
completer=CHCompleter(client, metadata),
history=FileHistory(
filename=os.path.expanduser('~/.clickhouse-cli_history')
),
enable_history_search=True,
accept_action=AcceptAction.RETURN_DOCUMENT,
is_multiline=is_multiline,
**kwargs
)
def main(args):
os.environ[ENV_ADDITIONAL_USER_AGENT] = 'AZURECLISHELL/' + __version__
parser = argparse.ArgumentParser(prog='az-shell')
parser.add_argument(
'--style', dest='style', help='the colors of the shell',
choices=get_options())
args = parser.parse_args(args)
azure_folder = cli_config_dir()
if not os.path.exists(azure_folder):
os.makedirs(azure_folder)
ACCOUNT.load(os.path.join(azure_folder, 'azureProfile.json'))
CONFIG.load(os.path.join(azure_folder, 'az.json'))
SESSION.load(os.path.join(azure_folder, 'az.sess'), max_age=3600)
config = SHELL_CONFIGURATION
shell_config_dir = azclishell.configuration.get_config_dir
if args.style:
given_style = args.style
config.set_style(given_style)
else:
given_style = config.get_style()
style = style_factory(given_style)
if config.BOOLEAN_STATES[config.config.get('DEFAULT', 'firsttime')]:
print("When in doubt, ask for 'help'")
config.firsttime()
shell_app = Shell(
completer=AZCOMPLETER,
lexer=AzLexer,
history=FileHistory(
os.path.join(shell_config_dir(), config.get_history())),
app=APPLICATION,
styles=style
)
shell_app.run()
def __init__(self, refresh_resources=True):
shell_dir = os.path.expanduser("~/.kube/shell/")
self.history = FileHistory(os.path.join(shell_dir, "history"))
if not os.path.exists(shell_dir):
os.makedirs(shell_dir)
self.toolbar = Toolbar(self.get_cluster_name, self.get_namespace, self.get_user, self.get_inline_help)
def __init__(self, config, completer):
self._cli = None
self._env = os.environ.copy()
self.history = InMemoryHistory()
self.file_history = FileHistory(
"{}/history".format(AzureShellCache.Instance().get('base_dir')))
self._config = config
self.completer = completer
def classifier_repl(classifier, **kwargs):
mxlen = int(kwargs.get('mxlen', 100))
k = int(kwargs.get('k', 1))
thresh = float(kwargs.get('thresh', 0.0))
zeropad = int(kwargs.get('zeropad', 0))
prompt_name = kwargs.get('prompt', 'class> ')
history_file = kwargs.get('history_file', '.history')
history = FileHistory(history_file)
while True:
text = prompt(prompt_name, history=history)
text = text.strip()
if text == 'quit':
break
try:
tokens = text.split(' ')
outcomes = classifier.classify_text(tokens, mxlen=mxlen, zeropad=zeropad)
k = min(k, len(outcomes))
probs = outcomes[:k]
for prob_i in probs:
if prob_i[1] > thresh:
print('Guess [%s]: %.3f' % prob_i)
except Exception as e:
logging.exception('Error')
def read_stdin():
while True:
try:
if get_variable('HISTORY'):
fileHistory = FileHistory(get_variable('HISTORY'))
else:
fileHistory = None
line = prompt(program + " > ", history=fileHistory, completer=CommandCompleter())
process_line(line)
except EOFError:
break
def cmdloop(self):
history_location = os.path.join(os.path.dirname(os.path.abspath(__file__)),".history")
history = FileHistory(history_location)
print(frontend_message.get_welcome_message())
while True:
try:
while True:
text = prompt(self.prompt, lexer=PythonLexer,
completer=self.completer,
style=DocumentStyle, history=history)
res = self.processor.forward(text)
if(isinstance(res, str)):
res = res.strip()
if(len(res)>0):
print(res)
elif res is not None:
print(res)
except KeyboardInterrupt as ex:
print("^C")
except EOFError:
print(frontend_message.get_bye_message())
break
except Exception as ex:
print(ex)
def _setup_cli_history(self):
self.cli_history = FileHistory('.cli_history')
def shell():
c = Chitin()
cmd_history = FileHistory(os.path.expanduser('~') + '/.chitin.history')
print(WELCOME)
message = VERSION
def get_bottom_toolbar_tokens(cli):
return [(Token.Toolbar, ' '+message)]
style = style_from_dict({
Token.Toolbar: '#ffffff bg:#333333',
})
completer = SystemCompleter()
del completer.completers["executable"]
# Check whether files in and around the current directory have been changed...
for failed in util.check_integrity_set(set(".")):
print("[WARN] '%s' has been modified outside of lab book." % failed)
try:
while True:
cmd_str = ""
while len(cmd_str.strip()) == 0:
cmd_str = prompt(u'===> ',
history=cmd_history,
auto_suggest=AutoSuggestFromHistory(),
completer=completer,
lexer=PygmentsLexer(BashLexer),
get_bottom_toolbar_tokens=get_bottom_toolbar_tokens,
style=style,
on_abort=AbortAction.RETRY,
)
fields = cmd_str.split(" ")
command_set = [" ".join(fields)]
skip, special_command_set = c.attempt_special(cmd_str)
if skip:
continue
if len(special_command_set) > 0:
command_set = special_command_set
#####################################
handled = c.super_handle(command_set)
if handled:
if "message" in handled:
message = handled["message"]
else:
message = VERSION
#####################################
except EOFError:
print("Bye!")
def __init__(self, uri, auth, secure=True, verbose=False):
try:
self.driver = GraphDatabase.driver(uri, auth=auth, encrypted=secure)
except ServiceUnavailable as error:
raise ConsoleError("Could not connect to {} ({})".format(uri, error))
self.uri = uri
self.history = FileHistory(HISTORY_FILE)
self.prompt_args = {
"history": self.history,
"lexer": PygmentsLexer(CypherLexer),
"style": style_from_pygments(VimStyle, {
Token.Prompt: "#ansi{}".format(self.prompt_colour.replace("cyan", "teal")),
Token.TxCounter: "#ansi{} bold".format(self.tx_colour.replace("cyan", "teal")),
})
}
self.lexer = CypherLexer()
self.result_writer = TabularResultWriter()
if verbose:
from .watcher import watch
self.watcher = watch("neo4j.bolt")
self.commands = {
"//": self.set_multi_line,
"/e": self.edit,
"/?": self.help,
"/h": self.help,
"/help": self.help,
"/x": self.exit,
"/exit": self.exit,
"/r": self.run_read_tx,
"/read": self.run_read_tx,
"/w": self.run_write_tx,
"/write": self.run_write_tx,
"/csv": self.set_csv_result_writer,
"/table": self.set_tabular_result_writer,
"/tsv": self.set_tsv_result_writer,
"/config": self.config,
"/kernel": self.kernel,
}
self.session = None
self.tx = None
self.tx_counter = 0
def main(style=None):
if APPLICATION.session["az_interactive_active"]:
logger.warning("You're in the interactive shell already.\n")
return
os.environ[ENV_ADDITIONAL_USER_AGENT] = 'AZURECLISHELL/' + __version__
azure_folder = cli_config_dir()
if not os.path.exists(azure_folder):
os.makedirs(azure_folder)
ACCOUNT.load(os.path.join(azure_folder, 'azureProfile.json'))
CONFIG.load(os.path.join(azure_folder, 'az.json'))
SESSION.load(os.path.join(azure_folder, 'az.sess'), max_age=3600)
config = azclishell.configuration.CONFIGURATION
shell_config_dir = azclishell.configuration.get_config_dir
try:
commands = GatherCommands()
az_completer = AzCompleter(commands)
except IOError: # if there is no cache
az_completer = None
if style:
given_style = style
config.set_style(given_style)
else:
given_style = config.get_style()
style_obj = style_factory(given_style)
if config.BOOLEAN_STATES[config.config.get('DEFAULT', 'firsttime')]:
config.firsttime()
ask_feedback = False
if not config.has_feedback() and frequent_user:
print("\n\nAny comments or concerns? You can use the \'feedback\' command!" +
" We would greatly appreciate it.\n")
ask_feedback = True
shell_app = Shell(
completer=az_completer,
lexer=AzLexer,
history=FileHistory(
os.path.join(shell_config_dir(), config.get_history())),
app=APPLICATION,
styles=style_obj,
user_feedback=ask_feedback
)
shell_app.app.session["az_interactive_active"] = True
shell_app.run()
shell_app.app.session["az_interactive_active"] = False
def start_repl(self, quiet: bool) -> None:
"""
Start the objection repl.
"""
banner = ("""
_ _ _ _
___| |_ |_|___ ___| |_|_|___ ___
| . | . | | | -_| _| _| | . | |
|___|___|_| |___|___|_| |_|___|_|_|
|___|(object)inject(ion) v{0}
Runtime Mobile Exploration
by: @leonjza from @sensepost
""").format(__version__)
if not quiet:
click.secho(banner, bold=True)
click.secho('[tab] for command suggestions', fg='white', dim=True)
# the main application loop is here, reading inputs provided by
# prompt_toolkit and sending it off the the needed handlers
while True:
try:
document = prompt(
get_prompt_tokens=self.get_prompt_tokens,
completer=self.completer,
style=PromptStyle().get_style(),
history=FileHistory(os.path.expanduser('~/.objection/objection_history')),
auto_suggest=AutoSuggestFromHistory(),
on_abort=AbortAction.RETRY,
reserve_space_for_menu=4
)
# check if this is an exit command
if document.strip() in ('quit', 'exit', 'bye'):
click.secho('Exiting...', dim=True)
break
# if we got the reconnect command, handle just that
if self.handle_reconnect(document):
continue
# dispatch to the command handler. if something goes horribly
# wrong, catch it instead of crashing the REPL
try:
# find something to run
self.run_command(document)
except Exception as e:
click.secho(('\n\nAn exception occurred while processing the command. If this '
'looks like a code related error, please file a bug report!'), fg='red')
click.secho('Error: {0}'.format(e), fg='red', bold=True)
except (KeyboardInterrupt, EOFError):
click.secho('Exiting...', dim=True)
break
