162 lines
5.0 KiB
Python
162 lines
5.0 KiB
Python
import subprocess
|
|
import xml.etree.ElementTree as ET
|
|
from typing import Optional
|
|
|
|
from . import Entry
|
|
|
|
class BackgroundEntry(Entry):
|
|
"""
|
|
Entry type for long-running background processes.
|
|
|
|
Attributes:
|
|
script: The script/command to execute
|
|
stdout: Captured standard output
|
|
stderr: Captured standard error
|
|
process: The running subprocess.Popen instance
|
|
exit_code: Exit code when process completes
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
id: str,
|
|
script: str,
|
|
):
|
|
"""
|
|
Initialize a new background entry.
|
|
|
|
Args:
|
|
id: Unique identifier for this entry
|
|
script: The script/command to execute
|
|
"""
|
|
super().__init__(id)
|
|
self.script = script
|
|
self._process: Optional[subprocess.Popen] = None
|
|
self.stdout = ""
|
|
self.stderr = ""
|
|
self.exit_code = None
|
|
|
|
@property
|
|
def pid(self) -> Optional[int]:
|
|
"""Get the process ID (None if not running)."""
|
|
return self._process.pid if self._process is not None else None
|
|
|
|
def cleanup(self) -> None:
|
|
"""
|
|
Clean up the background process if it's still running.
|
|
Ensures process is terminated and file handles are closed.
|
|
"""
|
|
if self._process is not None:
|
|
try:
|
|
if self._process.stdout:
|
|
self._process.stdout.close()
|
|
if self._process.stderr:
|
|
self._process.stderr.close()
|
|
if self._process.poll() is None:
|
|
self._process.terminate()
|
|
try:
|
|
self._process.wait(timeout=1.0)
|
|
except subprocess.TimeoutExpired:
|
|
self._process.kill()
|
|
self._process.wait()
|
|
except:
|
|
pass # Ignore cleanup errors
|
|
finally:
|
|
self._process = None
|
|
|
|
def reset(self) -> None:
|
|
"""
|
|
Cleans up existing process and resets output buffers.
|
|
"""
|
|
self.cleanup()
|
|
self.stdout = ""
|
|
self.stderr = ""
|
|
self.exit_code = None
|
|
self.notify_change()
|
|
|
|
def update(self) -> None:
|
|
"""
|
|
Start the process if not running and collect any new output.
|
|
Updates stdout and stderr with any new output.
|
|
"""
|
|
if self._process is None and self.exit_code is None:
|
|
self._process = subprocess.Popen(
|
|
self.script,
|
|
shell=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1 # Line buffered
|
|
)
|
|
self.notify_change()
|
|
return
|
|
|
|
if self._process is None:
|
|
return
|
|
|
|
exit_code = self._process.poll()
|
|
if exit_code is not None:
|
|
try:
|
|
remaining_out, remaining_err = self._process.communicate(timeout=0.1)
|
|
self.stdout += remaining_out
|
|
self.stderr += remaining_err
|
|
except subprocess.TimeoutExpired:
|
|
pass # Process didn't finish communicating, try again next update
|
|
|
|
self.exit_code = exit_code
|
|
self.cleanup()
|
|
self.notify_change()
|
|
return
|
|
|
|
if self._process.stdout:
|
|
while True:
|
|
try:
|
|
line = self._process.stdout.readline()
|
|
if not line:
|
|
break
|
|
self.stdout += line
|
|
except:
|
|
break
|
|
|
|
if self._process.stderr:
|
|
while True:
|
|
try:
|
|
line = self._process.stderr.readline()
|
|
if not line:
|
|
break
|
|
self.stderr += line
|
|
except:
|
|
break
|
|
self.notify_change()
|
|
|
|
def generate_context(self) -> ET.Element:
|
|
"""
|
|
Generate an XML Element representing this background entry.
|
|
|
|
Returns:
|
|
ET.Element: XML element containing the entry's data
|
|
"""
|
|
element = ET.Element("background", {"id": self.id})
|
|
if self._process is not None:
|
|
element.set("pid", str(self._process.pid))
|
|
elif self.exit_code is not None:
|
|
element.set("exit_code", str(self.exit_code))
|
|
element.text = self.script
|
|
stdout_elem = ET.SubElement(element, "stdout")
|
|
stdout_elem.text = self.stdout
|
|
stderr_elem = ET.SubElement(element, "stderr")
|
|
stderr_elem.text = self.stderr
|
|
|
|
return element
|
|
|
|
def serialize(self):
|
|
"""
|
|
Collect all public attributes of the entry into a dictionary.
|
|
"""
|
|
return {
|
|
"type": "background",
|
|
"id": self.id,
|
|
"script": self.script,
|
|
"stdout": self.stdout,
|
|
"stderr": self.stderr,
|
|
"exit_code": self.exit_code,
|
|
} |