dtrx: comparison scripts/x

-:0a3ef1b9f6d4
+:6f9e1bb59719
 import sys
 import tempfile
 from cStringIO import StringIO
-VERSION = "2.0"
+VERSION = "2.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
 MATCHING_DIRECTORY = 1
 # ONE_DIRECTORY = 2
 BOMB = 3
 EMPTY = 4
+DECOMPRESSED = 5
 mimetypes.encodings_map.setdefault('.bz2', 'bzip2')
 mimetypes.types_map['.exe'] = 'application/x-msdos-program'
+def run_command(command, description, stdout=None, stderr=None, stdin=None):
+process = subprocess.Popen(command, stdin=stdin, stdout=stdout,
+stderr=stderr)
+status = process.wait()
+for pipe in (process.stdout, process.stderr):
+try:
+pipe.close()
+except AttributeError:
+pass
+if status != 0:
+return ("%s error: '%s' returned status code %s" %
+(description, ' '.join(command), status))
+return None
+class FilenameChecker(object):
+def __init__(self, original_name):
+self.original_name = original_name
+def is_free(self, filename=None):
+if filename is None:
+filename = self.original_name
+return self._is_free(filename)
+def _is_free(self, filename):
+return not os.path.exists(filename)
+def check(self):
+for suffix in [''] + ['.%s' % (x,) for x in range(1, 10)]:
+filename = '%s%s' % (self.original_name, suffix)
+if self.is_free(filename):
+return filename
+raise ValueError("all alternatives for name %s taken" %
+(self.original_name,))
+class DirectoryChecker(FilenameChecker):
+def _is_free(self, filename):
+try:
+os.mkdir(filename)
+except OSError, error:
+if error.errno == errno.EEXIST:
+return False
+raise
+return True
 class ExtractorError(Exception):
 pass
 class BaseExtractor(object):
 decoders = {'bzip2': 'bzcat', 'gzip': 'zcat', 'compress': 'zcat'}
+name_checker = DirectoryChecker
 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.filename = os.path.realpath(filename)
 self.mimetype = mimetype
 self.encoding = encoding
 self.included_archives = []
 try:
 self.archive = open(filename, 'r')
 self.pipe([self.decoders[encoding]], "decoding")
 self.prepare()
 def run(self, command, description="extraction", stdout=None, stderr=None,
 stdin=None):
-process = subprocess.Popen(command, stdin=stdin, stdout=stdout,
+error = run_command(command, description, stdout, stderr, stdin)
-stderr=stderr)
+if error:
-status = process.wait()
+raise ExtractorError(error)
-if status != 0:
-raise ExtractorError("%s error: '%s' returned status code %s" %
-(description, ' '.join(command), status))
-for pipe in (process.stdout, process.stderr):
-try:
-pipe.close()
-except AttributeError:
-pass
 def pipe(self, command, description, stderr=None):
 output = tempfile.TemporaryFile()
 self.run(command, description, output, stderr, self.archive)
 self.archive.close()
 mimetypes.suffix_map.has_key(extension)):
 pieces.pop()
 return '.'.join(pieces)
 def extract(self, path):
+old_path = os.path.realpath(os.curdir)
+os.chdir(path)
 self.archive.seek(0, 0)
 self.extract_archive()
+os.chdir(old_path)
 class TarExtractor(BaseExtractor):
 def get_filenames(self):
 return ProcessStreamer(['tar', '-t'], self.archive)
 def extract_archive(self):
 self.run(['tar', '-x'], stdin=self.archive)
 class ZipExtractor(BaseExtractor):
 def __init__(self, filename, mimetype, encoding):
-self.filename = filename
+self.filename = os.path.realpath(filename)
 self.mimetype = mimetype
 self.encoding = encoding
 self.included_archives = []
 self.archive = StringIO()
 def get_filenames(self):
 return ProcessStreamer(['zipinfo', '-1', self.filename], None)
