Adapt to new version of unadf 1.2.

Adflib 0.8 has been released June 2023, so that this script needed to be
adapted to the new version.
This commit is contained in:
2023-10-18 15:04:48 +02:00
parent 09b2349152
commit ed85f32837

91
uadf
View File

@@ -20,14 +20,15 @@ The patched sources are available from: https://github.com/lclevy/ADFlib
It also requires xdms utility, for optional dms support. It also requires xdms utility, for optional dms support.
Changelog: Changelog:
1.4 Adapt to unadf 1.2 and use Latin1 as default encoding
1.3 Switch to Python3 1.3 Switch to Python3
1.2 Added failsafe for filenames in archive with spaces and nodos message. 1.2 Added failsafe for filenames in archive with spaces and nodos message.
1.1 Moved common code into extfslib library 1.1 Moved common code into extfslib library
1.0 Initial release 1.0 Initial release
Author: Roman 'gryf' Dobosz <gryf73@gmail.com> Author: Roman 'gryf' Dobosz <gryf73@gmail.com>
Date: 2019-06-30 Date: 2023-10-16
Version: 1.3 Version: 1.4
Licence: BSD Licence: BSD
""" """
@@ -46,17 +47,17 @@ class UAdf(Archive):
""" """
Class for interact with c1541 program and MC Class for interact with c1541 program and MC
""" """
LINE_PAT = re.compile(b'\s*(?P<size>\d+)?' LINE_PAT = re.compile(r'\s*(?P<size>\d+)?'
b'\s{2}(?P<date>\d{4}/\d{2}/\d{2})' r'\s{2}(?P<date>\d{4}/\d{2}/\d{2})'
b'\s{2}\s?(?P<time>\d+:\d{2}:\d{2})' r'\s{2}\s?(?P<time>\d+:\d{2}:\d{2})'
b'\s{2}(?P<fpath>.*)') r'\s{2}(?P<fpath>.*)')
ARCHIVER = b"unadf" ARCHIVER = "unadf"
DMS = b"xdms" DMS = "xdms"
CMDS = {"list": b"-lr", CMDS = {"list": "-lr",
"read": b"r", "read": "r",
"write": b"w", "write": "w",
"delete": b"d"} "delete": "d"}
DATETIME = b"%s-%s-%s %02d:%s" DATETIME = "%s-%s-%s %02d:%s"
def __init__(self, fname): def __init__(self, fname):
"""Prepare archive content for operations""" """Prepare archive content for operations"""
@@ -81,8 +82,8 @@ class UAdf(Archive):
def _parse_dt(self, date, time): def _parse_dt(self, date, time):
"""Return parsed datetime which fulfill extfs standards date.""" """Return parsed datetime which fulfill extfs standards date."""
year, month, day = date.split(b"/") year, month, day = date.split("/")
hours, minutes, _unused = time.split(b":") hours, minutes, _unused = time.split(":")
return self.DATETIME % (month, day, year, int(hours), minutes) return self.DATETIME % (month, day, year, int(hours), minutes)
def _ungzip(self): def _ungzip(self):
@@ -103,25 +104,29 @@ class UAdf(Archive):
os.close(fdesc) os.close(fdesc)
try: try:
subprocess.check_call([self.DMS, b'-q', b'u', self._arch, subprocess.check_call([self.DMS, '-q', 'u', self._arch,
"+" + tmp_fname]) "+" + tmp_fname])
self._arch = tmp_fname self._arch = tmp_fname
self._clean = False self._clean = False
except (subprocess.CalledProcessError, OSError): except (subprocess.CalledProcessError, OSError):
pass pass
def _map_name(self, name):
if name.startswith(" "):
new_name = "".join(["~", name[1:]])
return new_name
return name
def _get_dir(self): def _get_dir(self):
"""Retrieve directory""" """Retrieve directory"""
contents = [] contents = []
with open(os.devnull, "w") as fnull: out = subprocess.run([self.ARCHIVER, self.CMDS['list'], self._arch],
try: capture_output=True, encoding="latin-1")
out = subprocess.check_output([self.ARCHIVER,
self.CMDS['list'],
self._arch], stderr=fnull)
except subprocess.CalledProcessError:
return contents
for line in out.split(b"\n"): if out.stderr:
sys.stderr.write(out.stderr)
for line in out.stdout.split("\n"):
match = self.LINE_PAT.match(line) match = self.LINE_PAT.match(line)
if not match: if not match:
continue continue
@@ -129,17 +134,17 @@ class UAdf(Archive):
match_entry = match.groupdict() match_entry = match.groupdict()
entry = {} entry = {}
for key in match_entry: for key in match_entry:
entry[bytes(key, 'utf-8')] = match_entry[key] entry[key] = match_entry[key]
del match_entry del match_entry
entry[b'perms'] = b"-rw-r--r--" entry['perms'] = "-rw-r--r--"
if not entry[b'size']: if not entry['size']:
entry[b'perms'] = b"drwxr-xr-x" entry['perms'] = "drwxr-xr-x"
entry[b'size'] = b"0" entry['size'] = "0"
entry[b'display_name'] = self._map_name(entry[b'fpath']) entry['display_name'] = self._map_name(entry['fpath'])
entry[b'datetime'] = self._parse_dt(entry[b'date'], entry[b'time']) entry['datetime'] = self._parse_dt(entry['date'], entry['time'])
entry[b'uid'] = bytes(str(self._uid), 'utf-8') entry['uid'] = str(self._uid)
entry[b'gid'] = bytes(str(self._gid), 'utf-8') entry['gid'] = str(self._gid)
contents.append(entry) contents.append(entry)
return contents return contents
@@ -151,37 +156,41 @@ class UAdf(Archive):
Add suffix to show user what kind of file do he dealing with. Add suffix to show user what kind of file do he dealing with.
""" """
if not self._contents: if not self._contents:
sys.stderr.write("Nodos or archive error\n") sys.stderr.write("Nodos or image error\n")
return 1 return 1
for entry in self._contents: for entry in self._contents:
sys.stdout.buffer.write(self.ITEM % entry) print(self.ITEM.decode('utf-8')[:-1] % entry)
return 0 return 0
def copyout(self, src, dst): def copyout(self, src, dst):
"""Copy file form the adf image.""" """Copy file form the adf image."""
real_src = self._get_real_name(src) real_src = [e['display_name'] for e in self._contents
if e['display_name'] == src]
if not real_src: if not real_src:
raise IOError("No such file or directory") raise IOError("No such file or directory")
if b" " in real_src: real_src = real_src[0].encode('latin-1')
if " " in real_src:
sys.stderr.write("unadf is unable to operate on filepath with " sys.stderr.write("unadf is unable to operate on filepath with "
"space inside.\nUse affs to mount image and than" "space inside.\nUse affs to mount image and than"
" extract desired files.\n") " extract desired files.\n")
return 1 return 1
extract_dir = tempfile.mkdtemp() extract_dir = tempfile.mkdtemp()
cmd = [self.ARCHIVER, self._arch, real_src, b"-d", extract_dir] cmd = [self.ARCHIVER, "-d", extract_dir, self._arch, real_src]
if check_call(cmd, stdout=open(os.devnull, 'wb'), if check_call(cmd, stdout=open(os.devnull, 'wb'),
stderr=open(os.devnull, 'wb')) != 0: stderr=open(os.devnull, 'wb')) != 0:
shutil.rmtree(extract_dir) shutil.rmtree(extract_dir)
sys.stderr.write("unadf returned with nonzero exit code\n") sys.stderr.write("unadf returned with nonzero exit code\n")
return 1 return 1
shutil.move(os.path.join(bytes(extract_dir, "utf8"), real_src), # use subprocess, as shutil will crash on binary encoded filenames
bytes(dst, "utf8")) subprocess.run([b'mv', os.path.join(extract_dir.encode("latin-1"),
real_src), dst.encode('latin-1')])
shutil.rmtree(extract_dir) shutil.rmtree(extract_dir)
return 0 return 0