"""
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)