run_before_script: Refactor

Author: tony

src/tmuxp/util.py

@@ -28,41 +28,71 @@ def run_before_script(
     script_file: str | pathlib.Path,
     cwd: pathlib.Path | None = None,
 ) -> int:
-    """Execute a shell script, wraps :meth:`subprocess.check_call()` in a try/catch."""
-    try:
-        proc = subprocess.Popen(
-            shlex.split(str(script_file)),
-            stderr=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            cwd=cwd,
-            text=True,
-            errors="backslashreplace",
-            encoding="utf-8",
+    """Execute shell script, ``tee``-ing output to both terminal (if TTY) and buffer."""
+    script_cmd = shlex.split(str(script_file))
+    proc = subprocess.Popen(
+        script_cmd,
+        cwd=cwd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,  # decode to str
+        errors="backslashreplace",
+    )
+
+    out_buffer = []
+    err_buffer = []
+
+    # While process is running, read lines from stdout/stderr
+    # and write them to this process's stdout/stderr if isatty
+    is_out_tty = sys.stdout.isatty()
+    is_err_tty = sys.stderr.isatty()
+
+    # You can do a simple loop reading in real-time:
+    while True:
+        # Use .poll() to check if the child has exited
+        return_code = proc.poll()
+
+        # Read one line from stdout, if available
+        if proc.stdout:
+            line_out = proc.stdout.readline()
+        else:
+            line_out = ""
+
+        # Read one line from stderr, if available
+        if proc.stderr:
+            line_err = proc.stderr.readline()
+        else:
+            line_err = ""
+
+        if line_out:
+            out_buffer.append(line_out)
+            if is_out_tty:
+                sys.stdout.write(line_out)
+                sys.stdout.flush()
+
+        if line_err:
+            err_buffer.append(line_err)
+            if is_err_tty:
+                sys.stderr.write(line_err)
+                sys.stderr.flush()
+
+        # If no more data from pipes and process ended, break
+        if not line_out and not line_err and return_code is not None:
+            break
+
+    # At this point, the process has finished
+    return_code = proc.wait()
+
+    if return_code != 0:
+        # Join captured stderr lines for your exception
+        stderr_str = "".join(err_buffer).strip()
+        raise exc.BeforeLoadScriptError(
+            return_code,
+            os.path.abspath(script_file),  # NOQA: PTH100
+            stderr_str,
         )
-        if proc.stdout is not None:
-            for line in iter(proc.stdout.readline, ""):
-                sys.stdout.write(line)
-        proc.wait()
-
-        if proc.returncode and proc.stderr is not None:
-            stderr = proc.stderr.read()
-            proc.stderr.close()
-            stderr_strlist = stderr.split("\n")
-            stderr_str = "\n".join(list(filter(None, stderr_strlist)))  # filter empty
-
-            raise exc.BeforeLoadScriptError(
-                proc.returncode,
-                os.path.abspath(script_file),  # NOQA: PTH100
-                stderr_str,
-            )
-    except OSError as e:
-        if e.errno == 2:
-            raise exc.BeforeLoadScriptNotExists(
-                e,
-                os.path.abspath(script_file),  # NOQA: PTH100
-            ) from e
-        raise
-    return proc.returncode
+
+    return return_code
 
 
 def oh_my_zsh_auto_title() -> None: