1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import absolute_import, print_function, unicode_literals
import json
import os
from ..cli import BaseTryParser
from ..push import push_to_try, history_path
class AgainParser(BaseTryParser):
name = "again"
arguments = [
[
["--index"],
{
"default": 0,
"type": int,
"help": "Index of entry in the history to re-push, "
"where '0' is the most recent (default 0). "
"Use --list to display indices.",
},
],
[
["--list"],
{
"default": False,
"action": "store_true",
"dest": "list_configs",
"help": "Display history and exit",
},
],
[
["--list-tasks"],
{
"default": 0,
"action": "count",
"dest": "list_tasks",
"help": "Like --list, but display selected tasks "
"for each history entry, up to 10. Repeat "
"to display all selected tasks.",
},
],
[
["--purge"],
{
"default": False,
"action": "store_true",
"help": "Remove all history and exit",
},
],
]
common_groups = ["push"]
def run(
index=0, purge=False, list_configs=False, list_tasks=0, message="{msg}", **pushargs
):
if purge:
os.remove(history_path)
return
if not os.path.isfile(history_path):
print("error: history file not found: {}".format(history_path))
return 1
with open(history_path, "r") as fh:
history = fh.readlines()
if list_configs or list_tasks > 0:
for i, data in enumerate(history):
msg, config = json.loads(data)
version = config.get("version", "1")
if version == 1:
tasks = config["tasks"]
elif version == 2:
try_config = config.get("parameters", {}).get("try_task_config", {})
tasks = try_config.get("tasks")
else:
tasks = None
if tasks is not None:
n = len(tasks)
print(
"{index}. ({n} task{s}) {msg}".format(
index=i, msg=msg, n=n, s="" if n == 1 else "s"
)
)
if list_tasks > 0:
indent = " " * 4
if list_tasks > 1:
shown_tasks = tasks
else:
shown_tasks = tasks[:10]
print(indent + ("\n" + indent).join(shown_tasks))
num_hidden_tasks = len(tasks) - len(shown_tasks)
if num_hidden_tasks > 0:
print("{}... and {} more".format(indent, num_hidden_tasks))
else:
print(
"{index}. {msg}".format(
index=i,
msg=msg,
)
)
return
msg, try_task_config = json.loads(history[index])
return push_to_try(
"again", message.format(msg=msg), try_task_config=try_task_config, **pushargs
)
|
