Upload 2 files

unzipper/lib/extractor/__init__.py

@@ -0,0 +1,61 @@
+# ===================================================================== #
+#                      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

unzipper/lib/extractor/errors.py

@@ -0,0 +1,5 @@
+# Exceptions raised by the Extractor class
+
+class ExtractionFailed(Exception):
+    def __init__(self) -> None:
+        super().__init__("Extraction failed due to an unknown error! Please make sure that your archive isn't corrupted")