refactor(sysdeps) : centralisation de la logique métier dans le Core, ajout du support macOS et amélioration de la robustesse

Author: raidos23

pycompiler_ark/Core/SysDependencyManager.py

@@ -65,9 +65,49 @@ def detect_linux_package_manager(self) -> Optional[str]:
                 return pm
         return None
 
+    def detect_macos_package_manager(self) -> Optional[str]:
+        if shutil.which("brew"):
+            return "brew"
+        return None
+
     def which(self, cmd: str) -> Optional[str]:
         return shutil.which(cmd)
 
+    def get_install_command(self, packages: list[str]) -> Optional[str]:
+        """
+        Generate the appropriate installation command string for the current platform.
+        Returns None if no supported package manager is found.
+        """
+        if not packages:
+            return None
+
+        system = platform.system()
+        if system == "Linux":
+            pm = self.detect_linux_package_manager()
+            if not pm:
+                return None
+            pkgs = " ".join(packages)
+            # Standard install command for most Linux package managers
+            return f"{pm} install -y {pkgs}"
+
+        elif system == "Windows":
+            if not shutil.which("winget"):
+                return None
+            # Building a combined winget command
+            parts = [
+                f"winget install --id {p} --silent --accept-source-agreements --accept-package-agreements"
+                for p in packages
+            ]
+            return " && ".join(parts)
+
+        elif system == "Darwin":  # macOS
+            pm = self.detect_macos_package_manager()
+            if pm == "brew":
+                pkgs = " ".join(packages)
+                return f"brew install {pkgs}"
+
+        return None
+
     def shell_run(
         self,
         cmd: Union[str, list[str]],
@@ -77,9 +117,6 @@ def shell_run(
     ) -> Optional[QProcess]:
         """Runs a command without elevation."""
         try:
-            # IMPORTANT: We create QProcess WITHOUT a parent here. 
-            # This allows it to be created and waited for (waitForFinished) 
-            # in the background thread without violating Qt thread ownership rules.
             proc = QProcess()
             if cwd:
                 proc.setWorkingDirectory(cwd)
@@ -88,20 +125,37 @@ def shell_run(
                 proc.setProgram(cmd[0])
                 proc.setArguments(cmd[1:])
             else:
-                proc.setProgram("/bin/bash" if platform.system() != "Windows" else "cmd.exe")
-                args = ["-lc", cmd] if platform.system() != "Windows" else ["/c", cmd]
+                proc.setProgram(
+                    "/bin/bash" if platform.system() != "Windows" else "cmd.exe"
+                )
+                args = (
+                    ["-lc", cmd]
+                    if platform.system() != "Windows"
+                    else ["/c", cmd]
+                )
                 proc.setArguments(args)
 
-            if on_output:
-                proc.readyReadStandardOutput.connect(lambda: on_output(proc.readAllStandardOutput().data().decode()))
-                proc.readyReadStandardError.connect(lambda: on_output(proc.readAllStandardError().data().decode()))
+            def _read_output():
+                # Use errors='replace' to avoid crashes on Windows with non-UTF8 encodings
+                raw = proc.readAllStandardOutput().data().decode(errors="replace")
+                if on_output:
+                    on_output(raw)
+
+            def _read_error():
+                raw = proc.readAllStandardError().data().decode(errors="replace")
+                if on_output:
+                    on_output(raw)
+
+            proc.readyReadStandardOutput.connect(_read_output)
+            proc.readyReadStandardError.connect(_read_error)
 
             def _finished_wrapper(ec, es):
-                if on_finished: on_finished(ec, es)
+                if on_finished:
+                    on_finished(ec, es)
                 self._unregister_task(proc)
 
             proc.finished.connect(_finished_wrapper)
-            
+
             # Note: The process is registered for global cleanup tracking
             self._register_task(proc, None, "commande système", "system command")
             proc.start()

pycompiler_ark/Ui/Gui/Dialogs/SysDependencyUI.py

@@ -109,57 +109,61 @@ def _close():
             if self._progress_dlg:
                 self._progress_dlg.close()
                 self._progress_dlg = None
+
         self._invoke_gui(_close)
 
-    def install_packages_linux(self, packages: list[str]) -> Optional[QProcess]:
-        pm = self.detect_linux_package_manager()
-        if not pm:
-            self.msg_error("Gestionnaire de paquets non détecté.", "Package manager not detected.")
+    def install_packages(self, packages: list[str]) -> Optional[QProcess]:
+        """
+        Unified entry point for package installation with GUI feedback.
+        """
+        cmd = self.get_install_command(packages)
+        if not cmd:
+            self.msg_error(
+                "Impossible de générer la commande d'installation (Gestionnaire de paquets manquant).",
+                "Cannot generate installation command (Missing package manager).",
+            )
             return None
 
-        pkgs = " ".join(packages)
-        cmd = f"{pm} install -y {pkgs}"
-        
-        self._show_progress("Installation système", "System installation", "Installation des dépendances...", "Installing dependencies...")
+        self._show_progress(
+            "Installation système",
+            "System installation",
+            "Installation des dépendances...",
+            "Installing dependencies...",
+        )
 
         def _on_output(text: str):
             lines = [l for l in text.strip().splitlines() if l.strip()]
-            if lines: self._update_progress(lines[-1][:100])
+            if lines:
+                self._update_progress(lines[-1][:100])
 
         def _on_finished(ec, es):
             self._close_progress()
-            if ec != 0: self.msg_error("L'installation a échoué.", "Installation failed.")
+            if ec != 0:
+                self.msg_error(
+                    "L'installation a échoué.", "Installation failed."
+                )
 
-        return self.run_elevated_shell(cmd, on_output=_on_output, on_finished=_on_finished)
+        # Elevation is required for system packages
+        proc = self.run_elevated_shell(
+            cmd, on_output=_on_output, on_finished=_on_finished
+        )
 
-    def install_packages_windows(self, packages: list[dict]) -> Optional[QProcess]:
-        if not shutil.which("winget"):
-            self.msg_error("winget est requis sur Windows.", "winget is required on Windows.")
-            return None
+        if proc and self._progress_dlg:
+            # Handle cancellation: if user cancels the dialog, kill the process
+            self._progress_dlg.canceled.connect(proc.terminate)
 
-        ids = [pkg.get("id") for pkg in packages if pkg.get("id")]
-        if not ids: return None
-        
-        cmd_parts = [f"winget install --id {pid} --silent --accept-source-agreements --accept-package-agreements" for pid in ids]
-        full_cmd = " && ".join(cmd_parts)
-        
-        self._show_progress("Installation Windows", "Windows installation", "Préparation de winget...", "Preparing winget...")
+        return proc
 
-        def _on_output(text: str):
-            lines = [l for l in text.strip().splitlines() if l.strip()]
-            if lines: self._update_progress(lines[-1][:100])
+    def install_packages_linux(self, packages: list[str]) -> Optional[QProcess]:
+        # Deprecated: use install_packages
+        return self.install_packages(packages)
 
-        def _on_finished(ec, es):
-            self._close_progress()
-            if ec != 0: self.msg_error("L'installation winget a échoué.", "winget installation failed.")
+    def install_packages_windows(self, packages: list[dict]) -> Optional[QProcess]:
+        # Deprecated: use install_packages
+        ids = [pkg.get("id") for pkg in packages if pkg.get("id")]
+        return self.install_packages(ids)
 
-        return self.shell_run(full_cmd, on_output=_on_output, on_finished=_on_finished)
 
 def install_system_packages(packages: list[str], gui) -> bool:
     mgr = SysDependencyUI(gui)
-    if platform.system() == "Linux":
-        return mgr.install_packages_linux(packages) is not None
-    elif platform.system() == "Windows":
-        win_pkgs = [{"id": p} for p in packages]
-        return mgr.install_packages_windows(win_pkgs) is not None
-    return False
+    return mgr.install_packages(packages) is not None