patool/patoolib/__init__.py

419 lines
14 KiB
Python

# -*- coding: utf-8 -*-
# Copyright (C) 2010 Bastian Kleineidam
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import shutil
import stat
from patoolib import util
# Supported archive commands
ArchiveCommands = ('list', 'extract', 'test', 'create')
# Supported archive formats
ArchiveFormats = ('gzip', 'bzip2', 'tar', 'zip', 'compress', '7z', 'rar',
'cab', 'arj', 'cpio', 'rpm', 'deb', 'lzop', 'lzma', 'xz', 'lzip', 'ace',
'ar', 'lzh')
# Supported encodings (used with tar for example)
# Note that all encodings must also be archive formats
ArchiveEncodings = ('gzip', 'bzip2', 'compress', 'lzma', 'xz', 'lzip')
# Map MIME types to archive format
ArchiveMimetypes = {
'application/x-bzip2': 'bzip2',
'application/x-tar': 'tar',
'application/x-gzip': 'gzip',
'application/zip': 'zip',
'application/java-archive': 'zip',
'application/x-7z-compressed': '7z',
'application/x-compress': 'compress',
'application/x-rar': 'rar',
'application/rar': 'rar',
'application/x-cab': 'cab',
'application/vnd.ms-cab-compressed': 'cab',
'application/x-arj': 'arj',
'application/x-cpio': 'cpio',
'application/x-redhat-package-manager': 'rpm',
'application/x-rpm': 'rpm',
'application/x-debian-package': 'deb',
'application/x-lzop': 'lzop',
'application/x-lzma': 'lzma',
'application/x-xz': 'xz',
'application/x-lzip': 'lzip',
'application/x-ace': 'ace',
'application/x-archive': 'ar',
'application/x-lha': 'lzh',
'application/x-lzh': 'lzh',
}
# List of programs supporting the given encoding
EncodingPrograms = {
'gzip': ('gzip',),
'bzip2': ('pbzip2', 'bzip2'),
'compress': ('compress',),
'lzma': ('lzma',),
'xz': ('xz',),
'lzip': ('lzip',),
}
# List of programs supporting the given archive format and command.
# If command is None, the program supports all commands (list, extract, ...)
ArchivePrograms = {
'ace': {
'extract': ('unace',),
'test': ('unace',),
'list': ('unace',),
},
'ar': {
None: ('ar',),
},
'bzip2': {
'extract': ('pbzip2', 'bzip2', '7z'),
'test': ('pbzip2', 'bzip2', '7z'),
'create': ('pbzip2', 'bzip2', '7z'),
'list': ('7z', 'echo',),
},
'tar': {
None: ('tar', 'star',),
},
'zip': {
'extract': ('unzip', '7z'),
'list': ('unzip', '7z'),
'test': ('unzip', '7z'),
'create': ('zip',),
},
'gzip': {
None: ('gzip', '7z'),
},
'lzh': {
None: ('lha',),
},
'lzip': {
'extract': ('lzip',),
'list': ('echo',),
'test': ('lzip',),
'create': ('lzip',),
},
'compress': {
'extract': ('gzip', '7z', 'uncompress.real'),
'list': ('7z', 'echo',),
'test': ('gzip', '7z'),
'create': ('compress',),
},
'7z': {
None: ('7z',),
},
'rar': {
None: ('rar',),
'extract': ('unrar', '7z'),
'list': ('unrar', '7z'),
'test': ('unrar', '7z'),
},
'cab': {
'extract': ('cabextract', '7z'),
'list': ('cabextract', '7z'),
'test': ('cabextract', '7z'),
},
'arj': {
None: ('arj',),
'extract': ('7z',),
'list': ('7z',),
'test': ('7z',),
},
'cpio': {
'extract': ('cpio', '7z'),
'list': ('cpio', '7z'),
'test': ('7z',),
'create': ('cpio',),
},
'rpm': {
# XXX rpm2cpio depends on cpio whose availability is not checked
'extract': ('rpm2cpio', '7z'),
'list': ('rpm', '7z'),
'test': ('rpm', '7z'),
},
'deb': {
'extract': ('dpkg-deb', '7z'),
'list': ('dpkg-deb', '7z'),
'test': ('dpkg-deb', '7z'),
},
'lzop': {
None: ('lzop',),
},
'lzma': {
'extract': ('lzma',),
'list': ('echo',),
'test': ('lzma',),
'create': ('lzma',),
},
'xz': {
'extract': ('xz',),
'list': ('echo',),
'test': ('xz',),
'create': ('xz',),
},
}
# only list those programs that have different python module names
ProgramModules = {
'7z': 'p7zip',
'uncompress.real': 'uncompress',
'dpkg-deb': 'dpkg',
}
def get_archive_format (filename):
"""Detect filename archive format and optional encoding."""
mime, encoding = util.guess_mime(filename)
if not (mime or encoding):
raise util.PatoolError("unknown archive format for file `%s'" % filename)
if mime in ArchiveMimetypes:
format = ArchiveMimetypes[mime]
else:
raise util.PatoolError("unknown archive format for file `%s' (mime-type is `%s')" % (filename, mime))
if format == encoding:
# file cannot be in same format encoded
encoding = None
return format, encoding
def check_archive_format (format, encoding):
"""Make sure format and encoding is known."""
if format not in ArchiveFormats:
raise util.PatoolError("unknown archive format `%s'" % format)
if encoding is not None and encoding not in ArchiveEncodings:
raise util.PatoolError("unkonwn archive encoding `%s'" % encoding)
def check_archive_command (command):
"""Make sure archive command is valid."""
if command not in ArchiveCommands:
raise util.PatoolError("invalid archive command `%s'" % command)
def find_archive_program (format, command):
"""Find suitable archive program for given format and mode."""
commands = ArchivePrograms[format]
programs = []
# first try the universal programs with key None
for key in (None, command):
if key in commands:
programs.extend(commands[key])
if not programs:
raise util.PatoolError("%s archive format `%s' is not supported" % (command, format))
# return the first existing program
for program in programs:
exe = util.find_program(program)
if exe:
if program == '7z' and format == 'rar' and not util.p7zip_supports_rar():
continue
return exe
# no programs found
raise util.PatoolError("could not find an executable program to %s format %s; candidates are (%s)," % (command, format, ",".join(programs)))
def find_encoding_program (program, encoding):
"""Find suitable encoding program and return it. Returns None if
no encoding program could be found"""
if program in ('tar', 'star'):
for enc_program in EncodingPrograms[encoding]:
found = util.find_program(enc_program)
if found:
return found
return None
def list_formats ():
for format in ArchiveFormats:
print format, "files:"
for command in ArchiveCommands:
programs = ArchivePrograms[format]
if command not in programs and None not in programs:
print " %8s: - (not supported)" % command
continue
try:
program = find_archive_program(format, command)
print " %8s: %s" % (command, program),
basename = os.path.basename(program)
if format == 'tar':
encs = [x for x in ArchiveEncodings if util.find_program(x)]
if encs:
print "(supported encodings: %s)" % ", ".join(encs),
elif format == '7z':
if util.p7zip_supports_rar():
print "(rar archives supported)",
else:
print "(rar archives not supported)",
print
except util.PatoolError:
handlers = programs.get(None, programs.get(command))
print " %8s: - (no program found; install %s)" % \
(command, util.strlist_with_or(handlers))
return 0
AllowedConfigKeys = ("verbose", "program")
def clean_config_keys (kwargs):
"""Remove invalid configuration keys from arguments."""
config_kwargs = dict(kwargs)
for key in kwargs:
if key not in AllowedConfigKeys:
del config_kwargs[key]
return config_kwargs
def parse_config (archive, format, encoding, command, **kwargs):
"""The configuration determines which program to use for which
archive format for the given command.
@raises: PatoolError if command for given format and encoding
is not supported.
"""
config = {
'verbose': False,
}
config['program'] = find_archive_program(format, command)
for key, value in kwargs.items():
if value is not None:
if key == 'program':
program = util.find_program(value)
if program:
value = program
config[key] = value
program = os.path.basename(config['program'])
if encoding and not find_encoding_program(program, encoding):
raise util.PatoolError("cannot %s archive `%s': encoding `%s' not supported by %s" % (archive, command, encoding, program))
return config
def move_outdir_orphan (outdir):
"""Move a single file or directory inside outdir a level up.
Never overwrite files.
Return (True, outfile) if successful, (False, reason) if not."""
entries = os.listdir(outdir)
reason = ""
if len(entries) == 1:
src = os.path.join(outdir, entries[0])
dst = os.path.join(os.path.dirname(outdir), entries[0])
if os.path.exists(dst):
return (False, "local file exists")
shutil.move(src, dst)
os.rmdir(outdir)
return (True, entries[0])
return (False, "multiple files in root")
def run_archive_cmdlist (archive_cmdlist):
"""Run archive command."""
# archive_cmdlist is a command list with optional keyword arguments
if isinstance(archive_cmdlist, tuple):
cmdlist, runkwargs = archive_cmdlist
else:
cmdlist, runkwargs = archive_cmdlist, {}
util.run(cmdlist, **runkwargs)
def make_file_readable (filename):
"""Make file user readable if it is not a link."""
if not os.path.islink(filename):
util.set_mode(filename, stat.S_IRUSR)
def make_dir_readable (filename):
"""Make directory user readable and executable."""
util.set_mode(filename, stat.S_IRUSR|stat.S_IXUSR)
def make_user_readable (directory):
"""Make all files in given directory user readable. Also recurse into
subdirectories."""
for root, dirs, files in os.walk(directory, onerror=util.log_error):
for filename in files:
make_file_readable(os.path.join(root, filename))
for dirname in dirs:
make_dir_readable(os.path.join(root, dirname))
def cleanup_outdir (archive, outdir):
"""Cleanup outdir after extraction and return target file name."""
make_user_readable(outdir)
if outdir:
# move single directory or file in outdir
(res, msg) = move_outdir_orphan(outdir)
if res:
target = "`%s'" % msg
else:
target = "`%s' (%s)" % (outdir, msg)
else:
target = "`%s'" % util.stripext(archive)
return target
def _handle_archive (archive, command, *args, **kwargs):
"""Handle archive command; raising PatoolError on errors."""
if command != 'create':
# check that archive is a regular file
util.check_filename(archive)
format, encoding = kwargs.get("format"), kwargs.get("encoding")
if format is None:
format, encoding = get_archive_format(archive)
check_archive_format(format, encoding)
check_archive_command(command)
config_kwargs = clean_config_keys(kwargs)
config = parse_config(archive, format, encoding, command, **config_kwargs)
# check if archive already exists
if command == 'create' and os.path.exists(archive):
raise util.PatoolError("archive `%s' already exists" % archive)
program = config['program']
# get python module for given archive program
key = util.stripext(os.path.basename(program).lower())
module = ProgramModules.get(key, key)
# import archive handler (eg. patoolib.programs.star.extract_tar())
exec "from patoolib.programs.%s import %s_%s as func" % (module, command, format)
get_archive_cmdlist = locals()['func']
# prepare func() call arguments
kwargs = dict(verbose=config['verbose'])
outdir = None
if command == 'extract':
outdir = util.tmpdir(dir=os.getcwd())
kwargs['outdir'] = outdir
try:
cmdlist = get_archive_cmdlist(archive, encoding, program, *args, **kwargs)
run_archive_cmdlist(cmdlist)
if command == 'extract':
target = cleanup_outdir(archive, outdir)
print "%s: extracted to %s" % (archive, target)
finally:
if outdir:
try:
os.rmdir(outdir)
except OSError:
pass
def handle_archive (archive, command, *args, **kwargs):
"""Handle archive file command; with nice error reporting."""
try:
_handle_archive(archive, command, *args, **kwargs)
res = 0
except util.PatoolError, msg:
util.log_error(msg)
res = 1
except StandardError, msg:
util.log_internal_error()
res = 1
return res