1# Copyright (C) 2015, Ansgar Burchardt <ansgar@debian.org> 

2# 

3# This program is free software; you can redistribute it and/or modify 

4# it under the terms of the GNU General Public License as published by 

5# the Free Software Foundation; either version 2 of the License, or 

6# (at your option) any later version. 

7# 

8# This program is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU General Public License for more details. 

12# 

13# You should have received a copy of the GNU General Public License along 

14# with this program; if not, write to the Free Software Foundation, Inc., 

15# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 

16 

17""" 

18Helper methods to deal with (de)compressing files 

19""" 

20 

21import os 

22import shutil 

23import subprocess 

24from typing import IO, Optional 

25 

26 

27def decompress_zstd(input: IO, output: IO) -> None: 

28 subprocess.check_call(["zstd", "--decompress"], stdin=input, stdout=output) 

29 

30 

31def decompress_xz(input: IO, output: IO) -> None: 

32 subprocess.check_call(["xz", "--decompress", "-T0"], stdin=input, stdout=output) 

33 

34 

35def decompress_bz2(input: IO, output: IO) -> None: 

36 subprocess.check_call(["bzip2", "--decompress"], stdin=input, stdout=output) 

37 

38 

39def decompress_gz(input: IO, output: IO) -> None: 

40 subprocess.check_call(["gzip", "--decompress"], stdin=input, stdout=output) 

41 

42 

43decompressors = { 

44 '.zst': decompress_zstd, 

45 '.xz': decompress_xz, 

46 '.bz2': decompress_bz2, 

47 '.gz': decompress_gz, 

48} 

49 

50 

51def decompress(input: IO, output: IO, filename: Optional[str] = None) -> None: 

52 if filename is None: 

53 filename = input.name 

54 

55 base, ext = os.path.splitext(filename) 

56 decompressor = decompressors.get(ext, None) 

57 if decompressor is not None: 

58 decompressor(input, output) 

59 else: 

60 shutil.copyfileobj(input, output)