From 50f05b621384bb8005a35c7e071934c4963045cc Mon Sep 17 00:00:00 2001 From: geens Date: Wed, 13 Nov 2024 11:44:58 +0100 Subject: [PATCH] Better response parsing --- sia/response_parser.py | 84 ++++++++---------------------------------- 1 file changed, 16 insertions(+), 68 deletions(-) diff --git a/sia/response_parser.py b/sia/response_parser.py index 040c433..41dacac 100644 --- a/sia/response_parser.py +++ b/sia/response_parser.py @@ -36,46 +36,6 @@ class ResponseParser: io_buffer: Buffer to use for IO operations """ self.io_buffer = io_buffer - - def _extract_content(self, xml: str, root: ET.Element) -> str: - """ - Extract content from XML, handling CDATA sections and preserving special characters. - - Args: - xml: Original XML string - root: Parsed XML element - - Returns: - str: Extracted content with proper handling of CDATA and special characters - """ - if root.text is None: - return "" - - # Find opening tag end - tag_name = root.tag - tag_start = xml.find('<' + tag_name) - tag_end = xml.find('>', tag_start) - if tag_end == -1: - return "" - - # Check for self-closing tag - if xml[tag_end-1] == '/': - return "" - - # Find closing tag start - close_tag = '' - close_start = xml.rfind(close_tag) - if close_start == -1: - return "" - - # Extract content - content = xml[tag_end+1:close_start] - - # Handle CDATA sections - if content.startswith(''): - content = content[9:-3] # Remove CDATA markers - - return content def parse(self, xml: str) -> Union[Command, Entry]: """ @@ -91,61 +51,49 @@ class ResponseParser: entry_id = str(uuid.uuid4()) timestamp = datetime.now() - # First try to parse the XML try: root = ET.fromstring(xml.strip()) except ET.ParseError as e: return ParseErrorEntry(xml, f"Invalid XML: {str(e)}", entry_id, timestamp) - - # Extract content preserving special characters and handling CDATA - content = self._extract_content(xml, root) - # Convert to appropriate type based on root tag try: if root.tag == 'delete': - # Extract target ID and create delete command target_id = root.get('id') if not target_id: return ParseErrorEntry(xml, "Delete command missing required 'id' attribute", entry_id, timestamp) return DeleteCommand(target_id) elif root.tag == 'stop': - # Create stop command return StopCommand() elif root.tag == 'background': - # Create background script entry - if not content: - return ParseErrorEntry(xml, "Background entry missing script content", entry_id, timestamp) - return BackgroundEntry(content, entry_id, timestamp) + if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': + return ParseErrorEntry(xml, "Background entry requires (only) script content", entry_id, timestamp) + return ParseErrorEntry(xml, "Background entry requires script content", entry_id, timestamp) + return BackgroundEntry(root.text, entry_id, timestamp) elif root.tag == 'repeat': - # Create repeat script entry - if not content: - return ParseErrorEntry(xml, "Repeat entry missing script content", entry_id, timestamp) - return RepeatEntry(content, entry_id, timestamp) + if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': + return ParseErrorEntry(xml, "Repeat entry requires (only) script content", entry_id, timestamp) + return RepeatEntry(root.text, entry_id, timestamp) elif root.tag == 'single': - # Create single shot script entry - if not content: - return ParseErrorEntry(xml, "Single shot entry missing script content", entry_id, timestamp) - return SingleEntry(content, entry_id, timestamp) + if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': + return ParseErrorEntry(xml, "Single entry requires (only) script content", entry_id, timestamp) + return SingleEntry(root.text, entry_id, timestamp) elif root.tag == 'reasoning': - # Create reasoning entry - if not content: - return ParseErrorEntry(xml, "Reasoning entry missing content", entry_id, timestamp) - return ReasoningEntry(content, entry_id, timestamp) + if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': + return ParseErrorEntry(xml, "Reasoning entry requires (only) text content", entry_id, timestamp) + return ReasoningEntry(root.text, entry_id, timestamp) elif root.tag == 'read_stdin': - # Create read entry - no content required return ReadEntry(self.io_buffer, entry_id, timestamp) elif root.tag == 'write_stdout': - # Create write entry - if not content: - return ParseErrorEntry(xml, "Write entry missing content", entry_id, timestamp) - return WriteEntry(content, self.io_buffer, entry_id, timestamp) + if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': + return ParseErrorEntry(xml, "Write stdout entry requires (only) text content", entry_id, timestamp) + return WriteEntry(root.text, self.io_buffer, entry_id, timestamp) else: return ParseErrorEntry(xml, f"Unknown root element: {root.tag}", entry_id, timestamp)