[go: up one dir, main page]

File: compress.py

package info (click to toggle)
kiwi 10.2.33-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 7,528 kB
  • sloc: python: 67,299; sh: 3,980; xml: 3,379; ansic: 391; makefile: 354
file content (136 lines) | stat: -rw-r--r-- 4,526 bytes parent folder | download
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
# Copyright (c) 2015 SUSE Linux GmbH.  All rights reserved.
#
# This file is part of kiwi.
#
# kiwi 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 3 of the License, or
# (at your option) any later version.
#
# kiwi 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 kiwi.  If not, see <http://www.gnu.org/licenses/>
#
import os
import logging
from typing import List, Optional

# project
from kiwi.utils.temporary import Temporary
from kiwi.command import Command
from kiwi.defaults import Defaults

from kiwi.exceptions import (
    KiwiFileNotFound,
    KiwiCompressionFormatUnknown
)

log = logging.getLogger('kiwi')


class Compress:
    """
    **File compression / decompression**

    :param bool keep_source: Request to keep the uncompressed source
    :param str source_filename: Source file name to compress
    :param list supported_zipper: List of supported compression tools
    :param str compressed_filename: Compressed file name path with
        compression suffix
    :param str uncompressed_filename:
        Uncompressed file name path
    """
    def __init__(
        self, source_filename: str, keep_source_on_compress: bool = False
    ) -> None:
        if not os.path.exists(source_filename):
            raise KiwiFileNotFound(
                f'compression source file {source_filename} not found'
            )
        self.keep_source = keep_source_on_compress
        self.source_filename = source_filename
        self.supported_zipper = [
            'xz', 'gzip'
        ]
        self.compressed_filename: Optional[str] = None
        self.uncompressed_filename: Optional[str] = None

    def xz(self, options: Optional[List[str]] = None) -> str:
        """
        Create XZ compressed file

        :param list options: custom xz compression options
        """
        if not options:
            options = Defaults.get_xz_compression_options()
        assert options
        if self.keep_source:
            options.append('--keep')
        Command.run(
            ['xz', '-f'] + options + [self.source_filename]
        )
        self.compressed_filename = self.source_filename + '.xz'
        return self.compressed_filename

    def gzip(self) -> str:
        """
        Create gzip(max compression) compressed file
        """
        options = [
            '-9'
        ]
        if self.keep_source:
            options.append('--keep')
        Command.run(
            ['gzip', '-f'] + options + [self.source_filename]
        )
        self.compressed_filename = self.source_filename + '.gz'
        return self.compressed_filename

    def uncompress(self, temporary: bool = False) -> str:
        """
        Uncompress with format autodetection

        By default the original source file will be changed into
        the uncompressed variant. If temporary is set to True
        a temporary file is created instead

        :param bool temporary: uncompress to a temporary file
        """
        zipper = self.get_format()
        if not zipper:
            raise KiwiCompressionFormatUnknown(
                f'could not detect compression format for {self.source_filename}'
            )
        if not temporary:
            Command.run([zipper, '-d', self.source_filename])
            self.uncompressed_filename = self.source_filename
        else:
            self.temp_file = Temporary().new_file()
            bash_command = [
                zipper, '-c', '-d', self.source_filename,
                '>', self.temp_file.name
            ]
            Command.run(['bash', '-c', ' '.join(bash_command)])
            self.uncompressed_filename = self.temp_file.name
        return self.uncompressed_filename

    def get_format(self) -> Optional[str]:
        """
        Detect compression format

        :return: compression format name or None if it couldn't be inferred

        :rtype: Optional[str]
        """
        for zipper in self.supported_zipper:
            result = Command.run(
                [zipper, '-l', self.source_filename], raise_on_error=False
            )
            if result.returncode == 0:
                return zipper
        return None