forked from philleonard/maven-progress-bar
-
Notifications
You must be signed in to change notification settings - Fork 0
/
mvnp.py
executable file
·147 lines (123 loc) · 4.68 KB
/
mvnp.py
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#!/usr/bin/env python3
import sys
import re
import progressbar2
import argparse
from colorama import init, Fore
init()
progressbar2.streams.wrap_stdout()
progressbar2.streams.wrap_stderr()
parser = argparse.ArgumentParser()
parser.add_argument('-o', action='store_true', help="Output everything")
parser.add_argument('-no', action='store_true', help="Output nothing")
parser.add_argument('-m', action='store_true', help="Output by Maven only")
parser.add_argument('-i', action='store_true', help="Output INFOS produced by Maven")
parser.add_argument('-w', action='store_true', help="Output WARNINGS produced by Maven")
parser.add_argument('-dc', action='store_true', help="Disable all console colouring")
parser.add_argument('-e', action='store_true', help="Output everything after ERROR produced by Maven")
parser.add_argument('-t', action='store_true', help="Write estimated finish time. ☕️ or ⚔️ ? https://xkcd.com/303/")
parser.add_argument('-n', action='store_true', help="Output artifact names built by Maven")
args = parser.parse_args()
output = args.o
nothing = args.no
maven = args.m
info = args.i
warn = args.w or info
artifacts = args.n
absolute_time = args.t
after_error = args.e
disable_colour = args.dc
def get_colour(colour):
if disable_colour:
return ""
return colour
error_c = "[" + get_colour(Fore.LIGHTRED_EX) + "ERROR" + get_colour(Fore.RESET) + "]"
info_c = "[" + get_colour(Fore.CYAN) + "INFO" + get_colour(Fore.RESET) + "]"
warning_c = "[" + get_colour(Fore.YELLOW) + "WARNING" + get_colour(Fore.RESET) + "]"
error_m = "\[ERROR\]" if maven else "ERROR"
warning_m = "\[WARNING\]" if maven else "WARN"
info_m = "\[INFO\]" if maven else "INFO"
bar_format = \
[
"Maven build: ",
get_colour(Fore.YELLOW),
progressbar2.Percentage(),
get_colour(Fore.RESET),
" ",
progressbar2.Counter(format='(%(value)d of %(max_value)d)'),
get_colour(Fore.LIGHTGREEN_EX),
progressbar2.GranularBar(markers=" ▏▎▍▌▋▊▉█"),
get_colour(Fore.RESET),
" ",
progressbar2.Timer(),
" ",
get_colour(Fore.MAGENTA),
progressbar2.AbsoluteETA(format='Finishes: %(eta)s', format_finished='Finished at %(eta)s')
if absolute_time else progressbar2.AdaptiveETA(),
get_colour(Fore.RESET)
]
def outputline(line):
sys.stdout.write(line.replace("[ERROR]", error_c).replace("[INFO]", info_c).replace("[WARNING]", warning_c))
def match():
count = 0
bar = None
error = False
current_max = 0
for line in sys.stdin:
outputted = nothing
if not outputted:
match_error = re.findall(error_m, line)
if len(match_error) > 0 or (error & after_error):
error = True
outputted = True
outputline(line)
if warn and not outputted:
match_warn = re.findall(warning_m, line)
if len(match_warn) > 0:
outputted = True
outputline(line)
if info and not outputted:
match_info = re.findall(info_m, line)
if len(match_info) > 0:
outputted = True
outputline(line)
if output and not outputted:
outputline(line)
matched = re.findall("\[\d+/\d+\]", line)
if len(matched) > 0:
if artifacts:
nline = line.strip("[INFO] ")
art = find_between(nline, "Building", "[")
fline = "{} {}".format("⚒️️", nline.replace(art, get_colour(Fore.CYAN) + art + get_colour(Fore.RESET)))
sys.stdout.write(fline)
prog = matched[0][1:len(matched[0]) - 1]
fraction = prog.split("/")
if bar is None or int(fraction[1]) != current_max:
current_max = int(fraction[1])
bar = progressbar2.ProgressBar(
widgets=bar_format,
widget_kwargs={'samples': 2},
max_value=current_max,
redirect_stdout=True)
count += 1
# Corner case to allow for chained mvn, or if build resumed then sync
if count > current_max:
count = int(fraction[0])
if bar is not None:
bar.update(count)
if bar is not None and not error:
bar.finish()
sys.stderr.flush()
progressbar2.streams.flush()
def find_between(s, first, last):
try:
start = s.index(first) + len(first)
end = s.index(last, start)
return s[start:end]
except ValueError:
return ""
if __name__ == "__main__":
try:
match()
except KeyboardInterrupt:
progressbar2.streams.unwrap_stdout()