userver: /data/code/userver/testsuite/pytest_plugins/pytest_userver/utils/colorize.py Source File
Loading...
Searching...
No Matches
colorize.py
1import argparse
2import contextlib
3import enum
4import itertools
5import json
6import sys
7
8from . import tskv
9
10
11class ColorArg(enum.Enum):
12 AUTO = 'auto'
13 ALWAYS = 'always'
14 NEVER = 'never'
15
16
17# Info about colors https://misc.flogisoft.com/bash/tip_colors_and_formatting
18class Colors:
19 BLACK = '\033[30m'
20 RED = '\033[31m'
21 GREEN = '\033[32m'
22 YELLOW = '\033[33m'
23 BLUE = '\033[34m'
24 GRAY = '\033[37m'
25 DARK_GRAY = '\033[37m'
26 BRIGHT_RED = '\033[91m'
27 BRIGHT_GREEN = '\033[92m'
28 BRIGHT_YELLOW = '\033[93m'
29 DEFAULT = '\033[0m'
30 DEFAULT_BG = '\033[49m'
31 BG_BLACK = '\033[40m'
32
33 # No bright red color, no close colors, no too dark colors
34 __NICE_COLORS = [
35 '\033[38;5;{}m'.format(x)
36 for x in itertools.chain(
37 range(2, 7),
38 range(10, 15),
39 range(38, 51, 2),
40 range(75, 87, 2),
41 range(112, 123, 3),
42 range(128, 159, 3),
43 range(164, 195, 3),
44 range(203, 231, 3),
45 )
46 ]
47
48 @staticmethod
49 def colorize(value):
50 ind = hash(value) % len(Colors.__NICE_COLORS)
51 return Colors.__NICE_COLORS[ind]
52
53
54LEVEL_COLORS = {
55 'TRACE': Colors.DARK_GRAY,
56 'DEBUG': Colors.GRAY,
57 'INFO': Colors.GREEN,
58 'WARNING': Colors.YELLOW,
59 'ERROR': Colors.RED,
60 'CRITICAL': Colors.BRIGHT_RED,
61 'none': Colors.DEFAULT,
62}
63HTTP_STATUS_COLORS = {
64 '2': Colors.GREEN,
65 '3': Colors.GREEN,
66 '4': Colors.YELLOW,
67 '5': Colors.RED,
68}
69
70
72 def __init__(self, *, verbose=False, colors_enabled=True):
73 self._requests = {}
74 self.verbose = verbose
75 self.colors_enabled = colors_enabled
76
77 def colorize_line(self, line):
78 if not line.startswith('tskv\t'):
79 return line
80 return self.colorize_tskv(line)
81
82 def colorize_tskv(self, line):
83 row = tskv.parse_line(line)
84 return self.colorize_row(row)
85
86 def colorize_row(self, row):
87 row = row.copy()
88 flowid = '-'.join([row.get(key, '') for key in ('link', 'trace_id')])
89
90 entry_type = row.pop('_type', None)
91 link = row.get('link', None)
92 level = row.pop('level', 'none')
93 text = row.pop('text', '')
94
95 extra_fields = []
96 if entry_type == 'request':
97 self._requests[link] = self._build_request_info(row)
98 if 'body' in row:
99 extra_fields.append(
100 'request_body='
101 + self.textcolor(
102 try_reformat_json(row.pop('body')), Colors.YELLOW,
103 ),
104 )
105 elif entry_type == 'response':
106 if 'meta_code' in row:
107 status_code = row.pop('meta_code')
108 extra_fields.append(
109 self._http_status('meta_code', status_code),
110 )
111 if not text:
112 text = 'Response finished'
113 extra_fields.append(
114 'response_body='
115 + self.textcolor(
116 try_reformat_json(row.pop('body')), Colors.YELLOW,
117 ),
118 )
119 elif entry_type == 'mockserver_request':
120 text = 'Mockserver request finished'
121 if 'meta_code' in row:
122 status_code = str(row.pop('meta_code'))
123 extra_fields.append(
124 self._http_status('meta_code', status_code),
125 )
126 for key in ('method', 'url', 'status', 'exc_info', 'delay'):
127 value = row.pop(key, None)
128 if value:
129 extra_fields.append(f'{key}={value}')
130
131 if link in self._requests:
132 logid = f'[{self._requests[link]}]'
133 elif link is not None:
134 logid = f'[{link}]'
135 else:
136 logid = '<userver>'
137
138 level_color = LEVEL_COLORS.get(level)
139 flow_color = Colors.colorize(flowid)
140
141 fields = [
142 self.textcolor(f'{level:<8}', level_color),
143 self.textcolor(logid, flow_color),
144 ]
145 if text:
146 fields.append(text)
147 elif self.verbose:
148 fields.append('<NO TEXT>')
149 else:
150 return None
151
152 fields.extend(extra_fields)
153 if self.verbose:
154 fields.extend([f'{k}={v}' for k, v in row.items()])
155 return ' '.join(fields)
156
157 def textcolor(self, text, color):
158 if not self.colors_enabled:
159 return str(text)
160 return f'{color}{text}{Colors.DEFAULT}'
161
162 def _http_status(self, key, status):
163 color = HTTP_STATUS_COLORS.get(status[:1], Colors.DEFAULT)
164 return self.textcolor(f'{key}={status}', color)
165
166 def _build_request_info(self, row):
167 if 'uri' not in row:
168 return None
169 uri = row['uri']
170 method = row.get('method', 'UNKNOWN')
171 return f'{method} {uri}'
172
173
174def format_json(obj):
175 encoded = json.dumps(
176 obj,
177 indent=2,
178 separators=(',', ': '),
179 sort_keys=True,
180 ensure_ascii=False,
181 )
182 return encoded
183
184
185def try_reformat_json(body):
186 try:
187 # TODO: unescape string
188 data = json.loads(body)
189 return format_json(data)
190 except ValueError:
191 return body
192
193
194def colorize(stream, verbose=False, colors_enabled=True):
195 colorizer = Colorizer(verbose=verbose, colors_enabled=colors_enabled)
196 for line in stream:
197 line = line.rstrip('\r\n')
198 color_line = colorizer.colorize_line(line)
199 if color_line is not None:
200 print(color_line)
201
202
203def parse_color(value):
204 if value in ('always', 'force', 'yes', 'enable'):
205 return ColorArg.ALWAYS
206 if value in ('never', 'no', 'disable'):
207 return ColorArg.NEVER
208 if value == 'auto':
209 return ColorArg.AUTO
210 raise ValueError(f'Unknown color option {value!r}')
211
212
213def colorize_main():
214 parser = argparse.ArgumentParser(description='Colorize userver log file.')
215 parser.add_argument(
216 '--verbose', '-v', action='store_true', help='Be verbose',
217 )
218 parser.add_argument(
219 '--color',
220 metavar='WHEN',
221 help=(
222 'Control color highlighting, WHEN is always, never or '
223 'auto (default)'
224 ),
225 nargs='?',
226 type=parse_color,
227 default=ColorArg.AUTO,
228 const=ColorArg.ALWAYS,
229 )
230 parser.add_argument(
231 '--no-color',
232 help='Turn off color highlighting',
233 dest='color',
234 action='store_const',
235 const=ColorArg.NEVER,
236 )
237 parser.add_argument(
238 'log',
239 help='File to colorize, by default stdin is used',
240 default='-',
241 nargs='?',
242 )
243 args = parser.parse_args()
244
245 if args.log == '-':
246 stream = sys.stdin
247 else:
248 stream = open(args.log, 'r')
249
250 if args.color == ColorArg.AUTO:
251 colors_enabled = sys.stdout.isatty()
252 elif args.color == ColorArg.ALWAYS:
253 colors_enabled = True
254 else:
255 colors_enabled = False
256
257 with contextlib.closing(stream):
258 colorize(stream, verbose=args.verbose, colors_enabled=colors_enabled)
259
260
261if __name__ == '__main__':
262 colorize_main()