from datetime import datetime import subprocess import xml.etree.ElementTree as ET from typing import Optional from .entry import Entry class RepeatEntry(Entry): """ Entry type for scripts that are executed on every update. Attributes: script: The script/command to execute timeout: Maximum time to wait for script execution limit: Maximum number of characters to capture from stdout/stderr 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 """ default_timeout = 1 default_limit = 1024 def __init__( self, id: str, timestamp: datetime, script: str, timeout: Optional[float], limit: Optional[int], ): """ Initialize a new repeat entry. Args: id: Unique identifier for this entry timestamp: Creation timestamp for this entry script: The script/command to execute timeout: Maximum time to wait for script execution limit: Maximum number of characters to capture from stdout/stderr """ super().__init__(id, timestamp) self._script = script self._timeout = timeout self._limit = limit self._stdout = "" self._stderr = "" self._exit_code: Optional[int] = None self._timed_out = 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 and update the output. Captures stdout, stderr and exit code from each execution. """ try: process = subprocess.run( self._script, timeout=(self._timeout or self.default_timeout), shell=True, capture_output=True, text=True ) self._stdout = process.stdout self._stderr = process.stderr self._exit_code = process.returncode except subprocess.TimeoutExpired as e: self._timed_out = True def generate_context(self) -> ET.Element: """ Generate an XML Element representing this repeat entry. Returns: ET.Element: XML element containing the entry's data """ element = ET.Element("repeat", { "id": self.id, }) if self._timeout: element.set("timeout", str(self._timeout)) element.text = self.script if self._timed_out: element.set("timed_out", "true") elif self._exit_code is not None: element.set("exit_code", str(self._exit_code)) if self._limit: element.set("limit", str(self._limit)) if len(self._stdout) > (self._limit or self.default_limit): element.set("stdout_truncated", "true") element.set("stdout_length", str(len(self._stdout))) stdout_elem = ET.SubElement(element, "stdout") stdout_elem.text = self._stdout[:(self._limit or self.default_limit)] if len(self._stderr) > (self._limit or self.default_limit): element.set("stderr_truncated", "true") element.set("stderr_length", str(len(self._stderr))) stderr_elem = ET.SubElement(element, "stderr") stderr_elem.text = self._stderr[:(self._limit or self.default_limit)] return element