dtrx: comparison scripts/dtrx

-:69cd94403088
+:37d555407334
 import logging
 import mimetypes
 import optparse
 import os
 import re
+import shutil
 import stat
 import subprocess
 import sys
 import tempfile
 import textwrap
 import traceback
 from sets import Set
-VERSION = "5.0"
+VERSION = "6.0"
 VERSION_BANNER = """dtrx version %s
-Copyright (c) 2006, 2007 Brett Smith <brettcsmith@brettcsmith.org>
+Copyright (c) 2006, 2007, 2008 Brett Smith <brettcsmith@brettcsmith.org>
 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.
 class ExtractorError(Exception):
 pass
+class ExtractorUnusable(Exception):
+pass
+EXTRACTION_ERRORS = (ExtractorError, ExtractorUnusable, OSError, IOError)
 class BaseExtractor(object):
 decoders = {'bzip2': 'bzcat', 'gzip': 'zcat', 'compress': 'zcat',
 'lzma': 'lzcat'}
 stdin = processes[-1].stdout
 if index == last_pipe:
 stdout = final_stdout
 else:
 stdout = subprocess.PIPE
-processes.append(subprocess.Popen(command, stdin=stdin,
+try:
-stdout=stdout,
+processes.append(subprocess.Popen(command, stdin=stdin,
-stderr=subprocess.PIPE))
+stdout=stdout,
+stderr=subprocess.PIPE))
+except OSError, error:
+if error.errno == errno.ENOENT:
+raise ExtractorUnusable("could not run %s" % (command[0],))
+raise
 exit_codes = [pipe.wait() for pipe in processes]
 self.archive.close()
 for index in range(last_pipe):
 processes[index].stdout.close()
 processes[index].stderr.close()
 os.chdir(self.target)
 try:
 self.archive.seek(0, 0)
 self.extract_archive()
 self.check_contents()
-except ExtractorError:
+except EXTRACTION_ERRORS:
 os.chdir(old_path)
-subprocess.call(['rm', '-rf', self.target])
+shutil.rmtree(self.target, ignore_errors=True)
 raise
 os.chdir(old_path)
 def get_filenames(self):
 self.run_pipes()
 return
 yield line.rstrip('\n')
 class CompressionExtractor(BaseExtractor):
+file_type = 'compressed file'
 name_checker = FilenameChecker
 def basename(self):
 pieces = os.path.basename(self.filename).split('.')
 extension = '.' + pieces[-1]
 output_fd, self.target = tempfile.mkstemp(prefix='.dtrx-', dir='.')
 except (OSError, IOError), error:
 raise ExtractorError("cannot extract here: %s" % (error.strerror,))
 try:
 self.run_pipes(output_fd)
-except ExtractorError:
+except EXTRACTION_ERRORS:
 os.close(output_fd)
 os.unlink(self.target)
 raise
 os.close(output_fd)
 class TarExtractor(BaseExtractor):
+file_type = 'tar file'
 def get_filenames(self):
 self.pipe(['tar', '-t'], "listing")
 return BaseExtractor.get_filenames(self)
 def extract_archive(self):
 self.pipe(['tar', '-x'])
 self.run_pipes()
 class CpioExtractor(BaseExtractor):
+file_type = 'cpio file'
 def get_filenames(self):
 self.pipe(['cpio', '-t'], "listing")
 return BaseExtractor.get_filenames(self)
 def extract_archive(self):
 '--no-absolute-filenames'])
 self.run_pipes()
 class RPMExtractor(CpioExtractor):
+file_type = 'RPM'
 def prepare(self):
 self.pipe(['rpm2cpio', '-'], "rpm2cpio")
 def basename(self):
 pieces = os.path.basename(self.filename).split('.')
 self.check_included_archives(os.listdir('.'))
 self.content_type = BOMB
 class DebExtractor(TarExtractor):
+file_type = 'Debian package'
 def prepare(self):
 self.pipe(['ar', 'p', self.filename, 'data.tar.gz'],
 "data.tar.gz extraction")
 self.pipe(['zcat'], "data.tar.gz decompression")
 "control.tar.gz extraction")
 self.pipe(['zcat'], "control.tar.gz decompression")
 class GemExtractor(TarExtractor):
+file_type = 'Ruby gem'
 def prepare(self):
 self.pipe(['tar', '-xO', 'data.tar.gz'], "data.tar.gz extraction")
 self.pipe(['zcat'], "data.tar.gz decompression")
 def check_contents(self):
 self.check_included_archives(os.listdir('.'))
 self.content_type = BOMB
 class GemMetadataExtractor(CompressionExtractor):
+file_type = 'Ruby gem'
 def prepare(self):
 self.pipe(['tar', '-xO', 'metadata.gz'], "metadata.gz extraction")
 self.pipe(['zcat'], "metadata.gz decompression")
 def basename(self):
 BaseExtractor.__init__(self, '/dev/null', None)
 self.filename = os.path.realpath(filename)
 class ZipExtractor(NoPipeExtractor):
+file_type = 'Zip file'
 def get_filenames(self):
 self.pipe(['zipinfo', '-1', self.filename], "listing")
 return BaseExtractor.get_filenames(self)
 def extract_archive(self):
 self.pipe(['unzip', '-q', self.filename])
 self.run_pipes()
 class SevenExtractor(NoPipeExtractor):
+file_type = '7z file'
 border_re = re.compile('^[- ]+$')
 def get_filenames(self):
 self.pipe(['7z', 'l', self.filename], "listing")
 self.run_pipes()
 self.pipe(['7z', 'x', self.filename])
 self.run_pipes()
 class CABExtractor(NoPipeExtractor):
+file_type = 'CAB archive'
 border_re = re.compile(r'^[-\+]+$')
 def get_filenames(self):
 self.pipe(['cabextract', '-l', self.filename], "listing")
 self.run_pipes()
 self.target = None
 def report(self, function, *args):
 try:
 error = function(*args)
-except (ExtractorError, IOError, OSError), exception:
+except EXTRACTION_ERRORS, exception:
 error = str(exception)
 logger.debug(''.join(traceback.format_exception(*sys.exc_info())))
 return error
 class ListAction(BaseAction):
 def __init__(self, options, filenames):
 BaseAction.__init__(self, options, filenames)
+self.count = 0
 def get_list(self, extractor):
 # Note: The reason I'm getting all the filenames up front is
 # because if we run into trouble partway through the archive, we'll
 # try another extractor.  So before we display anything we have to
 # be sure this one is successful.  We maybe don't have to be quite
 # this conservative but this is the easy way out for now.
 self.filelist = list(extractor.get_filenames())
 def show_list(self, filename):
+self.count += 1
 if len(self.filenames) != 1:
-if filename != self.filenames[0]:
+if self.count > 1:
 print
 print "%s:" % (filename,)
 print '\n'.join(self.filelist)
 def run(self, filename, extractor):
-self.current_filename = filename
 return (self.report(self.get_list, extractor) or
 self.report(self.show_list, filename))
 class ExtractorApplication(object):
 return error.strerror
 if stat.S_ISDIR(result.st_mode):
 return "cannot extract a directory"
 def try_extractors(self, filename, builder):
-last_error = "could not find a way to extract this"
+errors = []
-while True:
+for extractor in builder:
-try:
-extractor = builder.next()
-except StopIteration:
-return last_error
-except (IOError, OSError, ExtractorError), error:
-return str(error)
 error = self.action.run(filename, extractor)
 if error:
-logger.info("%s: %s" % (filename, error))
+errors.append((extractor.file_type, extractor.encoding, error))
-last_error = error
 else:
 self.recurse(filename, extractor, self.action)
 return
+logger.error("could not handle %s" % (filename,))
+if not errors:
+logger.error("not a known archive type")
+return True
+for file_type, encoding, error in errors:
+message = ["treating as", file_type, "failed:", error]
+if encoding:
+message.insert(1, "%s-encoded" % (encoding,))
+logger.error(' '.join(message))
+return True
 def run(self):
 if self.options.show_list:
 action = ListAction
 else:
 for filename in self.filenames:
 builder = ExtractorBuilder(filename, self.options)
 error = (self.check_file(filename) or
 self.try_extractors(filename, builder.get_extractor()))
 if error:
-logger.error("%s: %s" % (filename, error))
+if error != True:
+logger.error("%s: %s" % (filename, error))
 self.failures.append(filename)
 else:
 self.successes.append(filename)
 self.options.one_entry_policy.permanent_policy = EXTRACT_WRAP
 if self.failures:

Mercurial > dtrx / file comparison

comparison: scripts/dtrx

scripts/dtrx