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
|
def _line_is_dashes(line):
line = line.strip()
return line == "-" * len(line) and len(line) >= 3
def _process_buf(buf):
for idx, line in enumerate(buf):
if _line_is_dashes(line):
line = line[1:]
buf[idx] = line
if buf and buf[-1][-1:] == "\n":
buf[-1] = buf[-1][:-1]
return buf[:]
def tokenize(iterable, interesting_keys=None, encoding=None):
"""This tokenizes an iterable of newlines as bytes into key value
pairs out of the lektor bulk format. By default it will process all
fields, but optionally it can skip values of uninteresting keys and
will instead yield `None`. The values are left as list of decoded
lines with their endings preserved.
This will not perform any other processing on the data other than
decoding and basic tokenizing.
"""
key = []
buf = []
want_newline = False
is_interesting = True
def _flush_item():
the_key = key[0]
if not is_interesting:
value = None
else:
value = _process_buf(buf)
del key[:], buf[:]
return the_key, value
if encoding is not None:
iterable = (x.decode(encoding, "replace") for x in iterable)
for line in iterable:
line = line.rstrip("\r\n") + "\n"
if line.rstrip() == "---":
want_newline = False
if key:
yield _flush_item()
elif key:
if want_newline:
want_newline = False
if not line.strip():
continue
if is_interesting:
buf.append(line)
else:
bits = line.split(":", 1)
if len(bits) == 2:
key = [bits[0].strip()]
if interesting_keys is None:
is_interesting = True
else:
is_interesting = key[0] in interesting_keys
if is_interesting:
first_bit = bits[1].strip("\t ")
if first_bit.strip():
buf = [first_bit]
else:
buf = []
want_newline = True
if key:
yield _flush_item()
def serialize(iterable, encoding=None):
"""Serializes an iterable of key value pairs into a stream of
string chunks. If an encoding is provided, it will be encoded into that.
This is primarily used by the editor to write back data to a source file.
"""
def _produce(item, escape=False):
if escape:
if _line_is_dashes(item):
item = "-" + item
if encoding is not None:
item = item.encode(encoding)
return item
for idx, (key, value) in enumerate(iterable):
value = value.replace("\r\n", "\n").replace("\r", "\n")
if idx > 0:
yield _produce("---\n")
if "\n" in value or value.strip("\t ") != value:
yield _produce(key + ":\n")
yield _produce("\n")
for line in value.splitlines(True):
yield _produce(line, escape=True)
yield _produce("\n")
else:
yield _produce(f"{key}: {value}\n")
|