from __future__ import absolute_import
import bz2
import gzip
import stat
import tarfile
from .files import (
BaseFile, File,
)
DEFAULT_MTIME = 1451606400
def create_tar_from_files(fp, files):
with tarfile.open(name='', mode='w', fileobj=fp, dereference=True) as tf:
for archive_path, f in sorted(files.items()):
if not isinstance(f, BaseFile):
f = File(f)
ti = tarfile.TarInfo(archive_path)
ti.mode = f.mode or 0o0644
ti.type = tarfile.REGTYPE
if not ti.isreg():
raise ValueError('not a regular file: %s' % f)
if ti.mode & (stat.S_ISUID | stat.S_ISGID):
raise ValueError('cannot add file with setuid or setgid set: '
'%s' % f)
ti.uid = 0
ti.gid = 0
ti.uname = ''
ti.gname = ''
ti.mtime = DEFAULT_MTIME
ti.size = f.size()
tf.addfile(ti, f.open())
def create_tar_gz_from_files(fp, files, filename=None, compresslevel=9):
gf = gzip.GzipFile(filename=filename or '', mode='wb', fileobj=fp,
compresslevel=compresslevel, mtime=DEFAULT_MTIME)
with gf:
create_tar_from_files(gf, files)
class _BZ2Proxy(object):
def __init__(self, fp, compresslevel=9):
self.fp = fp
self.compressor = bz2.BZ2Compressor(compresslevel=compresslevel)
self.pos = 0
def tell(self):
return self.pos
def write(self, data):
data = self.compressor.compress(data)
self.pos += len(data)
self.fp.write(data)
def close(self):
data = self.compressor.flush()
self.pos += len(data)
self.fp.write(data)
def create_tar_bz2_from_files(fp, files, compresslevel=9):
proxy = _BZ2Proxy(fp, compresslevel=compresslevel)
create_tar_from_files(proxy, files)
proxy.close()