-
Notifications
You must be signed in to change notification settings - Fork 0
/
handlers.py
47 lines (40 loc) · 1.51 KB
/
handlers.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
from itertools import zip_longest
from pathlib import PurePath
from collections.abc import Callable
def lines(filename: str, chunksize: int) -> None:
filename = PurePath(filename)
with open(filename) as fin:
for part, group in enumerate(zip_longest(*([iter(fin)] * chunksize))):
newfile = filename.with_stem(f"line_part_{part}_{filename.stem}")
with open(newfile, 'w') as fout:
for j, line in enumerate(group, 1):
if line is None:
break
if j == len(group):
fout.write(line.rstrip())
else:
fout.write(line)
def chars(filename: str, chunksize: int) -> None:
filename = PurePath(filename)
groups = []
with open(filename) as fin:
c = fin.read(chunksize)
while c:
groups.append(c)
c = fin.read(chunksize)
for part, group in enumerate(groups, 0):
newfile = filename.with_stem(f"line_part_{part}_{filename.stem}")
with open(newfile, 'w') as fout:
line = 0
for j, ch in enumerate(group, 1):
if ch is None:
break
if j == len(group) or line == 0:
fout.write(ch.rstrip())
else:
fout.write(ch)
line += 1
HANDLERS: dict[str, Callable[[str, int], None]] = {
"lines": lines,
"chars": chars
}