Analisador Avançado de Arquivo de Log para um Formato Personalizado

def parse_log(log_content: str) -> list: lines = log_content.split('\n') transactions = {} active_transaction = None completed_transactions = [] for line in lines: line = line.strip() # Skip empty lines and comments if not line or line.startswith('COMMENT') or line.startswith('#'): continue # Parse START line if line.startswith('START '): parts = line.split(' ', 2) if len(parts) >= 3: transaction_id = parts[1] timestamp = parts[2] # If there's an active transaction, it's abandoned if active_transaction is not None: if active_transaction in transactions: del transactions[active_transaction] # Start new transaction active_transaction = transaction_id transactions[transaction_id] = { 'transaction_id': transaction_id, 'start_time': timestamp, 'end_time': None, 'status': None, 'events': [] } continue # Parse END line if line.startswith('END '): parts = line.split(' ', 3) if len(parts) >= 4: transaction_id = parts[1] status = parts[2] timestamp = parts[3] # Only process if this transaction exists and is active if transaction_id in transactions and active_transaction == transaction_id: transactions[transaction_id]['end_time'] = timestamp transactions[transaction_id]['status'] = status completed_transactions.append(transactions[transaction_id]) active_transaction = None continue # Parse EVENT line if line.startswith('EVENT '): if active_transaction is not None and active_transaction in transactions: event_content = line[6:] # Remove 'EVENT ' prefix event_dict = parse_event(event_content) if event_dict: transactions[active_transaction]['events'].append(event_dict) continue return completed_transactions def parse_event(event_content: str) -> dict: event_dict = {} i = 0 while i < len(event_content): # Skip whitespace while i < len(event_content) and event_content[i] == ' ': i += 1 if i >= len(event_content): break # Find key key_start = i while i < len(event_content) and event_content[i] not in '= ': i += 1 key = event_content[key_start:i] # Skip whitespace and '=' while i < len(event_content) and event_content[i] in ' =': i += 1 if i >= len(event_content): break # Parse value if event_content[i] == '"': # Quoted value i += 1 value_start = i while i < len(event_content) and event_content[i] != '"': i += 1 value = event_content[value_start:i] if i < len(event_content): i += 1 # Skip closing quote else: # Unquoted value value_start = i while i < len(event_content) and event_content[i] != ' ': i += 1 value = event_content[value_start:i] event_dict[key] = value return event_dict

import re from typing import List, Dict, Optional def parse_log(log_content: str) -> list: """Parse custom log content and return a list of completed transactions. Rules implemented: - Processes sequentially. - Maintains at most one active transaction (most recently started). - START while active abandons previous. - EVENT outside active is ignored. - Only START+matching END produces an output transaction. - COMMENT/blank/malformed lines are ignored. """ # Basic ISO-8601-ish token (kept permissive but requires no spaces) start_re = re.compile(r'^START\s+(\S+)\s+(\S+)\s*$') end_re = re.compile(r'^END\s+(\S+)\s+(\S+)\s+(\S+)\s*$') comment_re = re.compile(r'^COMMENT\s+#.*$') # EVENT key/value parsing: key=value or key="value with spaces" # Values may be empty. Quoted values can contain escaped quotes (\") and escaped backslashes. event_kv_re = re.compile( r'(\S+?)=' # key (no spaces, minimal up to '=') r'(?:"((?:\\.|[^"\\])*)"' r'|([^\s]*))' # or unquoted value up to whitespace (can be empty) ) def unescape_quoted(s: str) -> str: # Interpret backslash escapes for \\ and \" and common sequences # without raising on unknown; keeps unknown as literal after backslash. out = [] i = 0 while i < len(s): c = s[i] if c == '\\' and i + 1 < len(s): nxt = s[i + 1] if nxt in ['\\', '"']: out.append(nxt) elif nxt == 'n': out.append('\n') elif nxt == 't': out.append('\t') elif nxt == 'r': out.append('\r') else: out.append(nxt) i += 2 else: out.append(c) i += 1 return ''.join(out) def parse_event_payload(payload: str) -> Optional[Dict[str, str]]: # Parse the payload as a sequence of key=value tokens, where quoted values may contain spaces. # Malformed if any non-whitespace remains unmatched. pos = 0 n = len(payload) data: Dict[str, str] = {} while True: while pos < n and payload[pos].isspace(): pos += 1 if pos >= n: break m = event_kv_re.match(payload, pos) if not m: return None key = m.group(1) qv = m.group(2) uv = m.group(3) if qv is not None: val = unescape_quoted(qv) else: val = uv if uv is not None else '' data[key] = val pos = m.end() return data if data else None completed: List[Dict] = [] active: Optional[Dict] = None # {'transaction_id', 'start_time', 'events'} for raw_line in log_content.splitlines(): line = raw_line.strip('\n') stripped = line.strip() if not stripped: continue if comment_re.match(stripped): continue m = start_re.match(stripped) if m: # Abandon any currently active transaction tid, ts = m.group(1), m.group(2) active = { 'transaction_id': tid, 'start_time': ts, 'events': [] } continue m = end_re.match(stripped) if m: tid, status, ts = m.group(1), m.group(2), m.group(3) if active is not None and active.get('transaction_id') == tid: completed.append({ 'transaction_id': tid, 'start_time': active.get('start_time'), 'end_time': ts, 'status': status, 'events': active.get('events', []) }) active = None # If END doesn't match an active transaction, ignore. continue if stripped.startswith('EVENT'): if active is None: continue # Require at least one whitespace after EVENT if stripped == 'EVENT': continue if not stripped.startswith('EVENT '): continue payload = stripped[len('EVENT '):] event_dict = parse_event_payload(payload) if event_dict is None: continue active['events'].append(event_dict) continue # Any other malformed/unknown line is ignored. continue # Any active transaction at EOF is abandoned and discarded. return completed