gryf/mc_ulzx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Separate ulzx from extfslib

Browse Source
This commit is contained in:
gryf
2022-10-02 10:27:12 +02:00
parent 01b4bf7a41
commit 2d1c387f38
5 changed files with 12 additions and 852 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
README.rst
View File

@@ -1,49 +1,11 @@
======================== =======================
Midnight Commander extfs Midnight Commander ulzx
======================== =======================
Those are Midnight Commander extfs plugins for handling several archive types Midnight Commander extfs plugin for handling lzx Amiga archives.
mostly known from AmigaOS - like **lha**, **lzx** and disk images like **adf**
and **dms**.
Installation Description
============ ===========
See individual installation plugins below. Basically it comes down to:
* copying ``extfslib.py`` and plugin files to ``~/.local/share/mc/extfs.d/``
* installing binary handlers (lha, unlzx, xdms and unadf)
* adding an entry in ``~/.config/mc/mc.ext``::
# arch
regex/\.pattern$
Open=%cd %p/handler_filename://
ULha
====
ULha is an extfs plugin which can be used with lha/lzh/lharc archives.
Personally, I've use it almost exclusively for archives created long time ago
on my Amiga. Both reading from and writing into archive was implemented.
Requirements
------------
ULha requires `free lha <http://lha.sourceforge.jp>`_ implementation to work.
Installation
------------
* copy ``extfslib.py`` and ``ulha`` to ``~/.local/share/mc/extfs.d/``
* add or change entry for files handle in ``~/.config/mc/mc.ext``::
# lha
regex/\.[lL]([Hh][aA]|[Zz][hH])$
Open=%cd %p/ulha://
View=%view{ascii} lha l %f
ULzx
====
ULzx is an extfs plugin which can be used to browse and extract lzx archives, ULzx is an extfs plugin which can be used to browse and extract lzx archives,
which are known almost exclusively from Amiga. which are known almost exclusively from Amiga.
@@ -54,7 +16,7 @@ only reading is supported. Also be aware, that
`unlzx <ftp://us.aminet.net/pub/aminet/misc/unix/unlzx.c.gz.readme>`_ cannot `unlzx <ftp://us.aminet.net/pub/aminet/misc/unix/unlzx.c.gz.readme>`_ cannot
extract files individually, so copying entire archive content is not extract files individually, so copying entire archive content is not
recommended, since on every single file a full archive extract would be recommended, since on every single file a full archive extract would be
performed, which in the end would have impact on performance. done, which in the end would have impact on performance.
Requirements Requirements
------------ ------------
@@ -65,7 +27,8 @@ ULzx requires
Installation Installation
------------ ------------
* copy ``extfslib.py`` and ``ulzx`` to ``~/.local/share/mc/extfs.d/`` * install `extfslib`_
* copy ``ulzx`` to ``~/.local/share/mc/extfs.d/``
* add or change entry for files handle in ``~/.config/mc/mc.ext``:: * add or change entry for files handle in ``~/.config/mc/mc.ext``::
# lzx # lzx
@@ -73,82 +36,11 @@ Installation
Open=%cd %p/ulzx:// Open=%cd %p/ulzx://
View=%view{ascii} unlzx -v %f View=%view{ascii} unlzx -v %f
UAdf
====
UAdf is an extfs plugin suitable for reading .adf, .adz and .dms Amiga floppy
disk images. Due to limitations of the
`unadf <http://freecode.com/projects/unadf>`_, file access inside disk image is
read only.
In case of corrupted or no-dos images, message will be shown.
Requirements
------------
It requires ``unadf`` utility from `ADFlib <https://github.com/lclevy/ADFlib>`_
repository, with included `that commit
<https://github.com/lclevy/ADFlib/commit/d36dc2f395f3e8fcee81f66bc86994e166b6140f>`_
in particular, which introduced separation between filename and comment
attribute on Amiga Fast File System.
If it turns out that your distribution doesn't provide proper version of ADFlib,
there will be a need for building it by hand.
It may be done by using following steps:
#. Grab the `sources
<http://http.debian.net/debian/pool/main/u/unadf/unadf_0.7.11a.orig.tar.gz>`_
and `patches
<http://http.debian.net/debian/pool/main/u/unadf/unadf_0.7.11a-3.debian.tar.gz>`_
from `Debian repository <http://packages.debian.org/sid/unadf>`_.
#. Extract ``unadf_0.7.11a-3.debian.tar.gz`` and ``unadf_0.7.11a.orig.tar.gz``
into some temporary directory::
$ mkdir temp
$ cd temp
$ tar zxf ~/Downloads/unadf_0.7.11a-3.debian.tar.gz
$ tar zxf ~/Downloads/unadf_0.7.11a.orig.tar.gz
$ cd unadf-0.7.11a
#. Apply Debian patches::
$ for i in `cat ../debian/patches/series`; do
> patch -Np1 < "../debian/patches/${i}"
> done
#. Apply the patch from extras directory::
$ patch -Np1 < [path_to_this_repo]/extras/unadf_separate_comment.patch
$ make
$ cp Demo/unadf [destination_path]
#. Place ``unadf`` binary under directory reachable by ``$PATH``.
For optional dms support, `xdms <http://zakalwe.fi/~shd/foss/xdms/>`_ utility is
needed.
Installation
------------
* copy ``extfslib.py`` and ``uadf`` to ``~/.local/share/mc/extfs.d/``
* add or change entry for files handle in ``~/.config/mc/mc.ext``::
# adf
type/^Amiga\ .* disk
Open=%cd %p/uadf://
View=%view{ascii} unadf -lr %f
# adz
regex/\.([aA][dD][zZ])$
Open=%cd %p/uadf://
# dms
regex/\.([dD][mM][sS])$
Open=%cd %p/uadf://
License License
======= =======
This software is licensed under 3-clause BSD license. See LICENSE file for This software is licensed under 3-clause BSD license. See LICENSE file for
details. details.
.. _extfslib: https://github.com/gryf/mc_extfs

