Module refinery.units.formats.archive.xtiex

Expand source code Browse git 
from __future__ import annotations

from refinery import perc, xtcab
from refinery.lib.id import buffer_contains, is_likely_pe
from refinery.lib.tools import exception_to_string
from refinery.lib.types import buf
from refinery.units import Chunk
from refinery.units.formats.archive import ArchiveUnit


class xtiex(xtcab, docs='{0}{p}{PathExtractorUnit}'):
    """
    Extract files from IExpress setup self-extracting archive. If the archive contains more than one
    resource named CABINET, the files of each cabinet are listed below the path of its resource.
    """
    def unpack(self, data: buf):
        with data | perc('RCDATA/CABINET/*') as resources:
            cabinets = [Chunk.Wrap(resource) for resource in resources]
        if not cabinets:
            raise ValueError('no RCDATA resource named CABINET was found')
        if len(cabinets) == 1:
            yield from super().unpack(cabinets[0])
            return
        readable = 0
        for cabinet in cabinets:
            resource = str(cabinet.meta['path'])
            try:
                results = list(super().unpack(cabinet))
            except Exception as error:
                self.log_warn(F'failed to read the resource {resource}: {exception_to_string(error)}')
                continue
            readable += 1
            for result in results:
                result.path = F'{resource}/{result.path}'
                yield result
        if not readable:
            raise ValueError(F'none of the {len(cabinets)} resources named CABINET could be read')

    def filter(self, chunks):
        yield from super(ArchiveUnit, self).filter(chunks)

    @classmethod
    def handles(cls, data: buf) -> bool:
        if not is_likely_pe(data):
            return False
        return buffer_contains(data, B'CABINET')

Classes

class xtiex (*paths, list=False, join_path=False, drop_path=False, fuzzy=0, exact=False, regex=False, path=b'path', exclude=None, date=b'date', pwd=b'')

Extract files from IExpress setup self-extracting archive. If the archive contains more than one resource named CABINET, the files of each cabinet are listed below the path of its resource.

This unit extracts items with an associated virtual path from a container; each extracted item is emitted as a separate chunk with a corresponding meta variable named "path".

Positional arguments to PathExtractorUnit are patterns to filter the extracted items. Use the -x flag to add an exclusion pattern. To extract all files with a foo or bar extension, but none that has the word "temp" in its path:

PathExtractorUnit .foo .bar -x temp

To view only the paths of all chunks, use the listing switch:

emit data | ... | PathExtractorUnit -l

Otherwise, extracted items are written to the standard output port and usually require a frame to properly process. In order to dump all extracted data to disk, the following pipeline can be used:

emit data | ... | PathExtractorUnit [| dump extracted/{path} ]

The value {path} is a placeholder which is substituted by the virtual path of the extracted item. When several items have the same path, each of them gets a number appended to its path in the order in which PathExtractorUnit lists them, and a pattern that matches the path before the number selects all of them. When using PathExtractorUnit to unpack a file on disk, the following pattern can be useful:

ef pack.bin [| PathExtractorUnit -j | d2p ]

The unit ef is also a path extractor. By specifying -j (or --join), the paths of extracted items are combined. Here, d2p is a shortcut for dump {path}. It deconflicts the joined paths with the local file system: If pack.bin contains items one.txt and two.txt, the following local file tree would be the result:

pack.bin
pack/one.txt
pack/two.txt

Finally, the -d (or --drop) switch can be used to not create (or alter) the path metadata at all, which is useful in cases where path metadata from a previous unit should be preserved.

Expand source code Browse git 
class xtiex(xtcab, docs='{0}{p}{PathExtractorUnit}'):
    """
    Extract files from IExpress setup self-extracting archive. If the archive contains more than one
    resource named CABINET, the files of each cabinet are listed below the path of its resource.
    """
    def unpack(self, data: buf):
        with data | perc('RCDATA/CABINET/*') as resources:
            cabinets = [Chunk.Wrap(resource) for resource in resources]
        if not cabinets:
            raise ValueError('no RCDATA resource named CABINET was found')
        if len(cabinets) == 1:
            yield from super().unpack(cabinets[0])
            return
        readable = 0
        for cabinet in cabinets:
            resource = str(cabinet.meta['path'])
            try:
                results = list(super().unpack(cabinet))
            except Exception as error:
                self.log_warn(F'failed to read the resource {resource}: {exception_to_string(error)}')
                continue
            readable += 1
            for result in results:
                result.path = F'{resource}/{result.path}'
                yield result
        if not readable:
            raise ValueError(F'none of the {len(cabinets)} resources named CABINET could be read')

    def filter(self, chunks):
        yield from super(ArchiveUnit, self).filter(chunks)

    @classmethod
    def handles(cls, data: buf) -> bool:
        if not is_likely_pe(data):
            return False
        return buffer_contains(data, B'CABINET')

Ancestors

Subclasses

Methods

def unpack(self, data)
Expand source code Browse git 
def unpack(self, data: buf):
    with data | perc('RCDATA/CABINET/*') as resources:
        cabinets = [Chunk.Wrap(resource) for resource in resources]
    if not cabinets:
        raise ValueError('no RCDATA resource named CABINET was found')
    if len(cabinets) == 1:
        yield from super().unpack(cabinets[0])
        return
    readable = 0
    for cabinet in cabinets:
        resource = str(cabinet.meta['path'])
        try:
            results = list(super().unpack(cabinet))
        except Exception as error:
            self.log_warn(F'failed to read the resource {resource}: {exception_to_string(error)}')
            continue
        readable += 1
        for result in results:
            result.path = F'{resource}/{result.path}'
            yield result
    if not readable:
        raise ValueError(F'none of the {len(cabinets)} resources named CABINET could be read')

Inherited members