-def extract(self, path):
+def extract_archive(self):
-self.run(['unzip', '-q', os.path.join(path, self.filename)])
+self.run(['unzip', '-q', self.filename])
 class CpioExtractor(BaseExtractor):
 def get_filenames(self):
 return ProcessStreamer(['cpio', '-t'], self.archive,
 def check_contents(self):
 TarExtractor.check_contents(self)
 return BOMB
-class MatchHandler(object):
+class CompressionExtractor(BaseExtractor):
-def __init__(self, extractor, contents):
+name_checker = FilenameChecker
+def basename(self):
+pieces = os.path.basename(self.filename).split('.')
+extension = '.' + pieces[-1]
+if mimetypes.encodings_map.has_key(extension):
+pieces.pop()
+return '.'.join(pieces)
+def suggest_target(self):
+return FilenameChecker().check(self.basename())
+def check_contents(self):
+return self.basename()
+def extract(self, path):
+output = open(path, 'w')
+self.run(['cat'], "output write", stdin=self.archive, stdout=output)
+output.close()
+class BaseHandler(object):
+def __init__(self, extractor, contents, options):
 self.logger = logging.getLogger('x-log')
 self.extractor = extractor
 self.contents = contents
-self.directory = extractor.basename()
+self.options = options
-def extract(self, directory='.'):
+def extract(self):
+checker = self.extractor.name_checker(self.extractor.basename())
+if self.options.overwrite or checker.is_free():
+self.target = self.extractor.basename()
+self.overwrite()
+else:
+self.target = checker.check()
+self.safe_extract()
+def do_extract(self, directory):
 try:
 self.extractor.extract(directory)
 except ExtractorError, error:
 return error.strerror
 def cleanup(self):
-command = 'chmod'
+command = 'find'
-status = subprocess.call(['chmod', '-R', 'u+rw', self.directory])
+status = subprocess.call(['find', self.target, '-type', 'd',
+'-exec', 'chmod', 'u+rwx', '{}', ';'])
 if status == 0:
-command = 'find'
+command = 'chmod'
-status = subprocess.call(['find', self.directory, '-type', 'd',
+status = subprocess.call(['chmod', '-R', 'u+rw', self.target])
-'-exec', 'chmod', 'u+x', '{}', ';'])
 if status != 0:
 return "%s returned with exit status %s" % (command, status)
-class BombHandler(MatchHandler):
+class MatchHandler(BaseHandler):
-def __init__(self, extractor, contents):
+def overwrite(self):
-MatchHandler.__init__(self, extractor, contents)
+return self.do_extract('.')
-basename = self.directory
-for suffix in [''] + ['.%s' % (x,) for x in range(1, 10)]:
+def safe_extract(self):
-self.directory = '%s%s' % (basename, suffix)
+tempdir = tempfile.mkdtemp()
-try:
+result = self.do_extract(tempdir)
-os.mkdir(self.directory)
+if result is None:
-except OSError, error:
+os.rename(os.path.join(tempdir, self.extractor.basename()),
-if error.errno == errno.EEXIST:
+self.target)
-continue
+os.rmdir(tempdir)
-raise ValueError("could not make extraction directory %s: %s" %
+return result
-(error.filename, error.strerror))
-if suffix != '':
-self.logger.warning("%s: extracted to %s",
+class BombHandler(BaseHandler):
-extractor.filename, self.directory)
+def safe_extract(self):
-break
+return self.do_extract(self.target)
-else:
-raise ValueError("all good names for an extraction directory taken")
+def overwrite(self):
+self.target = self.extractor.basename()
-def extract(self):
+return self.do_extract(self.target)
-os.chdir(self.directory)
-return MatchHandler.extract(self, '..')
-def cleanup(self):
-os.chdir('..')
-return MatchHandler.cleanup(self)
 class EmptyHandler(object):
-def __init__(self, extractor, contents): pass
+def __init__(self, extractor, contents, options): pass
 def extract(self): pass
 def cleanup(self): pass
 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}
 handler_map = {EMPTY: EmptyHandler,
-MATCHING_DIRECTORY: MatchHandler}
+MATCHING_DIRECTORY: MatchHandler,
+DECOMPRESSED: BombHandler}
 class ExtractorApplication(object):
 def __init__(self, arguments):
 self.parse_options(arguments)
 self.setup_logger()
 action='count', default=3,
 help='suppress warning/error messages')
 parser.add_option('-v', '--verbose', dest='verbose',
 action='count', default=0,
 help='be verbose/print debugging information')
+parser.add_option('-o', '--overwrite', dest='overwrite',
+action='store_true', default=False,
+help='overwrite any existing target directory')
+##         parser.add_option('-f', '--flat', '--no-directory', dest='flat',
+##                           action='store_true', default=False,
+##                           help="don't put contents in their own directory")
+##         parser.add_option('-n', '--noninteractive', dest='batch',
+##                           action='store_true', default=False,
+##                           help="don't ask how to handle special cases")
 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 get_extractor(self):
 mimetype, encoding = mimetypes.guess_type(self.current_filename)
 try:
 extractor = extractor_map[mimetype]
 except KeyError:
-return "not a known archive type"
+if encoding:
+extractor = CompressionExtractor
+else:
+return "not a known archive type"
 try:
 self.current_extractor = extractor(self.current_filename, mimetype,
 encoding)
 content = self.current_extractor.check_contents()
 handler = handler_map.get(content, BombHandler)
-self.current_handler = handler(self.current_extractor, content)
+self.current_handler = handler(self.current_extractor, content,
+self.options)
 except ExtractorError, error:
 return str(error)
 def recurse(self):
 if not self.options.recursive:
 return
 for filename in self.current_extractor.included_archives:
 tail_path, basename = os.path.split(filename)
 directory = os.path.join(self.current_directory,
-self.current_handler.directory, tail_path)
+self.current_handler.target, tail_path)
 self.archives.setdefault(directory, []).append(basename)
 def report(self, function, *args):
 error = function(*args)
 if error:

Mercurial > dtrx / file comparison

comparison: scripts/x

scripts/x