241
extfslib.py
View File

@@ -1,241 +0,0 @@
"""
extfslib is a library which contains Archive class to support writing extfs
plugins for Midnight Commander.
Tested against python 3.6 and mc 4.8.22
Changelog:
1.2 Switch to python3
1.1 Added item pattern, and common git/uid attrs
1.0 Initial release
Author: Roman 'gryf' Dobosz <gryf73@gmail.com>
Date: 2019-06-30
Version: 1.2
Licence: BSD
"""
import argparse
import os
import sys
import re
from subprocess import check_output, CalledProcessError
class Archive(object):
"""Archive handle. Provides interface to MC's extfs subsystem"""
LINE_PAT = re.compile(b"^(?P<size>)\s"
b"(?P<perms>)\s"
b"(?P<uid>)\s"
b"(?P<gid>)\s"
b"(?P<date>)\s+"
b"(?P<time>)\s"
b"(?P<fpath>)")
ARCHIVER = b"archiver_name"
CMDS = {"list": b"l",
"read": b"r",
"write": b"w",
"delete": b"d"}
ITEM = (b"%(perms)s 1 %(uid)-8s %(gid)-8s %(size)8s %(datetime)s "
b"%(display_name)s\n")
def __init__(self, fname):
"""Prepare archive content for operations"""
if not os.path.exists(fname):
raise OSError("No such file or directory `%s'" % fname)
self._uid = os.getuid()
self._gid = os.getgid()
self._arch = fname
self.name_map = {}
self._contents = self._get_dir()
def _map_name(self, name):
"""MC still have a bug in extfs subsystem, in case of filepaths with
leading space. This is workaround to this bug, which replaces leading
space with tilda."""
if name.startswith(b" "):
new_name = b"".join([b"~", name[1:]])
return new_name
return name
def _get_real_name(self, name):
"""Get real filepath of the file. See _map_name docstring for
details."""
for item in self._contents:
if item[b'display_name'] == name.encode('utf-8',
'surrogateescape'):
return item[b'fpath']
return None
def _get_dir(self):
"""Prepare archive file listing. Expected keys which every entry
should have are: size, perms, uid, gid, date, time, fpath and
display_name."""
contents = []
out = self._call_command("list")
if not out:
return
for line in out.split(b"\n"):
match = self.LINE_PAT.match(line)
if not match:
continue
entry = match.groupdict()
contents.append(entry)
return contents
def _call_command(self, cmd, src=None, dst=None):
"""
Return status of the provided command, which can be one of:
write
read
delete
list
"""
command = [self.ARCHIVER, self.CMDS.get(cmd), self._arch]
if src and dst:
command.append(src)
command.append(dst)
elif src or dst:
command.append(src and src or dst)
try:
output = check_output(command)
except CalledProcessError:
sys.exit(1)
return output
def list(self):
"""Output contents of the archive to stdout"""
sys.stderr.write("Not supported")
return 1
def run(self, dst):
"""Execute file out of archive"""
sys.stderr.write("Not supported")
return 1
def copyout(self, src, dst):
"""Copy file out of archive"""
sys.stderr.write("Not supported")
return 1
def rm(self, dst):
"""Remove file from archive"""
sys.stderr.write("Not supported")
return 1
def mkdir(self, dst):
"""Create empty directory in archive"""
sys.stderr.write("Not supported")
return 1
def rmdir(self, dst):
"""Removes directory from archive"""
sys.stderr.write("Not supported")
return 1
def copyin(self, dst, src=None):
"""Copy file to the archive"""
sys.stderr.write("Not supported")
return 1
def usage():
"""Print out usage information"""
print ("Usage: %(prg)s {copyin,copyout} ARCHNAME SOURCE DESTINATION\n"
"or: %(prg)s list ARCHNAME\n"
"or: %(prg)s {mkdir,rm,rmdir,run} ARCHNAME TARGET" %
{"prg": sys.argv[0]})
def _parse_args(arch_class):
"""Use ArgumentParser to check for script arguments and execute."""
CALL_MAP = {'list': lambda a: arch_class(a.arch).list(),
'copyin': lambda a: arch_class(a.arch).copyin(a.src, a.dst),
'copyout': lambda a: arch_class(a.arch).copyout(a.src, a.dst),
'mkdir': lambda a: arch_class(a.arch).mkdir(a.dst),
'rm': lambda a: arch_class(a.arch).rm(a.dst),
'run': lambda a: arch_class(a.arch).run(a.dst)}
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(help='supported commands')
parser_list = subparsers.add_parser('list', help="List contents of "
"archive")
parser_copyin = subparsers.add_parser('copyin', help="Copy file into "
"archive")
parser_copyout = subparsers.add_parser('copyout', help="Copy file out of "
"archive")
parser_rm = subparsers.add_parser('rm', help="Delete file from archive")
parser_mkdir = subparsers.add_parser('mkdir', help="Create directory in "
"archive")
parser_run = subparsers.add_parser('run', help="Execute archived file")
parser_list.add_argument('arch', help="Archive filename")
parser_list.set_defaults(func=CALL_MAP['list'])
parser_copyin.add_argument('arch', help="Archive filename")
parser_copyin.add_argument('src', help="Source filename")
parser_copyin.add_argument('dst', help="Destination filename (to be "
"written into archive)")
parser_copyin.set_defaults(func=CALL_MAP['copyin'])
parser_copyout.add_argument('arch', help="D64 Image filename")
parser_copyout.add_argument('src', help="Source filename (to be read from"
" archive")
parser_copyout.add_argument('dst', help="Destination filename")
parser_copyout.set_defaults(func=CALL_MAP['copyout'])
parser_rm.add_argument('arch', help="D64 Image filename")
parser_rm.add_argument('dst', help="File inside archive to be deleted")
parser_rm.set_defaults(func=CALL_MAP['rm'])
parser_mkdir.add_argument('arch', help="archive filename")
parser_mkdir.add_argument('dst', help="Directory name inside archive to "
"be created")
parser_mkdir.set_defaults(func=CALL_MAP['mkdir'])
parser_run.add_argument('arch', help="archive filename")
parser_run.add_argument('dst', help="File to be executed")
parser_run.set_defaults(func=CALL_MAP['run'])
args = parser.parse_args()
return args.func(args)
def parse_args(arch_class):
"""Retrive and parse arguments from commandline and apply them into passed
arch_class class object."""
try:
if sys.argv[1] not in ('list', 'copyin', 'copyout', 'rm', 'mkdir',
"run", "rmdir"):
usage()
sys.exit(2)
except IndexError:
usage()
sys.exit(2)
arch = src = dst = None
try:
arch = sys.argv[2]
if sys.argv[1] in ('copyin', 'copyout'):
src = sys.argv[3]
dst = sys.argv[4]
elif sys.argv[1] in ('rm', 'rmdir', 'run', 'mkdir'):
dst = sys.argv[3]
except IndexError:
usage()
sys.exit(2)
call_map = {'copyin': lambda a, s, d: arch_class(a).copyin(s, d),
'copyout': lambda a, s, d: arch_class(a).copyout(s, d),
'list': lambda a, s, d: arch_class(a).list(),
'mkdir': lambda a, s, d: arch_class(a).mkdir(d),
'rm': lambda a, s, d: arch_class(a).rm(d),
'rmdir': lambda a, s, d: arch_class(a).rmdir(d),
'run': lambda a, s, d: arch_class(a).run(d)}
return call_map[sys.argv[1]](arch, src, dst)

