mc_ulha/ulha

#!/usr/bin/env python3
"""
Lha Virtual filesystem executive for Midnight Commander.

Tested against python 3.8, lha[1] 1.14 and mc 4.8.29

[1] http://lha.sourceforge.jp

Changelog:
    1.4 Cleanup code, use latin1
    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: 2023-10-22
Version: 1.4
Licence: BSD
"""
import os
import re
import shutil
import subprocess
import sys
import tempfile

import extfslib


class ULha(extfslib.Archive):
    """Archive handle. Provides interface to MC's extfs subsystem"""

    LINE_PAT = re.compile(r"^((?P<perms>[d-][rswx-]{9})|(\[generic\])|"
                          r"(\[unknown\]))"
                          r"((\s+\d+/\d+\s+)|(\s+))"
                          r"(?P<uid>)(?P<gid>)"  # just for the record
                          r"(?P<size>\d+)"
                          r"\s+(\*{6}|\d+\.\d%)"
                          r"\s(?P<month>[JFMASOND][a-z]{2})\s+"  # month
                          r"(?P<day>\d+)\s+"  # day
                          r"(?P<yh>\d{4}|(\d{2}:\d{2}))"  # year/hour
                          r"\s(?P<fpath>.*)")
    ARCHIVER = "lha"
    CMDS = {"list": "lq",
            "read": "pq",
            "write": "aq",
            "delete": "dq"}
    DATETIME = "%(month)s %(day)s %(yh)s"

    def _get_dir(self):
        """Prepare archive file listing"""
        contents = []

        out = subprocess.run([self.ARCHIVER, self.CMDS['list'], self._arch],
                             capture_output=True, encoding="Latin1")
        if out.returncode != 0:
            sys.stderr.write(out.stderr)
            return

        for line in out.stdout.split("\n"):
            # -lhd- can store empty directories
            perms = "-rw-r--r--"
            if line.endswith(os.path.sep):
                line = line[:-1]
                perms = "drw-r--r--"

            match = self.LINE_PAT.match(line)
            if not match:
                continue

            match_entry = match.groupdict()
            entry = {}
            for key in match_entry:
                entry[key] = 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['uid'] = str(self._uid)
            entry['gid'] = str(self._gid)
            entry['datetime'] = self.DATETIME % entry

            if not entry['perms']:
                entry['perms'] = perms

            entry['display_name'] = self._map_name(entry['fpath'])
            contents.append(entry)

        return contents

    def list(self):
        """Output contents of the archive to stdout"""
        for entry in self._contents:
            sys.stdout.write(self.ITEM.decode('utf-8') % 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(os.path.sep):
            dst += os.path.sep

        res = subprocess.run([self.ARCHIVER, self.CMDS['delete'], dst],
                             capture_output=True)
        return res.returncode

    def run(self, dst):
        """Execute file out of archive"""
        fdesc, tmp_file = tempfile.mkstemp()
        os.close(fdesc)
        result = 0

        if self.copyout(dst, tmp_file) != 0:
            result = 1

        os.chmod(tmp_file, int("700", 8))

        try:
            result = subprocess.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 = tempfile.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)

        res = subprocess.run([self.ARCHIVER, self.CMDS["write"], arch_abspath,
                              dst], capture_output=True, encoding='utf-8')
        os.chdir(current_dir)
        shutil.rmtree(tmpdir)
        return res.returncode

    def copyout(self, src, dst):
        """Copy file out form archive."""
        src = self._get_real_name(src)
        with open(dst, "wb") as fobj:
            res = subprocess.run([self.ARCHIVER, self.CMDS['read'], self._arch,
                                  src], stdout=fobj)
        return res.returncode


if __name__ == "__main__":
    sys.exit(extfslib.parse_args(ULha))