File size: 2,605 Bytes
39f5ef5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
# ===================================================================== #
#                      Copyright (c) 2022 Itz-fork                      #
#                                                                       #
# 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, see <http://www.gnu.org/licenses/>   #
# ===================================================================== #

from os import path, mkdir
from .errors import ExtractionFailed
from unzipper.helpers_nexa.utils import run_shell_cmds, run_cmds_on_cr


class Extractor:
    """
    Unzip archives using 7z and zstd
    """

    def __init__(self) -> None:
        pass

    async def extract(self, arc_path: str, out: str, password: str = "", splitted: bool = False):
        """
        Extract archive using either 7z or zstd

        Parameters:

            - `arc_path` - Archive path
            - `out` - Output path
            - `password` - Password to use incase if the archive is password protected
            - `splitted` - Pass True if the archive is a splitted archive which usually ends with .001 (+) extensions
        """
        if path.splitext(arc_path)[1] == ".zst":
            mkdir(out)
            ex = await self._ext_zstd(out, arc_path)
            await self.__check_output(ex)
            return ex
        else:
            ex = await self._ext_7z(out, arc_path, password, splitted)
            await self.__check_output(ex)
            return ex

    async def _ext_7z(self, out: str, arc_path: str, password: str = "", splitted: bool = False):
        if password:
            command = f"7z x -o\"{out}\" -p\"{password}\" \"{arc_path}\" -y"
        else:
            command = f"7z x -o\"{out}\" \"{arc_path}\" -y"
        command += " -tsplit" if splitted else ""
        return await run_cmds_on_cr(run_shell_cmds, command)

    async def _ext_zstd(self, out: str, arc_path: str):
        command = f"zstd -f --output-dir-flat \"{out}\" -d \"{arc_path}\""
        return await run_cmds_on_cr(run_shell_cmds, command)

    async def __check_output(self, out: str):
        if any(e in out for e in ["Error", "Can't open as archive"]):
            raise ExtractionFailed