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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
from __future__ import absolute_import
import os
import re
import traceback
from typing import (Any, AnyStr, Callable, Dict, List, MutableMapping,
MutableSequence, Union)
import ruamel.yaml
import six
from ruamel.yaml.comments import CommentedBase, CommentedMap, CommentedSeq
from typing_extensions import Text # pylint: disable=unused-import
# move to a regular typing import when Python 3.3-3.6 is no longer supported
lineno_re = re.compile(u"^(.*?:[0-9]+:[0-9]+: )(( *)(.*))")
def _add_lc_filename(r, source): # type: (ruamel.yaml.comments.CommentedBase, AnyStr) -> None
if isinstance(r, ruamel.yaml.comments.CommentedBase):
r.lc.filename = source
if isinstance(r, MutableSequence):
for d in r:
_add_lc_filename(d, source)
elif isinstance(r, MutableMapping):
for d in six.itervalues(r):
_add_lc_filename(d, source)
def relname(source): # type: (Text) -> Text
if source.startswith("file://"):
source = source[7:]
source = os.path.relpath(source)
return source
def add_lc_filename(r, source): # type: (ruamel.yaml.comments.CommentedBase, Text) -> None
_add_lc_filename(r, relname(source))
def reflow(text, maxline, shift=""): # type: (Text, int, Text) -> Text
if maxline < 20:
maxline = 20
if len(text) > maxline:
sp = text.rfind(' ', 0, maxline)
if sp < 1:
sp = text.find(' ', sp+1)
if sp == -1:
sp = len(text)
if sp < len(text):
return "%s\n%s%s" % (text[0:sp], shift, reflow(text[sp+1:], maxline, shift))
return text
def indent(v, nolead=False, shift=u" ", bullet=u" "): # type: (Text, bool, Text, Text) -> Text
if nolead:
return v.splitlines()[0] + u"\n".join([shift + l for l in v.splitlines()[1:]])
else:
def lineno(i, l): # type: (int, Text) -> Text
r = lineno_re.match(l)
if r is not None:
return r.group(1) + (bullet if i == 0 else shift) + r.group(2)
else:
return (bullet if i == 0 else shift) + l
return u"\n".join([lineno(i, l) for i, l in enumerate(v.splitlines())])
def bullets(textlist, bul): # type: (List[Text], Text) -> Text
if len(textlist) == 1:
return textlist[0]
else:
return "\n".join(indent(t, bullet=bul) for t in textlist)
def strip_dup_lineno(text, maxline=None): # type: (Text, int) -> Text
if maxline is None:
maxline = int(os.environ.get("COLUMNS", "100"))
pre = None
msg = []
maxno = 0
for l in text.splitlines():
g = lineno_re.match(l)
if not g:
continue
maxno = max(maxno, len(g.group(1)))
for l in text.splitlines():
g = lineno_re.match(l)
if not g:
msg.append(l)
continue
if g.group(1) != pre:
shift = maxno + len(g.group(3))
g2 = reflow(g.group(2), maxline-shift, " " * shift)
pre = g.group(1)
msg.append(pre + " " * (maxno-len(g.group(1))) + g2)
else:
g2 = reflow(g.group(2), maxline-maxno, " " * (maxno+len(g.group(3))))
msg.append(" " * maxno + g2)
return "\n".join(msg)
def cmap(d, lc=None, fn=None): # type: (Union[int, float, str, Text, Dict, List], List[int], Text) -> Union[int, float, str, Text, CommentedMap, CommentedSeq]
if lc is None:
lc = [0, 0, 0, 0]
if fn is None:
fn = "test"
if isinstance(d, CommentedMap):
fn = d.lc.filename if hasattr(d.lc, "filename") else fn
for k,v in six.iteritems(d):
if k in d.lc.data:
d[k] = cmap(v, lc=d.lc.data[k], fn=fn)
else:
d[k] = cmap(v, lc, fn=fn)
return d
if isinstance(d, CommentedSeq):
fn = d.lc.filename if hasattr(d.lc, "filename") else fn
for k,v in enumerate(d):
if k in d.lc.data:
d[k] = cmap(v, lc=d.lc.data[k], fn=fn)
else:
d[k] = cmap(v, lc, fn=fn)
return d
if isinstance(d, MutableMapping):
cm = CommentedMap()
for k in sorted(d.keys()):
v = d[k]
if isinstance(v, CommentedBase):
uselc = [v.lc.line, v.lc.col, v.lc.line, v.lc.col]
vfn = v.lc.filename if hasattr(v.lc, "filename") else fn
else:
uselc = lc
vfn = fn
cm[k] = cmap(v, lc=uselc, fn=vfn)
cm.lc.add_kv_line_col(k, uselc)
cm.lc.filename = fn
return cm
if isinstance(d, MutableSequence):
cs = CommentedSeq()
for k,v in enumerate(d):
if isinstance(v, CommentedBase):
uselc = [v.lc.line, v.lc.col, v.lc.line, v.lc.col]
vfn = v.lc.filename if hasattr(v.lc, "filename") else fn
else:
uselc = lc
vfn = fn
cs.append(cmap(v, lc=uselc, fn=vfn))
cs.lc.add_kv_line_col(k, uselc)
cs.lc.filename = fn
return cs
else:
return d
class SourceLine(object):
def __init__(self, item, key=None, raise_type=six.text_type, include_traceback=False): # type: (Any, Any, Callable, bool) -> None
self.item = item
self.key = key
self.raise_type = raise_type
self.include_traceback = include_traceback
def __enter__(self): # type: () -> SourceLine
return self
def __exit__(self,
exc_type, # type: Any
exc_value, # type: Any
tb # type: Any
): # -> Any
if not exc_value:
return
if self.include_traceback:
raise self.makeError("\n".join(traceback.format_exception(exc_type, exc_value, tb)))
else:
raise self.makeError(six.text_type(exc_value))
def makeLead(self): # type: () -> Text
if self.key is None or self.item.lc.data is None or self.key not in self.item.lc.data:
return "%s:%i:%i:" % (self.item.lc.filename if hasattr(self.item.lc, "filename") else "",
(self.item.lc.line or 0)+1,
(self.item.lc.col or 0)+1)
else:
return "%s:%i:%i:" % (self.item.lc.filename if hasattr(self.item.lc, "filename") else "",
(self.item.lc.data[self.key][0] or 0)+1,
(self.item.lc.data[self.key][1] or 0)+1)
def makeError(self, msg): # type: (Text) -> Any
if not isinstance(self.item, ruamel.yaml.comments.CommentedBase):
return self.raise_type(msg)
errs = []
lead = self.makeLead()
for m in msg.splitlines():
if bool(lineno_re.match(m)):
errs.append(m)
else:
errs.append("%s %s" % (lead, m))
return self.raise_type("\n".join(errs))
|
