-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathmerge-bin-cue
executable file
·83 lines (69 loc) · 2.25 KB
/
merge-bin-cue
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
#!/usr/bin/env python
import argparse
import asyncio
import pathlib
import tempfile
async def create_img(path):
if path.suffix != '.cue':
raise RuntimeError
if path.is_dir():
raise RuntimeError
files = []
for line in path.read_text().splitlines():
line = line.strip()
if not line.startswith('FILE '):
continue
parts = line.split('"')
if len(parts) != 3:
raise RuntimeError
filepath = (path.parent / pathlib.Path(parts[1])).absolute()
if not filepath.is_file():
raise RuntimeError
files.append(filepath)
if len(files) == 1:
return
name = path.with_suffix('').name
with tempfile.TemporaryDirectory(
dir=path.parent, prefix='.archive_',
suffix=path.name + '.tmp') as tmp:
tmp = pathlib.Path(tmp).absolute()
proc = await asyncio.create_subprocess_exec(
'binmerge',
'--outdir', tmp,
path.absolute(),
name,
cwd=path.parent.absolute())
await proc.communicate()
assert proc.returncode == 0
tmp_files = list(pathlib.Path(tmp).iterdir())
assert len(tmp_files) == 2
newpaths = []
for tmppath in tmp_files:
newpath = (tmppath.parent.parent / tmppath.name).absolute()
tmppath.rename(newpath)
newpaths.append(newpath)
for oldpath in set(files) - set(newpaths):
oldpath.unlink()
def _all_paths(root):
if root.is_file():
return [root]
else:
stack = [root]
paths = {root}
if root.is_dir() and not root.is_symlink():
while stack:
for path in stack.pop().iterdir():
paths.add(path)
if path.is_dir() and not path.is_symlink():
stack.append(path)
# sorted for deterministic archives ex. tar
return sorted(paths)
async def main():
parser = argparse.ArgumentParser()
parser.add_argument('path', type=pathlib.Path)
args = parser.parse_args()
for path in _all_paths(args.path):
if path.suffix.lower() == '.cue':
await create_img(path)
if __name__ == '__main__':
asyncio.run(main())