122
extras/unadf_separate_comment.patch
View File

@@ -1,122 +0,0 @@
diff -ur unadf-0.7.11a.orig/Demo/unadf.c unadf-0.7.11a/Demo/unadf.c
--- unadf-0.7.11a.orig/Demo/unadf.c 2013-05-12 17:59:51.214905177 +0200
+++ unadf-0.7.11a/Demo/unadf.c 2013-05-12 17:50:06.843420519 +0200
@@ -58,6 +58,7 @@
puts(" -r : lists directory tree contents");
puts(" -c : use dircache data (must be used with -l)");
puts(" -s : display entries logical block pointer (must be used with -l)");
+ puts(" -m : display file comments, if exists (must be used with -l)");
putchar('\n');
puts(" -v n : mount volume #n instead of default #0 volume");
putchar('\n');
@@ -65,7 +66,8 @@
puts(" -d dir : extract to 'dir' directory");
}
-void printEnt(struct Volume *vol, struct Entry* entry, char *path, BOOL sect)
+void printEnt(struct Volume *vol, struct Entry* entry, char *path, BOOL sect,
+ BOOL comment)
{
/* do not print the links entries, ADFlib do not support them yet properly */
if (entry->type==ST_LFILE || entry->type==ST_LDIR || entry->type==ST_LSOFT)
@@ -89,7 +91,7 @@
printf("%s/",entry->name);
else
printf("%s",entry->name);
- if (entry->comment!=NULL && strlen(entry->comment)>0)
+ if (comment && entry->comment!=NULL && strlen(entry->comment)>0)
printf(", %s",entry->comment);
putchar('\n');
@@ -199,13 +201,14 @@
}
-void printTree(struct Volume *vol, struct List* tree, char* path, BOOL sect)
+void printTree(struct Volume *vol, struct List* tree, char* path, BOOL sect,
+ BOOL comment)
{
char *buf;
struct Entry* entry;
while(tree) {
- printEnt(vol, tree->content, path, sect);
+ printEnt(vol, tree->content, path, sect, comment);
if (tree->subdir!=NULL) {
entry = (struct Entry*)tree->content;
if (strlen(path)>0) {
@@ -215,11 +218,11 @@
return;
}
sprintf(buf,"%s/%s", path, entry->name);
- printTree(vol, tree->subdir, buf, sect);
+ printTree(vol, tree->subdir, buf, sect, comment);
free(buf);
}
else
- printTree(vol, tree->subdir, entry->name, sect);
+ printTree(vol, tree->subdir, entry->name, sect, comment);
}
tree = tree->next;
}
@@ -370,12 +373,10 @@
int main(int argc, char* argv[])
{
int i, j;
- BOOL rflag, lflag, xflag, cflag, vflag, sflag, dflag, pflag, qflag;
+ BOOL rflag, lflag, xflag, cflag, vflag, sflag, dflag, pflag, qflag, mflag;
struct List* files, *rtfiles;
char *devname, *dirname;
- char strbuf[80];
unsigned char *extbuf;
- int vInd, dInd, fInd, aInd;
BOOL nextArg;
struct Device *dev;
@@ -389,8 +390,7 @@
exit(0);
}
- rflag = lflag = cflag = vflag = sflag = dflag = pflag = qflag = FALSE;
- vInd = dInd = fInd = aInd = -1;
+ rflag = lflag = cflag = vflag = sflag = dflag = pflag = qflag = mflag = FALSE;
xflag = TRUE;
dirname = NULL;
devname = NULL;
@@ -430,6 +430,9 @@
case 's':
sflag = TRUE;
break;
+ case 'm':
+ mflag = TRUE;
+ break;
case 'c':
cflag = TRUE;
break;
@@ -522,13 +525,13 @@
if (!rflag) {
cell = list = adfGetDirEnt(vol,vol->curDirPtr);
while(cell) {
- printEnt(vol,cell->content,"", sflag);
+ printEnt(vol,cell->content,"", sflag, mflag);
cell = cell->next;
}
adfFreeDirList(list);
} else {
cell = list = adfGetRDirEnt(vol,vol->curDirPtr,TRUE);
- printTree(vol,cell,"", sflag);
+ printTree(vol,cell,"", sflag, mflag);
adfFreeDirList(list);
}
}else if (xflag) {
diff -ur unadf-0.7.11a.orig/Demo/unadf.usage unadf-0.7.11a/Demo/unadf.usage
--- unadf-0.7.11a.orig/Demo/unadf.usage 2006-12-03 15:27:00.000000000 +0100
+++ unadf-0.7.11a/Demo/unadf.usage 2013-05-12 17:40:23.116966854 +0200
@@ -3,6 +3,7 @@
-r : lists directory tree contents
-c : use dircache data (must be used with -l)
-s : display entries logical block pointer (must be used with -l)
+ -m : display file comments, if exists (must be used with -l)
-v n : mount volume #n instead of default #0 volume

187
uadf
View File

@@ -1,187 +0,0 @@
#!/usr/bin/env python3
"""
UADF Virtual filesystem
This extfs provides quick and dirty read-only access to disk image files for
the Commodore Amiga adf or adz (gzipped adfs) and dms.
It requires the unadf utility, unfortunately there is no original sources,
since authors page doesn't exists anymore. Luckily, there is a copy of the
source (and useful patches) in Debian repository:
http://packages.debian.org/sid/unadf
There should be one change made to the source of unadf, though. While using
"-lr" switch it by default also displays comments, separated by the comma.
However there is no way to distinguish where filename ends and comment starts,
if comment or filename already contains any comma.
The patched sources are available from: https://github.com/lclevy/ADFlib
It also requires xdms utility, for optional dms support.
Changelog:
1.3 Switch to Python3
1.2 Added failsafe for filenames in archive with spaces and nodos message.
1.1 Moved common code into extfslib library
1.0 Initial release
Author: Roman 'gryf' Dobosz <gryf73@gmail.com>
Date: 2019-06-30
Version: 1.3
Licence: BSD
"""
import sys
import re
import os
import gzip
from subprocess import check_output, check_call, CalledProcessError
from tempfile import mkstemp, mkdtemp
import shutil
from extfslib import Archive, parse_args
class UAdf(Archive):
"""
Class for interact with c1541 program and MC
"""
LINE_PAT = re.compile(b'\s*(?P<size>\d+)?'
b'\s{2}(?P<date>\d{4}/\d{2}/\d{2})'
b'\s{2}\s?(?P<time>\d+:\d{2}:\d{2})'
b'\s{2}(?P<fpath>.*)')
ARCHIVER = b"unadf"
DMS = b"xdms"
CMDS = {"list": b"-lr",
"read": b"r",
"write": b"w",
"delete": b"d"}
DATETIME = b"%s-%s-%s %02d:%s"
def __init__(self, fname):
"""Prepare archive content for operations"""
self._clean = True
self._arch = fname
if fname.lower().endswith(".adz"):
self._ungzip()
if fname.lower().endswith(".dms"):
self._undms()
super(UAdf, self).__init__(self._arch)
def __del__(self):
"""Cleanup"""
if not self._clean:
try:
os.unlink(self._arch)
except OSError:
pass
def _parse_dt(self, date, time):
"""Return parsed datetime which fulfill extfs standards date."""
year, month, day = date.split(b"/")
hours, minutes, _unused = time.split(b":")
return self.DATETIME % (month, day, year, int(hours), minutes)
def _ungzip(self):
"""Create temporary file for ungzipped adf file since unadf does not
accept gzipped content in any way including reading from stdin."""
fdesc, tmp_fname = mkstemp(suffix=".adf")
os.close(fdesc)
with gzip.open(self._arch) as gobj:
with open(tmp_fname, "wb") as fobj:
fobj.write(gobj.read())
self._arch = tmp_fname
self._clean = False
def _undms(self):
"""Create temporary adf file extracted from dms."""
fdesc, tmp_fname = mkstemp(suffix=".adf")
os.close(fdesc)
try:
check_call([self.DMS, b'-q', b'u', self._arch, "+" + tmp_fname])
self._arch = tmp_fname
self._clean = False
except (CalledProcessError, OSError):
pass
def _get_dir(self):
"""Retrieve directory"""
contents = []
with open(os.devnull, "w") as fnull:
try:
out = check_output([self.ARCHIVER, self.CMDS['list'],
self._arch], stderr=fnull)
except CalledProcessError:
return contents
for line in out.split(b"\n"):
match = self.LINE_PAT.match(line)
if not match:
continue
match_entry = match.groupdict()
entry = {}
for key in match_entry:
entry[bytes(key, 'utf-8')] = match_entry[key]
del match_entry
entry[b'perms'] = b"-rw-r--r--"
if not entry[b'size']:
entry[b'perms'] = b"drwxr-xr-x"
entry[b'size'] = b"0"
entry[b'display_name'] = self._map_name(entry[b'fpath'])
entry[b'datetime'] = self._parse_dt(entry[b'date'], entry[b'time'])
entry[b'uid'] = bytes(str(self._uid), 'utf-8')
entry[b'gid'] = bytes(str(self._gid), 'utf-8')
contents.append(entry)
return contents
def list(self):
"""
Output list contents of adf image.
Convert filenames to be Unix filesystem friendly
Add suffix to show user what kind of file do he dealing with.
"""
if not self._contents:
sys.stderr.write("Nodos or archive error\n")
return 1
for entry in self._contents:
sys.stdout.buffer.write(self.ITEM % entry)
return 0
def copyout(self, src, dst):
"""Copy file form the adf image."""
real_src = self._get_real_name(src)
if not real_src:
raise IOError("No such file or directory")
if b" " in real_src:
sys.stderr.write("unadf is unable to operate on filepath with "
"space inside.\nUse affs to mount image and than"
" extract desired files.\n")
return 1
extract_dir = mkdtemp()
cmd = [self.ARCHIVER, self._arch, real_src, b"-d", extract_dir]
if check_call(cmd, stdout=open(os.devnull, 'wb'),
stderr=open(os.devnull, 'wb')) != 0:
shutil.rmtree(extract_dir)
sys.stderr.write("unadf returned with nonzero exit code\n")
return 1
shutil.move(os.path.join(bytes(extract_dir, "utf8"), real_src),
bytes(dst, "utf8"))
shutil.rmtree(extract_dir)
return 0
if __name__ == "__main__":
sys.exit(parse_args(UAdf))

182
ulha
View File

@@ -1,182 +0,0 @@
#!/usr/bin/env python3
"""
Lha Virtual filesystem executive for Midnight Commander.
Tested against python 3.6, lha[1] 1.14 and mc 4.8.22
[1] http://lha.sourceforge.jp
Changelog:
1.3 Switch to python3
1.2 Moved item pattern to extfslib module
1.1 Moved common code into extfslib library
1.0 Initial release
Author: Roman 'gryf' Dobosz <gryf73@gmail.com>
Date: 2019-06-30
Version: 1.3
Licence: BSD
"""
import os
import sys
import re
import shutil
from subprocess import call, check_call, CalledProcessError
from tempfile import mkdtemp, mkstemp
from extfslib import Archive, parse_args
class ULha(Archive):
"""Archive handle. Provides interface to MC's extfs subsystem"""
LINE_PAT = re.compile(b"^((?P<perms>[d-][rswx-]{9})|(\[generic\])|"
b"(\[unknown\]))"
b"((\s+\d+/\d+\s+)|(\s+))"
b"(?P<uid>)(?P<gid>)" # just for the record
b"(?P<size>\d+)"
b"\s+(\*{6}|\d+\.\d%)"
b"\s(?P<month>[JFMASOND][a-z]{2})\s+" # month
b"(?P<day>\d+)\s+" # day
b"(?P<yh>\d{4}|(\d{2}:\d{2}))" # year/hour
b"\s(?P<fpath>.*)")
ARCHIVER = b"lha"
CMDS = {"list": b"lq",
"read": b"pq",
"write": b"aq",
"delete": b"dq"}
DATETIME = b"%(month)s %(day)s %(yh)s"
def _get_dir(self):
"""Prepare archive file listing"""
contents = []
out = self._call_command("list")
if not out:
return
for line in out.split(b"\n"):
# -lhd- can store empty directories
perms = b"-rw-r--r--"
if line.endswith(bytes(os.path.sep, 'utf-8')):
line = line[:-1]
perms = b"drw-r--r--"
match = self.LINE_PAT.match(line)
if not match:
continue
match_entry = match.groupdict()
entry = {}
for key in match_entry:
entry[bytes(key, 'utf-8')] = match_entry[key]
del match_entry
# UID and GID sometimes can have strange values depending on
# the information that was written into archive. Most of the
# times I was dealing with Amiga lha archives, so that i don't
# really care about real user/group
entry[b'uid'] = bytes(str(self._uid), 'utf-8')
entry[b'gid'] = bytes(str(self._gid), 'utf-8')
entry[b'datetime'] = self.DATETIME % entry
if not entry[b'perms']:
entry[b'perms'] = perms
entry[b'display_name'] = self._map_name(entry[b'fpath'])
contents.append(entry)
return contents
def list(self):
"""Output contents of the archive to stdout"""
for entry in self._contents:
sys.stdout.buffer.write(self.ITEM % entry)
return 0
def rm(self, dst):
"""Remove file from archive"""
dst = self._get_real_name(dst)
# deleting with quiet option enabled will output nothing, so we get
# empty string here or None in case of error. Not smart.
if self._call_command('delete', dst=dst) is None:
return 1
return 0
def rmdir(self, dst):
"""Remove empty directory"""
dst = self._get_real_name(dst)
if not dst.endswith(bytes(os.path.sep, 'utf-8')):
dst += bytes(os.path.sep, 'utf-8')
if self._call_command('delete', dst=dst) is None:
return 1
return 0
def run(self, dst):
"""Execute file out of archive"""
fdesc, tmp_file = mkstemp()
os.close(fdesc)
result = 0
if self.copyout(dst, tmp_file) != 0:
result = 1
os.chmod(tmp_file, int("700", 8))
try:
result = call([tmp_file])
finally:
try:
os.unlink(tmp_file)
except OSError:
pass
return result
def mkdir(self, dst):
"""Create empty directory in archive"""
return self.copyin(dst)
def copyin(self, dst, src=None):
"""Copy file to the archive or create direcotry inside.
If src is empty, create empty directory with dst name."""
current_dir = os.path.abspath(os.curdir)
tmpdir = mkdtemp()
arch_abspath = os.path.realpath(self._arch)
os.chdir(tmpdir)
if src:
os.makedirs(os.path.dirname(dst))
shutil.copy2(src, dst)
else:
os.makedirs(dst)
try:
result = check_call([self.ARCHIVER.decode('utf-8'),
self.CMDS["write"].decode('utf-8'),
arch_abspath, dst])
except CalledProcessError:
return 1
finally:
os.chdir(current_dir)
shutil.rmtree(tmpdir)
return result
def copyout(self, src, dst):
"""Copy file out form archive."""
src = self._get_real_name(src)
fobj = open(dst, "wb")
try:
result = check_call([self.ARCHIVER, self.CMDS['read'], self._arch,
src], stdout=fobj)
except CalledProcessError:
return 1
finally:
fobj.close()
return result
if __name__ == "__main__":
sys.exit(parse_args(ULha))