dtrx: comparison scripts/dtrx

-:a8f875e02c83
+:957b402d4b90
 def prepare(self):
 pass
 def check_included_archives(self, filenames):
 for filename in filenames:
-if (ExtractorBuilder.try_by_mimetype(filename)[0] or
+if (ExtractorBuilder.try_by_mimetype(filename) or
-ExtractorBuilder.try_by_extension(filename)[0]):
+ExtractorBuilder.try_by_extension(filename)):
 self.included_archives.append(filename)
 def check_contents(self):
 filenames = os.listdir('.')
 if not filenames:
 def extract(self):
 self.content_type = ONE_ENTRY_KNOWN
 self.content_name = self.basename()
 output_fd, self.target = tempfile.mkstemp(prefix='.dtrx-', dir='.')
-self.run_pipes(output_fd)
+try:
+self.run_pipes(output_fd)
+except ExtractorError:
+os.close(output_fd)
+os.unlink(self.target)
+raise
 os.close(output_fd)
 class TarExtractor(BaseExtractor):
 def get_filenames(self):
 def extract_archive(self):
 self.pipe(['tar', '-x'])
 self.run_pipes()
-class ZipExtractor(BaseExtractor):
-def __init__(self, filename, encoding):
-BaseExtractor.__init__(self, '/dev/null', None)
-self.filename = os.path.realpath(filename)
-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 CpioExtractor(BaseExtractor):
 def get_filenames(self):
 self.pipe(['cpio', '-t'], "listing")
 return BaseExtractor.get_filenames(self)
 def basename(self):
 return os.path.basename(self.filename) + '-metadata.txt'
-class SevenExtractor(BaseExtractor):
+class NoPipeExtractor(BaseExtractor):
-border_re = re.compile('^[- ]+$')
+# Some extraction tools won't accept the archive from stdin.  With
+# these, the piping infrastructure we normally set up generally doesn't
+# work, at least at first.  We can still use most of it; we just can't
+# seed self.archive with the archive file.  So instead we seed it with
+# /dev/null, and specify the filename on the command line as necessary.
+# This class doesn't do anything by itself; it's just meant to be a
+# base class for extractors that rely on these dumb tools.
 def __init__(self, filename, encoding):
 BaseExtractor.__init__(self, '/dev/null', None)
 self.filename = os.path.realpath(filename)
+class ZipExtractor(NoPipeExtractor):
+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):
+border_re = re.compile('^[- ]+$')
 def get_filenames(self):
 self.pipe(['7z', 'l', self.filename], "listing")
 self.run_pipes()
 self.archive.seek(0, 0)
 def extract_archive(self):
 self.pipe(['7z', 'x', self.filename])
 self.run_pipes()
+class CABExtractor(NoPipeExtractor):
+border_re = re.compile(r'^[-\+]+$')
+def get_filenames(self):
+self.pipe(['cabextract', '-l', self.filename], "listing")
+self.run_pipes()
+self.archive.seek(0, 0)
+fn_index = None
+for line in self.archive:
+if self.border_re.match(line):
+break
+for line in self.archive:
+try:
+yield line.split(' | ', 2)[2].rstrip('\n')
+except IndexError:
+break
+self.archive.close()
+def extract_archive(self):
+self.pipe(['cabextract', '-q', self.filename])
+self.run_pipes()
 class BaseHandler(object):
 def __init__(self, extractor, options):
 self.extractor = extractor
 self.options = options
 ((contents == ONE_ENTRY) and
 options.one_entry_policy.ok_for_match()))
 can_handle = staticmethod(can_handle)
 def organize(self):
+source = os.path.join(self.extractor.target,
+os.listdir(self.extractor.target)[0])
+if os.path.isdir(source):
+checker = DirectoryChecker
+else:
+checker = FilenameChecker
 if self.options.one_entry_policy == EXTRACT_HERE:
 destination = self.extractor.content_name.rstrip('/')
 else:
 destination = self.extractor.basename()
-self.target = self.extractor.name_checker(destination).check()
+self.target = checker(destination).check()
 if os.path.isdir(self.extractor.target):
-os.rename(os.path.join(self.extractor.target,
+os.rename(source, self.target)
-os.listdir(self.extractor.target)[0]),
-self.target)
 os.rmdir(self.extractor.target)
 else:
 os.rename(self.extractor.target, self.target)
 'deb': (DebExtractor, DebMetadataExtractor),
 'rpm': (RPMExtractor, None),
 'cpio': (CpioExtractor, None),
 'gem': (GemExtractor, GemMetadataExtractor),
 'compress': (CompressionExtractor, None),
