1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
# repack_tarball.py -- Repack files/dirs in to tarballs.
# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
#
# This file is part of bzr-builddeb.
#
# bzr-builddeb is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# bzr-builddeb is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with bzr-builddeb; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
#
import gzip
import os
from io import BytesIO
import tarfile
import bz2
import hashlib
import shutil
import time
import zipfile
from ...errors import (
BzrError,
DependencyNotPresent,
)
from ...transport import get_transport, FileExists
from .util import open_file, open_file_via_transport
class UnsupportedRepackFormat(BzrError):
_fmt = ('Either the file extension of "%(location)s" indicates that '
'it is a format unsupported for repacking or it is a '
'remote directory.')
def __init__(self, location):
BzrError.__init__(self, location=location)
class TgzRepacker:
"""Repacks something to be a .tar.gz"""
def __init__(self, source_f):
"""Create a repacker that repacks what is in source_f.
:param source_f: a file object to read the source from.
"""
self.source_f = source_f
def repack(self, target_f):
"""Repacks and writes the repacked tar.gz to target_f.
target_f should be closed after calling this method.
:param target_f: a file object to write the result to.
"""
raise NotImplementedError(self.repack)
class CopyRepacker(TgzRepacker):
"""A Repacker that just copies."""
def repack(self, target_f):
shutil.copyfileobj(self.source_f, target_f)
class TarTgzRepacker(TgzRepacker):
"""A TgzRepacker that just gzips the input."""
def repack(self, target_f):
with gzip.GzipFile(mode='w', fileobj=target_f) as gz:
shutil.copyfileobj(self.source_f, gz)
class Tbz2TgzRepacker(TgzRepacker):
"""A TgzRepacker that repacks from a .tar.bz2."""
def repack(self, target_f):
content = bz2.decompress(self.source_f.read())
with gzip.GzipFile(mode='w', fileobj=target_f) as gz:
gz.write(content)
class TarLzma2TgzRepacker(TgzRepacker):
"""A TgzRepacker that repacks from a .tar.lzma or .tar.xz."""
def repack(self, target_f):
try:
import lzma
except ImportError as e:
raise DependencyNotPresent('lzma', e) from e
content = lzma.decompress(self.source_f.read())
with gzip.GzipFile(mode='w', fileobj=target_f) as gz:
gz.write(content)
class ZipTgzRepacker(TgzRepacker):
"""A TgzRepacker that repacks from a .zip file."""
def _repack_zip_to_tar(self, zip, tar):
for info in zip.infolist():
tarinfo = tarfile.TarInfo(info.filename)
tarinfo.size = info.file_size
tarinfo.mtime = time.mktime(info.date_time + (0, 1, -1))
if info.filename.endswith("/"):
tarinfo.mode = 0o755
tarinfo.type = tarfile.DIRTYPE
else:
tarinfo.mode = 0o644
tarinfo.type = tarfile.REGTYPE
contents = BytesIO(zip.read(info.filename))
tar.addfile(tarinfo, contents)
def repack(self, target_f):
with zipfile.ZipFile(self.source_f, "r") as zip:
with tarfile.open(mode="w:gz", fileobj=target_f) as tar:
self._repack_zip_to_tar(zip, tar)
def get_filetype(filename):
types = {
".tar.gz": "gz",
".tgz": "gz",
".tar.bz2": "bz2",
".tar.xz": "xz",
".tar.lzma": "lzma",
".tbz2": "bz2",
".tar": "tar",
".zip": "zip"
}
for filetype, name in types.items():
if filename.endswith(filetype):
return name
def get_repacker_class(source_format, target_format):
"""Return the appropriate repacker based on the file extension."""
if source_format == target_format:
return CopyRepacker
known_formatters = {
("bz2", "gz"): Tbz2TgzRepacker,
("lzma", "gz"): TarLzma2TgzRepacker,
("xz", "gz"): TarLzma2TgzRepacker,
("tar", "gz"): TarTgzRepacker,
("zip", "gz"): ZipTgzRepacker,
}
return known_formatters.get((source_format, target_format))
def _error_if_exists(target_transport, new_name, source_name):
with open_file(source_name) as source_f:
source_sha = hashlib.sha1(source_f.read()).hexdigest()
with open_file_via_transport(new_name, target_transport) as target_f:
target_sha = hashlib.sha1(target_f.read()).hexdigest()
if source_sha != target_sha:
raise FileExists(new_name)
def _repack_directory(target_transport, new_name, source_name):
target_transport.ensure_base()
with target_transport.open_write_stream(new_name) as target_f:
with tarfile.open(mode='w:gz', fileobj=target_f) as tar:
tar.add(source_name, os.path.basename(source_name))
def _repack_other(target_transport, new_name, source_name):
source_filetype = get_filetype(source_name)
target_filetype = get_filetype(new_name)
repacker_cls = get_repacker_class(source_filetype, target_filetype)
if repacker_cls is None:
raise UnsupportedRepackFormat(source_name)
target_transport.ensure_base()
with target_transport.open_write_stream(new_name) as target_f:
with open_file(source_name) as source_f:
repacker = repacker_cls(source_f)
repacker.repack(target_f)
def repack_tarball(source_name, new_name, target_dir=None):
"""Repack the file/dir named to a .tar.gz with the chosen name.
This function takes a named file of either .tar.gz, .tar .tgz .tar.bz2
or .zip type, or a directory, and creates the file named in the second
argument in .tar.gz format.
If target_dir is specified then that directory will be created if it
doesn't exist, and the new_name will be interpreted relative to that
directory.
The source must exist, and the target cannot exist, unless it is identical
to the source.
:param source_name: the current name of the file/dir
:type source_name: string
:param new_name: the desired name of the tarball
:type new_name: string
:keyword target_dir: the directory to consider new_name relative to, and
will be created if non-existant.
:type target_dir: string
:return: None
:throws NoSuchFile: if source_name doesn't exist.
:throws FileExists: if the target filename (after considering target_dir)
exists, and is not identical to the source.
:throws BzrCommandError: if the source isn't supported for repacking.
"""
if target_dir is None:
target_dir = "."
extra, new_name = os.path.split(new_name)
target_transport = get_transport(os.path.join(target_dir, extra))
if target_transport.has(new_name):
source_format = get_filetype(source_name)
target_format = get_filetype(new_name)
if source_format != target_format:
raise FileExists(new_name)
_error_if_exists(target_transport, new_name, source_name)
return
if os.path.isdir(source_name):
_repack_directory(target_transport, new_name, source_name)
else:
_repack_other(target_transport, new_name, source_name)
|
