66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
""" Unpacks an `All-in-One WP Migration` package. """
|
|
|
|
import errno
|
|
import os
|
|
from error import Ai1wmError
|
|
from header import Ai1wmHeader
|
|
from tools import s__
|
|
|
|
|
|
class Ai1wmUnpacker(object):
|
|
""" Unpacks an `All-in-One WP Migration` package. """
|
|
|
|
@staticmethod
|
|
def __make_dirs(path, mode=0o777):
|
|
""" A simple wrapper of os.makedirs(), which does not raise exception if the leaf directory already exists. """
|
|
|
|
try:
|
|
os.makedirs(path, mode=mode)
|
|
except OSError as e:
|
|
if e.errno != errno.EEXIST:
|
|
raise Ai1wmError('error creating a directory: {}, error: {}'.format(path, e))
|
|
return path
|
|
|
|
@classmethod
|
|
def __extract_file(cls, stream, path, size):
|
|
""" Extracts a file from the input stream. """
|
|
|
|
block_size = 0x4000
|
|
|
|
with open(path, 'wb') as f:
|
|
while size > 0:
|
|
if block_size > size:
|
|
block_size = size
|
|
block = stream.read(block_size)
|
|
if len(block) != block_size:
|
|
raise Ai1wmError('error extracting a file: {}, error: bad file size'.format(path))
|
|
f.write(block)
|
|
size -= len(block)
|
|
|
|
@classmethod
|
|
def __unpack(cls, stream, target):
|
|
""" Unpacks a package. """
|
|
|
|
while True:
|
|
header = Ai1wmHeader.unpack(stream.read(Ai1wmHeader.SIZE))
|
|
if header.eof:
|
|
break
|
|
|
|
path = os.path.join(target, header.path)
|
|
cls.__make_dirs(path)
|
|
|
|
path = os.path.join(path, header.name)
|
|
cls.__extract_file(stream, path, header.size)
|
|
|
|
@classmethod
|
|
def unpack(cls, source, target):
|
|
""" Unpacks a package. """
|
|
|
|
source, target = s__(os.path.realpath(source)), s__(os.path.realpath(target))
|
|
try:
|
|
with open(source, 'rb') as f:
|
|
cls.__unpack(f, target)
|
|
except Exception as ex:
|
|
raise Ai1wmError('error unpacking a file: {}, error: {}'.format(source, ex))
|
|
return target
|