mirror of
https://codeberg.org/vlw/3rd.git
synced 2026-01-11 22:36:01 +01:00
104 lines
2.1 KiB
Python
104 lines
2.1 KiB
Python
import os
|
|
import uuid
|
|
import tempfile
|
|
import subprocess
|
|
from typing import Union
|
|
|
|
from .Stdout import Stdout
|
|
from .Enums import Namespace, StdoutLevel
|
|
|
|
class Cli():
|
|
def __init__(self):
|
|
"""
|
|
Create a new Cli instance
|
|
"""
|
|
|
|
self.id = str(uuid.uuid4())
|
|
self.__stdout = Stdout(Namespace.CLI)
|
|
|
|
@property
|
|
def stdout(self) -> str | None:
|
|
"""
|
|
Returns the contents of stdout if exists
|
|
|
|
Returns:
|
|
str | None: Stdout contents or None
|
|
"""
|
|
|
|
# Bail out if stdout file does not exist
|
|
if not os.path.isfile(self.__path_stdout):
|
|
return None
|
|
|
|
with open(self.__path_stdout, "r", encoding="utf-8") as f:
|
|
contents = f.read().strip()
|
|
return contents if contents else None
|
|
|
|
@property
|
|
def stderr(self) -> str | None:
|
|
"""
|
|
Returns the contents of stderr if exists
|
|
|
|
Returns:
|
|
str | None: Stderr contents or None
|
|
"""
|
|
|
|
# Bail out if stderr file does not exist
|
|
if not os.path.isfile(self.__path_stderr):
|
|
return None
|
|
|
|
with open(self.__path_stderr, "r", encoding="utf-8") as f:
|
|
contents = f.read().strip()
|
|
return contents if contents else None
|
|
|
|
@property
|
|
def __path_stdout(self) -> str:
|
|
"""
|
|
Returns the path to the standard output file
|
|
|
|
Returns:
|
|
str: Pathname to stdout file
|
|
"""
|
|
|
|
return f"{tempfile.gettempdir().rstrip('/')}/{self.id}_stdout.txt"
|
|
|
|
@property
|
|
def __path_stderr(self) -> str:
|
|
"""
|
|
Returns the path to the standard error file
|
|
|
|
Returns:
|
|
str: Pathname to stderr file
|
|
"""
|
|
|
|
return f"{tempfile.gettempdir().rstrip('/')}/{self.id}_stderr.txt"
|
|
|
|
def cleanup(self) -> None:
|
|
"""
|
|
Remove temporary files
|
|
"""
|
|
|
|
os.remove(self.__path_stdout)
|
|
os.remove(self.__path_stderr)
|
|
|
|
def run(self, args: list) -> None:
|
|
"""
|
|
Run a command
|
|
|
|
Args:
|
|
cmd (list): Subprocess run commands
|
|
"""
|
|
|
|
cmd = ' '.join(args)
|
|
|
|
self.__stdout.info(f"Running command: {cmd}").sleep()
|
|
|
|
with open(self.__path_stdout, "w") as stdout, open(self.__path_stderr, "w") as stderr:
|
|
subprocess.run(args, stdout=stdout, stderr=stderr)
|
|
|
|
self.__stdout.debug(f"{cmd}:\n{self.stdout}")
|
|
|
|
if self.stderr:
|
|
self.__stdout.error(f"{cmd}:\n{self.stderr}")
|
|
return
|
|
|
|
self.__stdout.info(f"Exit 0: {cmd}")
|