-'7z': (SevenExtractor, None)}
+'7z': (SevenExtractor, None),
+'cab': (CABExtractor, None)}
 mimetype_map = {}
 for mapping in (('tar', 'x-tar'),
 ('zip', 'x-msdos-program', 'zip'),
 ('deb', 'x-debian-package'),
 ('rpm', 'x-redhat-package-manager', 'x-rpm'),
 ('cpio', 'x-cpio'),
 ('gem', 'x-ruby-gem'),
-('7z', 'x-7z-compressed')):
+('7z', 'x-7z-compressed'),
+('cab', 'x-cab')):
 for mimetype in mapping[1:]:
 if '/' not in mimetype:
 mimetype = 'application/' + mimetype
 mimetype_map[mimetype] = mapping[0]
 for mapping in (('deb', 'Debian binary package'),
 ('cpio', 'cpio archive'),
 ('tar', 'POSIX tar archive'),
 ('zip', 'Zip archive'),
 ('rpm', 'RPM'),
-('7z', '7-zip archive')):
+('7z', '7-zip archive'),
+('cab', 'Microsoft Cabinet archive')):
 for pattern in mapping[1:]:
 magic_mime_map[re.compile(pattern)] = mapping[0]
 magic_encoding_map = {}
 for mapping in (('bzip2', 'bzip2 compressed'),
 ('zip', None, 'zip', 'exe'),
 ('deb', None, 'deb'),
 ('rpm', None, 'rpm'),
 ('cpio', None, 'cpio'),
 ('gem', None, 'gem'),
-('compress', None, 'Z', 'gz', 'bz2', 'lzma'),
+('compress', 'gzip', 'Z', 'gz'),
-('7z', None, '7z')):
+('compress', 'bzip2', 'bz2'),
+('compress', 'lzma', 'lzma'),
+('7z', None, '7z'),
+('cab', None, 'cab', 'exe')):
 for extension in mapping[2:]:
-extension_map[extension] = mapping[:2]
+extension_map.setdefault(extension, []).append(mapping[:2])
 def __init__(self, filename, options):
 self.filename = filename
 self.options = options
 extractor = extractors[0]
 return extractor(self.filename, encoding)
 def get_extractor(self):
 for func_name in ('mimetype', 'extension', 'magic'):
-archive_type, encoding = \
+logger.debug("getting extractors by %s" % (func_name,))
-getattr(self, 'try_by_' + func_name)(self.filename)
+extractor_types = \
-logger.debug("%s extractor is %s, %s" %
+getattr(self, 'try_by_' + func_name)(self.filename)
-(func_name, archive_type, encoding))
+logger.debug("done getting extractors")
-if archive_type is not None:
+for ext_args in extractor_types:
-yield self.build_extractor(archive_type, encoding)
+logger.debug("trying %s extractor from %s" %
+(ext_args, func_name))
+yield self.build_extractor(*ext_args)
 def try_by_mimetype(cls, filename):
 mimetype, encoding = mimetypes.guess_type(filename)
 try:
-return cls.mimetype_map[mimetype], encoding
+return [(cls.mimetype_map[mimetype], encoding)]
 except KeyError:
 if encoding:
-return 'compress', encoding
+return [('compress', encoding)]
-return None, None
+return []
 try_by_mimetype = classmethod(try_by_mimetype)
+def magic_map_matches(cls, output, magic_map):
+return [result for regexp, result in magic_map.items()
+if regexp.search(output)]
+magic_map_matches = classmethod(magic_map_matches)
 def try_by_magic(cls, filename):
 process = subprocess.Popen(['file', '-z', filename],
 stdout=subprocess.PIPE)
 status = process.wait()
 if status != 0:
-return None, None
+return []
 output = process.stdout.readline()
 process.stdout.close()
 if output.startswith('%s: ' % filename):
 output = output[len(filename) + 2:]
-results = [None, None]
+mimes = cls.magic_map_matches(output, cls.magic_mime_map)
-for index, mapping in enumerate((cls.magic_mime_map,
+encodings = cls.magic_map_matches(output, cls.magic_encoding_map)
-cls.magic_encoding_map)):
+if mimes and not encodings:
-for regexp, result in mapping.items():
+encodings = [None]
-if regexp.search(output):
+elif encodings and not mimes:
-results[index] = result
+mimes = ['compress']
-break
+return [(m, e) for m in mimes for e in encodings]
-return results
 try_by_magic = classmethod(try_by_magic)
 def try_by_extension(cls, filename):
 parts = filename.rsplit('.', 2)[1:]
+results = []
 while parts:
-try:
+results.extend(cls.extension_map.get('.'.join(parts), []))
-return cls.extension_map['.'.join(parts)]
+del parts[0]
-except KeyError:
+return results
-del parts[0]
-return [None, None]
 try_by_extension = classmethod(try_by_extension)
 class BaseAction(object):
 def __init__(self, options, filenames):
 if extractor.content_type == ONE_ENTRY:
 self.options.one_entry_policy.prep(self.current_filename,
 extractor.content_name)
 for handler in self.handlers:
 if handler.can_handle(extractor.content_type, self.options):
+logger.debug("using %s handler" % (handler.__name__,))
 self.current_handler = handler(extractor, self.options)
 break
 def run(self, filename, extractor):
 self.current_filename = filename
 parser.add_option('-n', '--noninteractive', dest='batch',
 action='store_true', default=False,
 help="don't ask how to handle special cases")
 parser.add_option('-m', '--metadata', dest='metadata',
 action='store_true', default=False,
-help="extract metadata from a .deb/.gem/etc.")
+help="extract metadata from a .deb/.gem")
 self.options, filenames = parser.parse_args(arguments)
 if not filenames:
 parser.error("you did not list any archives")
 self.options.one_entry_policy = OneEntryPolicy(self.options)
 self.options.recursion_policy = RecursionPolicy(self.options)

Mercurial > dtrx / file comparison

comparison: scripts/dtrx

scripts/dtrx