dtrx: comparison scripts/x

-:36f352abd093
+:77043f4e6a9f
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, 5th Floor, Boston, MA, 02111.
 import errno
 import mimetypes
+import optparse
 import os
 import subprocess
 import sys
 import tempfile
 from cStringIO import StringIO
+VERSION = "1.1"
+VERSION_BANNER = """x version %s
+Copyright (c) 2006 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 2 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.""" % (VERSION,)
+MATCHING_DIRECTORY = 1
+# ONE_DIRECTORY = 2
+BOMB = 3
+EMPTY = 4
 mimetypes.encodings_map.setdefault('.bz2', 'bzip2')
 mimetypes.types_map['.exe'] = 'application/x-msdos-program'
-MATCHING_DIRECTORY = 1
-ONE_DIRECTORY = 2
-BOMB = 3
-EMPTY = 4
 class ExtractorError(Exception):
 pass
 class BaseExtractor(object):
 decoders = {'bzip2': 'bzcat', 'gzip': 'zcat', 'compress': 'zcat'}
 def __init__(self, filename, mimetype, encoding):
+if encoding and (not self.decoders.has_key(encoding)):
+raise ValueError("unrecognized encoding %s" % (encoding,))
 self.filename = filename
 self.mimetype = mimetype
 self.encoding = encoding
+self.included_archives = []
 try:
 self.archive = open(filename, 'r')
 except (IOError, OSError), error:
 raise ExtractorError("could not open %s: %s" %
 (filename, error.strerror))
 def prepare(self):
 pass
 def check_contents(self):
 self.archive.seek(0, 0)
+archive_type = None
 filenames = self.get_filenames()
 try:
-first_part = filenames.next().split('/', 1)[0] + '/'
+filename = filenames.next()
+if extractor_map.has_key(mimetypes.guess_type(filename)[0]):
+self.included_archives.append(filename)
+first_part = filename.split('/', 1)[0] + '/'
 except StopIteration:
 filenames.stop()
 return EMPTY
 for filename in filenames:
-if not filename.startswith(first_part):
+if extractor_map.has_key(mimetypes.guess_type(filename)[0]):
-filenames.stop()
+self.included_archives.append(filename)
-return BOMB
+if (archive_type is None) and (not filename.startswith(first_part)):
+archive_type = BOMB
 filenames.stop()
+if archive_type:
+return archive_type
 if self.basename() == first_part[:-1]:
 return MATCHING_DIRECTORY
 return first_part
 def basename(self):
 class ZipExtractor(BaseExtractor):
 def __init__(self, filename, mimetype, encoding):
 self.filename = filename
 self.mimetype = mimetype
 self.encoding = encoding
+self.included_archives = []
 self.archive = StringIO()
 def get_filenames(self):
 return ProcessStreamer(['zipinfo', '-1', self.filename], None)
 elif len(pieces[-1]) < 6:
 pieces.pop()
 return '.'.join(pieces)
 def check_contents(self):
+CpioExtractor.check_contents(self)
 return BOMB
 class DebExtractor(TarExtractor):
 def prepare(self):
 self.pipe(['ar', 'p', self.filename, 'data.tar.gz'],
 "data.tar.gz extraction")
 elif (len(pieces[-1]) > 10) or (not pieces[-1].endswith('.deb')):
 return BaseExtractor.basename(self)
 return pieces[0]
 def check_contents(self):
+TarExtractor.check_contents(self)
 return BOMB
+extractor_map = {'application/x-tar': TarExtractor,
+'application/zip': ZipExtractor,
+'application/x-msdos-program': ZipExtractor,
+'application/x-debian-package': DebExtractor,
+'application/x-redhat-package-manager': RPMExtractor,
+'application/x-rpm': RPMExtractor,
+'application/x-cpio': CpioExtractor}
 class ExtractorApplication(object):
-extractor_map = {'application/x-tar': TarExtractor,
+actions = ['get_extractor', 'prepare_extraction', 'extract', 'recurse']
-'application/zip': ZipExtractor,
-'application/x-msdos-program': ZipExtractor,
-'application/x-debian-package': DebExtractor,
-'application/x-redhat-package-manager': RPMExtractor,
-'application/x-rpm': RPMExtractor,
-'application/x-cpio': CpioExtractor}
-actions = ['get_extractor', 'prepare_extraction', 'extract']
 def __init__(self, arguments):
-self.filenames = arguments
+self.parse_options(arguments)
 self.successes = []
 self.failures = []
+def parse_options(self, arguments):
+parser = optparse.OptionParser(
+usage="%prog [options] archive [archive2 ...]",
+description="Intelligent archive extractor",
+version=VERSION_BANNER
+)
+parser.add_option('-r', '--recursive', dest='recursive',
+action='store_true', default=False,
+help='extract archives contained in the ones listed')
+self.options, filenames = parser.parse_args(arguments)
+if not filenames:
+parser.error("you did not list any archives")
+self.archives = {os.path.realpath(os.curdir): filenames}
 def show_error(self, message):
 print >>sys.stderr, "%s: %s" % (self.current_filename, message)
 def get_extractor(self):
 mimetype, encoding = mimetypes.guess_type(self.current_filename)
 try:
-handler = self.extractor_map[mimetype]
+handler = extractor_map[mimetype]
 except KeyError:
 self.show_error("not a known archive type")
 return False
 try:
 self.current_extractor = handler(self.current_filename, mimetype,
 def prepare_extraction(self):
 self.current_path = '.'
 contents = self.current_extractor.check_contents()
 if contents not in (MATCHING_DIRECTORY, EMPTY):
-directory = self.prepare_target_directory()
+self.target_directory = self.prepare_target_directory()
-if directory is None:
+if self.target_directory is None:
 return False
 if contents == BOMB:
-os.chdir(directory)
+os.chdir(self.target_directory)
 self.current_path = '..'
-self.cleanup_actions.append((os.chdir, '..'))
 else:
-self.cleanup_actions.append((os.rename, contents, directory))
+self.cleanup_actions.append((os.rename, contents,
+self.target_directory))
+else:
+self.target_directory = os.curdir
 return True
 def extract(self):
 try:
 self.current_extractor.extract(self.current_path)
 except ExtractorError, error:
 self.show_error(error)
 return False
 return True
+def recurse(self):
+if not self.options.recursive:
+return True
+print "wow", self.current_extractor.included_archives
+for filename in self.current_extractor.included_archives:
+tail_path, basename = os.path.split(filename)
+directory = os.path.join(self.current_directory,
+self.target_directory, tail_path)
+self.archives.setdefault(directory, []).append(basename)
+print self.archives
+return True
 def run(self):
-for filename in self.filenames:
+while self.archives:
-running = True
+self.current_directory, filenames = self.archives.popitem()
-self.current_filename = filename
+for filename in filenames:
-self.cleanup_actions = []
+os.chdir(self.current_directory)
-actions = [getattr(self, name) for name in self.actions]
+running = True
-while running and actions:
+self.current_filename = filename
-running = actions.pop(0)()
+self.cleanup_actions = []
-for action in self.cleanup_actions:
+actions = [getattr(self, name) for name in self.actions]
-action[0](*action[1:])
+while running and actions:
-if running:
+running = actions.pop(0)()
-self.successes.append(self.current_filename)
+for action in self.cleanup_actions:
-else:
+action[0](*action[1:])
-self.failures.append(self.current_filename)
+if running:
+self.successes.append(self.current_filename)
+else:
+self.failures.append(self.current_filename)
 if self.failures:
 return 1
 return 0

Mercurial > dtrx / file comparison

comparison: scripts/x

scripts/x