121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
from datetime import datetime
|
|
import subprocess
|
|
import xml.etree.ElementTree as ET
|
|
from typing import Optional
|
|
|
|
from .entry import Entry
|
|
from .util import escape_text_for_xml
|
|
|
|
class SingleShotEntry(Entry):
|
|
"""
|
|
Entry type for one-time script executions.
|
|
|
|
Attributes:
|
|
script: The script/command to execute
|
|
stdout: Captured standard output from script execution
|
|
stderr: Captured standard error from script execution
|
|
exit_code: Process exit code after completion
|
|
_executed: Whether the script has been executed
|
|
"""
|
|
|
|
def __init__(self, script: str, id: str, timestamp: datetime):
|
|
"""
|
|
Initialize a new single shot entry.
|
|
|
|
Args:
|
|
script: The script/command to execute
|
|
id: Unique identifier for this entry
|
|
timestamp: Creation timestamp for this entry
|
|
"""
|
|
super().__init__(id, timestamp)
|
|
self._script = script
|
|
self._stdout = ""
|
|
self._stderr = ""
|
|
self._exit_code: Optional[int] = None
|
|
self._executed = False
|
|
|
|
@property
|
|
def script(self) -> str:
|
|
"""Get the script/command being executed."""
|
|
return self._script
|
|
|
|
@property
|
|
def stdout(self) -> str:
|
|
"""Get the captured standard output."""
|
|
return self._stdout
|
|
|
|
@property
|
|
def stderr(self) -> str:
|
|
"""Get the captured standard error."""
|
|
return self._stderr
|
|
|
|
@property
|
|
def exit_code(self) -> Optional[int]:
|
|
"""Get the exit code of the process (None if still running)."""
|
|
return self._exit_code
|
|
|
|
def update(self) -> None:
|
|
"""
|
|
Execute the script if not already executed.
|
|
Captures stdout, stderr and exit code.
|
|
"""
|
|
if self._executed:
|
|
return
|
|
|
|
try:
|
|
# Run script and capture output
|
|
process = subprocess.run(
|
|
self.script,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
# Store results
|
|
self._stdout = process.stdout
|
|
self._stderr = process.stderr
|
|
self._exit_code = process.returncode
|
|
|
|
except subprocess.SubprocessError as e:
|
|
# Handle subprocess errors by capturing the error message
|
|
self._stdout = ""
|
|
self._stderr = str(e)
|
|
self._exit_code = -1
|
|
|
|
except Exception as e:
|
|
# Handle any other errors
|
|
self._stdout = ""
|
|
self._stderr = f"Error executing script: {str(e)}"
|
|
self._exit_code = -1
|
|
|
|
self._executed = True
|
|
|
|
def generate_context(self) -> ET.Element:
|
|
"""
|
|
Generate an XML Element representing this single shot entry.
|
|
|
|
Returns:
|
|
ET.Element: XML element containing the entry's data
|
|
"""
|
|
# Create root element
|
|
element = ET.Element("single_shot", {
|
|
"id": self.id,
|
|
})
|
|
|
|
# Add script as CDATA or escaped text
|
|
element.text = escape_text_for_xml(self.script)
|
|
|
|
# Only add output elements if script has executed
|
|
if self._executed:
|
|
# Add stdout element if there is output
|
|
stdout_elem = ET.SubElement(element, "stdout")
|
|
stdout_elem.text = escape_text_for_xml(self._stdout)
|
|
|
|
# Add stderr element if there are errors
|
|
stderr_elem = ET.SubElement(element, "stderr")
|
|
stderr_elem.text = escape_text_for_xml(self._stderr)
|
|
|
|
# Add exit code attribute after execution
|
|
element.set("exit_code", str(self._exit_code))
|
|
|
|
return element |