"""
Helper code for file writing with optional compression.
@contact: Debian FTPMaster <ftpmaster@debian.org>
@copyright: 2011 Torsten Werner <twerner@debian.org>
@license: GNU General Public License version 2 or later
"""
################################################################################
# This program 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.
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
################################################################################
import errno
import os
import os.path
import subprocess
from dataclasses import dataclass
from typing import Optional, TextIO
[docs]@dataclass
class CompressionMethod:
keyword: str
extension: str
command: Optional[list[str]]
_compression_methods = (
CompressionMethod('bzip2', '.bz2', ['bzip2', '-9']),
CompressionMethod('gzip', '.gz', ['gzip', '-9cn', '--rsyncable', '--no-name']),
CompressionMethod('xz', '.xz', ['xz', '-c', '-e', '-T0']),
CompressionMethod('zstd', '.zst', ['zstd', '--compress']),
# 'none' must be the last compression method as BaseFileWriter
# handling it will remove the input file for other compressions
CompressionMethod('none', '', None),
)
[docs]class BaseFileWriter:
'''
Base class for compressed and uncompressed file writing.
'''
def __init__(self, template, **keywords):
'''
The template argument is a string template like
"dists/%(suite)s/%(component)s/Contents-%(architecture)s.gz" that
should be relative to the archive's root directory. The keywords
include strings for suite, component, architecture and booleans
uncompressed, gzip, bzip2.
'''
self.compression = keywords.get('compression', ['none'])
self.path = template % keywords
[docs] def open(self) -> TextIO:
'''
Returns a file object for writing.
'''
# create missing directories
try:
os.makedirs(os.path.dirname(self.path))
except:
pass
self.file = open(self.path + '.new', 'w')
return self.file
# internal helper function
[docs] def rename(self, filename: str) -> None:
tempfilename = filename + '.new'
os.chmod(tempfilename, 0o644)
os.rename(tempfilename, filename)
# internal helper function to compress output
[docs] def compress(self, cmd, suffix, path) -> None:
in_filename = "{0}.new".format(path)
out_filename = "{0}{1}.new".format(path, suffix)
if cmd is not None:
with open(in_filename, 'r') as in_fh, open(out_filename, 'w') as out_fh:
subprocess.check_call(cmd, stdin=in_fh, stdout=out_fh, close_fds=True)
self.rename("{0}{1}".format(path, suffix))
[docs] def close(self) -> None:
'''
Closes the file object and does the compression and rename work.
'''
self.file.close()
for method in _compression_methods:
if method.keyword in self.compression:
self.compress(method.command, method.extension, self.path)
else:
# Try removing the file that would be generated.
# It's not an error if it does not exist.
try:
os.unlink("{0}{1}".format(self.path, method.extension))
except OSError as e:
if e.errno != errno.ENOENT:
raise
else:
os.unlink(self.path + '.new')
[docs]class BinaryContentsFileWriter(BaseFileWriter):
def __init__(self, **keywords):
'''
The value of the keywords suite, component, and architecture are
strings. The value of component may be omitted if not applicable.
Output files are gzip compressed only.
'''
flags = {
'compression': ['gzip'],
}
flags.update(keywords)
if flags['debtype'] == 'deb':
template = "%(archive)s/dists/%(suite)s/%(component)s/Contents-%(architecture)s"
else: # udeb
template = "%(archive)s/dists/%(suite)s/%(component)s/Contents-udeb-%(architecture)s"
BaseFileWriter.__init__(self, template, **flags)
[docs]class SourceContentsFileWriter(BaseFileWriter):
def __init__(self, **keywords):
'''
The value of the keywords suite and component are strings.
Output files are gzip compressed only.
'''
flags = {
'compression': ['gzip'],
}
flags.update(keywords)
template = "%(archive)s/dists/%(suite)s/%(component)s/Contents-source"
BaseFileWriter.__init__(self, template, **flags)
[docs]class PackagesFileWriter(BaseFileWriter):
def __init__(self, **keywords):
'''
The value of the keywords suite, component, debtype and architecture
are strings. Output files are gzip compressed only.
'''
flags = {
'compression': ['gzip', 'xz'],
}
flags.update(keywords)
if flags['debtype'] == 'deb':
template = "%(archive)s/dists/%(suite)s/%(component)s/binary-%(architecture)s/Packages"
else: # udeb
template = "%(archive)s/dists/%(suite)s/%(component)s/debian-installer/binary-%(architecture)s/Packages"
BaseFileWriter.__init__(self, template, **flags)
[docs]class SourcesFileWriter(BaseFileWriter):
def __init__(self, **keywords):
'''
The value of the keywords suite and component are strings. Output
files are gzip compressed only.
'''
flags = {
'compression': ['gzip', 'xz'],
}
flags.update(keywords)
template = "%(archive)s/dists/%(suite)s/%(component)s/source/Sources"
BaseFileWriter.__init__(self, template, **flags)
[docs]class TranslationFileWriter(BaseFileWriter):
def __init__(self, **keywords):
'''
The value of the keywords suite, component and language are strings.
Output files are bzip2 compressed only.
'''
flags = {
'compression': ['bzip2'],
'language': 'en',
}
flags.update(keywords)
template = "%(archive)s/dists/%(suite)s/%(component)s/i18n/Translation-%(language)s"
super(TranslationFileWriter, self).__init__(template, **flags)