Refactor command logs (archlinux#3549)

Author: codefiles

archinstall/lib/general.py

@@ -237,19 +237,9 @@ def peak(self, output: str | bytes) -> bool:
 				except UnicodeDecodeError:
 					return False
 
-			peak_logfile = Path(f'{storage["LOG_PATH"]}/cmd_output.txt')
+			_cmd_output(output)
 
-			change_perm = False
-			if peak_logfile.exists() is False:
-				change_perm = True
-
-			with peak_logfile.open('a') as peek_output_log:
-				peek_output_log.write(str(output))
-
-			if change_perm:
-				peak_logfile.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP)
-
-			sys.stdout.write(str(output))
+			sys.stdout.write(output)
 			sys.stdout.flush()
 
 		return True
@@ -296,7 +286,7 @@ def execute(self) -> bool:
 
 		# https://stackoverflow.com/questions/4022600/python-pty-fork-how-does-it-work
 		if not self.pid:
-			_log_cmd(self.cmd)
+			_cmd_history(self.cmd)
 
 			try:
 				os.execve(self.cmd[0], list(self.cmd), {**os.environ, **self.environment_vars})
@@ -424,29 +414,36 @@ def trace_log(self) -> bytes | None:
 		return None
 
 
-def _log_cmd(cmd: list[str]) -> None:
-	history_logfile = Path(f'{storage["LOG_PATH"]}/cmd_history.txt')
+def _append_log(file: str, content: str) -> None:
+	path = Path(f'{storage["LOG_PATH"]}/{file}')
 
-	change_perm = False
-	if history_logfile.exists() is False:
-		change_perm = True
+	change_perm = not path.exists()
 
 	try:
-		with history_logfile.open('a') as cmd_log:
-			cmd_log.write(f'{time.time()} {cmd}\n')
+		with path.open('a') as f:
+			f.write(content)
 
 		if change_perm:
-			history_logfile.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP)
+			path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP)
 	except (PermissionError, FileNotFoundError):
-		# If history_logfile does not exist, ignore the error
+		# If the file does not exist, ignore the error
 		pass
 
 
+def _cmd_history(cmd: list[str]) -> None:
+	content = f'{time.time()} {cmd}\n'
+	_append_log('cmd_history.txt', content)
+
+
+def _cmd_output(output: str) -> None:
+	_append_log('cmd_output.txt', output)
+
+
 def run(
 	cmd: list[str],
 	input_data: bytes | None = None,
 ) -> subprocess.CompletedProcess[bytes]:
-	_log_cmd(cmd)
+	_cmd_history(cmd)
 
 	return subprocess.run(
 		cmd,