-
Notifications
You must be signed in to change notification settings - Fork 31
/
__init__.py
445 lines (359 loc) · 14 KB
/
__init__.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
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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
"""
Module for deserializing/serializing to and from VDF
"""
__version__ = "3.0"
__author__ = "Rossen Georgiev"
import re
import sys
import struct
from binascii import crc32
from io import StringIO as unicodeIO
from vdf.vdict import VDFDict
# Py2 & Py3 compatibility
if sys.version_info[0] >= 3:
string_type = str
int_type = int
BOMS = '\ufffe\ufeff'
def strip_bom(line):
return line.lstrip(BOMS)
else:
from StringIO import StringIO as strIO
string_type = basestring
int_type = long
BOMS = '\xef\xbb\xbf\xff\xfe\xfe\xff'
BOMS_UNICODE = '\\ufffe\\ufeff'.decode('unicode-escape')
def strip_bom(line):
return line.lstrip(BOMS if isinstance(line, str) else BOMS_UNICODE)
# string escaping
_unescape_char_map = {
r"\n": "\n",
r"\t": "\t",
r"\v": "\v",
r"\b": "\b",
r"\r": "\r",
r"\f": "\f",
r"\a": "\a",
r"\\": "\\",
r"\?": "?",
r"\"": "\"",
r"\'": "\'",
}
_escape_char_map = {v: k for k, v in _unescape_char_map.items()}
def _re_escape_match(m):
return _escape_char_map[m.group()]
def _re_unescape_match(m):
return _unescape_char_map[m.group()]
def _escape(text):
return re.sub(r"[\n\t\v\b\r\f\a\\\?\"']", _re_escape_match, text)
def _unescape(text):
return re.sub(r"(\\n|\\t|\\v|\\b|\\r|\\f|\\a|\\\\|\\\?|\\\"|\\')", _re_unescape_match, text)
# parsing and dumping for KV1
def parse(fp, mapper=dict, merge_duplicate_keys=True, escaped=True):
"""
Deserialize ``s`` (a ``str`` or ``unicode`` instance containing a VDF)
to a Python object.
``mapper`` specifies the Python object used after deserializetion. ``dict` is
used by default. Alternatively, ``collections.OrderedDict`` can be used if you
wish to preserve key order. Or any object that acts like a ``dict``.
``merge_duplicate_keys`` when ``True`` will merge multiple KeyValue lists with the
same key into one instead of overwriting. You can se this to ``False`` if you are
using ``VDFDict`` and need to preserve the duplicates.
"""
if not issubclass(mapper, dict):
raise TypeError("Expected mapper to be subclass of dict, got %s" % type(mapper))
if not hasattr(fp, 'readline'):
raise TypeError("Expected fp to be a file-like object supporting line iteration")
stack = [mapper()]
expect_bracket = False
re_keyvalue = re.compile(r'^("(?P<qkey>(?:\\.|[^\\"])+)"|(?P<key>#?[a-z0-9\-\_\\\?]+))'
r'([ \t]*('
r'"(?P<qval>(?:\\.|[^\\"])*)(?P<vq_end>")?'
r'|(?P<val>[a-z0-9\-\_\\\?\*\.]+)'
r'))?',
flags=re.I)
for idx, line in enumerate(fp):
if idx == 0:
line = strip_bom(line)
line = line.lstrip()
# skip empty and comment lines
if line == "" or line[0] == '/':
continue
# one level deeper
if line[0] == "{":
expect_bracket = False
continue
if expect_bracket:
raise SyntaxError("vdf.parse: expected openning bracket (line %d)" % (idx + 1))
# one level back
if line[0] == "}":
if len(stack) > 1:
stack.pop()
continue
raise SyntaxError("vdf.parse: one too many closing parenthasis (line %d)" % (idx + 1))
# parse keyvalue pairs
while True:
match = re_keyvalue.match(line)
if not match:
try:
line += next(fp)
continue
except StopIteration:
raise SyntaxError("vdf.parse: unexpected EOF (open key quote?)")
key = match.group('key') if match.group('qkey') is None else match.group('qkey')
val = match.group('val') if match.group('qval') is None else match.group('qval')
if escaped:
key = _unescape(key)
# we have a key with value in parenthesis, so we make a new dict obj (level deeper)
if val is None:
if merge_duplicate_keys and key in stack[-1]:
_m = stack[-1][key]
else:
_m = mapper()
stack[-1][key] = _m
stack.append(_m)
expect_bracket = True
# we've matched a simple keyvalue pair, map it to the last dict obj in the stack
else:
# if the value is line consume one more line and try to match again,
# until we get the KeyValue pair
if match.group('vq_end') is None and match.group('qval') is not None:
try:
line += next(fp)
continue
except StopIteration:
raise SyntaxError("vdf.parse: unexpected EOF (open value quote?)")
stack[-1][key] = _unescape(val) if escaped else val
# exit the loop
break
if len(stack) != 1:
raise SyntaxError("vdf.parse: unclosed parenthasis or quotes (EOF)")
return stack.pop()
def loads(s, **kwargs):
"""
Deserialize ``s`` (a ``str`` or ``unicode`` instance containing a JSON
document) to a Python object.
"""
if not isinstance(s, string_type):
raise TypeError("Expected s to be a str, got %s" % type(s))
try:
fp = unicodeIO(s)
except TypeError:
fp = strIO(s)
return parse(fp, **kwargs)
def load(fp, **kwargs):
"""
Deserialize ``fp`` (a ``.readline()``-supporting file-like object containing
a JSON document) to a Python object.
"""
return parse(fp, **kwargs)
def dumps(obj, pretty=False, escaped=True):
"""
Serialize ``obj`` to a VDF formatted ``str``.
"""
if not isinstance(obj, dict):
raise TypeError("Expected data to be an instance of``dict``")
if not isinstance(pretty, bool):
raise TypeError("Expected pretty to be of type bool")
if not isinstance(escaped, bool):
raise TypeError("Expected escaped to be of type bool")
return ''.join(_dump_gen(obj, pretty, escaped))
def dump(obj, fp, pretty=False, escaped=True):
"""
Serialize ``obj`` as a VDF formatted stream to ``fp`` (a
``.write()``-supporting file-like object).
"""
if not isinstance(obj, dict):
raise TypeError("Expected data to be an instance of``dict``")
if not hasattr(fp, 'write'):
raise TypeError("Expected fp to have write() method")
if not isinstance(pretty, bool):
raise TypeError("Expected pretty to be of type bool")
if not isinstance(escaped, bool):
raise TypeError("Expected escaped to be of type bool")
for chunk in _dump_gen(obj, pretty, escaped):
fp.write(chunk)
def _dump_gen(data, pretty=False, escaped=True, level=0):
indent = "\t"
line_indent = ""
if pretty:
line_indent = indent * level
for key, value in data.items():
if escaped and isinstance(key, string_type):
key = _escape(key)
if isinstance(value, dict):
yield '%s"%s"\n%s{\n' % (line_indent, key, line_indent)
for chunk in _dump_gen(value, pretty, escaped, level+1):
yield chunk
yield "%s}\n" % line_indent
else:
if escaped and isinstance(value, string_type):
value = _escape(value)
yield '%s"%s" "%s"\n' % (line_indent, key, value)
# binary VDF
class BASE_INT(int_type):
def __repr__(self):
return "%s(%d)" % (self.__class__.__name__, self)
class UINT_64(BASE_INT):
pass
class INT_64(BASE_INT):
pass
class POINTER(BASE_INT):
pass
class COLOR(BASE_INT):
pass
BIN_NONE = b'\x00'
BIN_STRING = b'\x01'
BIN_INT32 = b'\x02'
BIN_FLOAT32 = b'\x03'
BIN_POINTER = b'\x04'
BIN_WIDESTRING = b'\x05'
BIN_COLOR = b'\x06'
BIN_UINT64 = b'\x07'
BIN_END = b'\x08'
BIN_INT64 = b'\x0A'
BIN_END_ALT = b'\x0B'
def binary_loads(s, mapper=dict, merge_duplicate_keys=True, alt_format=False):
"""
Deserialize ``s`` (``bytes`` containing a VDF in "binary form")
to a Python object.
``mapper`` specifies the Python object used after deserializetion. ``dict` is
used by default. Alternatively, ``collections.OrderedDict`` can be used if you
wish to preserve key order. Or any object that acts like a ``dict``.
``merge_duplicate_keys`` when ``True`` will merge multiple KeyValue lists with the
same key into one instead of overwriting. You can se this to ``False`` if you are
using ``VDFDict`` and need to preserve the duplicates.
"""
if not isinstance(s, bytes):
raise TypeError("Expected s to be bytes, got %s" % type(s))
if not issubclass(mapper, dict):
raise TypeError("Expected mapper to be subclass of dict, got %s" % type(mapper))
# helpers
int32 = struct.Struct('<i')
uint64 = struct.Struct('<Q')
int64 = struct.Struct('<q')
float32 = struct.Struct('<f')
def read_string(s, idx, wide=False):
end = s.find(b'\x00\x00' if wide else b'\x00', idx)
if end == -1:
raise SyntaxError("Unterminated cstring, index: %d" % idx)
result = s[idx:end]
if wide:
result = result.decode('utf-16')
elif bytes is not str:
result = result.decode('utf-8')
return result, end + (2 if wide else 1)
stack = [mapper()]
idx = 0
CURRENT_BIN_END = BIN_END if not alt_format else BIN_END_ALT
while len(s) > idx:
t = s[idx:idx+1]
idx += 1
if t == CURRENT_BIN_END:
if len(stack) > 1:
stack.pop()
continue
break
key, idx = read_string(s, idx)
if t == BIN_NONE:
if merge_duplicate_keys and key in stack[-1]:
_m = stack[-1][key]
else:
_m = mapper()
stack[-1][key] = _m
stack.append(_m)
elif t == BIN_STRING:
stack[-1][key], idx = read_string(s, idx)
elif t == BIN_WIDESTRING:
stack[-1][key], idx = read_string(s, idx, wide=True)
elif t in (BIN_INT32, BIN_POINTER, BIN_COLOR):
val = int32.unpack_from(s, idx)[0]
if t == BIN_POINTER:
val = POINTER(val)
elif t == BIN_COLOR:
val = COLOR(val)
stack[-1][key] = val
idx += int32.size
elif t == BIN_UINT64:
stack[-1][key] = UINT_64(uint64.unpack_from(s, idx)[0])
idx += uint64.size
elif t == BIN_INT64:
stack[-1][key] = INT_64(int64.unpack_from(s, idx)[0])
idx += int64.size
elif t == BIN_FLOAT32:
stack[-1][key] = float32.unpack_from(s, idx)[0]
idx += float32.size
else:
raise SyntaxError("Unknown data type at index %d: %s" % (idx-1, repr(t)))
if len(s) != idx or len(stack) != 1:
raise SyntaxError("Binary VDF ended at index %d, but length is %d" % (idx, len(s)))
return stack.pop()
def binary_dumps(obj, alt_format=False):
"""
Serialize ``obj`` to a binary VDF formatted ``bytes``.
"""
return b''.join(_binary_dump_gen(obj, alt_format=alt_format))
def _binary_dump_gen(obj, level=0, alt_format=False):
if level == 0 and len(obj) == 0:
return
int32 = struct.Struct('<i')
uint64 = struct.Struct('<Q')
int64 = struct.Struct('<q')
float32 = struct.Struct('<f')
for key, value in obj.items():
if isinstance(key, string_type):
key = key.encode('ascii')
else:
raise TypeError("dict keys must be of type str, got %s" % type(key))
if isinstance(value, dict):
yield BIN_NONE + key + BIN_NONE
for chunk in _binary_dump_gen(value, level+1, alt_format=alt_format):
yield chunk
elif isinstance(value, UINT_64):
yield BIN_UINT64 + key + BIN_NONE + uint64.pack(value)
elif isinstance(value, INT_64):
yield BIN_INT64 + key + BIN_NONE + int64.pack(value)
elif isinstance(value, string_type):
try:
value = value.encode('ascii') + BIN_NONE
yield BIN_STRING
except:
value = value.encode('utf-16') + BIN_NONE*2
yield BIN_WIDESTRING
yield key + BIN_NONE + value
elif isinstance(value, float):
yield BIN_FLOAT32 + key + BIN_NONE + float32.pack(value)
elif isinstance(value, (COLOR, POINTER, int, int_type)):
if isinstance(value, COLOR):
yield BIN_COLOR
elif isinstance(value, POINTER):
yield BIN_POINTER
else:
yield BIN_INT32
yield key + BIN_NONE
yield int32.pack(value)
else:
raise TypeError("Unsupported type: %s" % type(value))
yield BIN_END if not alt_format else BIN_END_ALT
def vbkv_loads(s, mapper=dict, merge_duplicate_keys=True):
"""
Deserialize ``s`` (``bytes`` containing a VBKV to a Python object.
``mapper`` specifies the Python object used after deserializetion. ``dict` is
used by default. Alternatively, ``collections.OrderedDict`` can be used if you
wish to preserve key order. Or any object that acts like a ``dict``.
``merge_duplicate_keys`` when ``True`` will merge multiple KeyValue lists with the
same key into one instead of overwriting. You can se this to ``False`` if you are
using ``VDFDict`` and need to preserve the duplicates.
"""
if s[:4] != b'VBKV':
raise ValueError("Invalid header")
checksum, = struct.unpack('<i', s[4:8])
if checksum != crc32(s[8:]):
raise ValueError("Invalid checksum")
return binary_loads(s[8:], mapper, merge_duplicate_keys, alt_format=True)
def vbkv_dumps(obj):
"""
Serialize ``obj`` to a VBKV formatted ``bytes``.
"""
data = b''.join(_binary_dump_gen(obj, alt_format=True))
checksum = crc32(data)
return b'VBKV' + struct.pack('<i', checksum) + data