| 166 | |
| 167 | |
| 168 | class CompressionSpec: |
| 169 | def __init__(self, s): |
| 170 | if isinstance(s, CompressionSpec): |
| 171 | self.__dict__.update(s.__dict__) |
| 172 | return |
| 173 | values = s.split(",") |
| 174 | count = len(values) |
| 175 | if count < 1: |
| 176 | raise ArgumentTypeError("not enough arguments") |
| 177 | # --compression algo[,level] |
| 178 | self.name = values[0] |
| 179 | if self.name in ("none", "lz4"): |
| 180 | return |
| 181 | elif self.name in ("zlib", "lzma", "zlib_legacy"): # zlib_legacy just for testing |
| 182 | if count < 2: |
| 183 | level = 6 # default compression level in py stdlib |
| 184 | elif count == 2: |
| 185 | level = int(values[1]) |
| 186 | if not 0 <= level <= 9: |
| 187 | raise ArgumentTypeError("level must be >= 0 and <= 9") |
| 188 | else: |
| 189 | raise ArgumentTypeError("too many arguments") |
| 190 | self.level = level |
| 191 | elif self.name in ("zstd",): |
| 192 | if count < 2: |
| 193 | level = 3 # default compression level in zstd |
| 194 | elif count == 2: |
| 195 | level = int(values[1]) |
| 196 | if not 1 <= level <= 22: |
| 197 | raise ArgumentTypeError("level must be >= 1 and <= 22") |
| 198 | else: |
| 199 | raise ArgumentTypeError("too many arguments") |
| 200 | self.level = level |
| 201 | elif self.name == "auto": |
| 202 | if 2 <= count <= 3: |
| 203 | compression = ",".join(values[1:]) |
| 204 | else: |
| 205 | raise ArgumentTypeError("bad arguments") |
| 206 | self.inner = CompressionSpec(compression) |
| 207 | elif self.name == "obfuscate": |
| 208 | if 3 <= count <= 5: |
| 209 | level = int(values[1]) |
| 210 | if not ((1 <= level <= 6) or (110 <= level <= 123) or (level == 250)): |
| 211 | raise ArgumentTypeError("level must be (inclusively) within 1...6, 110...123 or equal to 250") |
| 212 | self.level = level |
| 213 | compression = ",".join(values[2:]) |
| 214 | else: |
| 215 | raise ArgumentTypeError("bad arguments") |
| 216 | self.inner = CompressionSpec(compression) |
| 217 | else: |
| 218 | raise ArgumentTypeError("unsupported compression type") |
| 219 | |
| 220 | @property |
| 221 | def compressor(self): |
| 222 | from ..compress import get_compressor |
| 223 | |
| 224 | if self.name in ("none", "lz4"): |
| 225 | return get_compressor(self.name) |
no outgoing calls