Parse a single log line.
(self, line: str)
| 177 | return iteration |
| 178 | |
| 179 | def parse_line(self, line: str) -> Optional[ParsedEvent]: |
| 180 | """Parse a single log line.""" |
| 181 | self.workflow.raw_lines.append(line) |
| 182 | |
| 183 | # Check for structured event |
| 184 | if EVENT_START in line and EVENT_END in line: |
| 185 | try: |
| 186 | start = line.index(EVENT_START) + len(EVENT_START) |
| 187 | end = line.index(EVENT_END) |
| 188 | event_json = line[start:end] |
| 189 | event_data = json.loads(event_json) |
| 190 | event = ParsedEvent( |
| 191 | type=event_data.get("type", "unknown"), |
| 192 | timestamp=event_data.get("timestamp", ""), |
| 193 | data=event_data.get("data", {}), |
| 194 | raw_line=line, |
| 195 | ) |
| 196 | self._process_event(event) |
| 197 | return event |
| 198 | except (json.JSONDecodeError, ValueError): |
| 199 | pass |
| 200 | |
| 201 | return None |
| 202 | |
| 203 | def _process_event(self, event: ParsedEvent): |
| 204 | """Process a parsed event and update workflow state.""" |
no test coverage detected