From 2c6da58eb7b4b014062e9b6c8922af2b07fe3d25 Mon Sep 17 00:00:00 2001 From: smypmsa Date: Mon, 4 Aug 2025 05:02:53 +0000 Subject: [PATCH] fix(core): fix event idl parser for logs --- src/platforms/pumpfun/event_parser.py | 195 ++++++++++++++++---------- src/utils/idl_parser.py | 82 +++++++---- 2 files changed, 180 insertions(+), 97 deletions(-) diff --git a/src/platforms/pumpfun/event_parser.py b/src/platforms/pumpfun/event_parser.py index af5417f..b3e1405 100644 --- a/src/platforms/pumpfun/event_parser.py +++ b/src/platforms/pumpfun/event_parser.py @@ -55,7 +55,7 @@ class PumpFunEventParser(EventParser): logs: list[str], signature: str ) -> TokenInfo | None: - """Parse token creation from pump.fun transaction logs using IDL instruction parsing. + """Parse token creation from pump.fun transaction logs using IDL event parsing. Args: logs: List of log strings from transaction @@ -72,88 +72,127 @@ class PumpFunEventParser(EventParser): if any("Program log: Instruction: CreateTokenAccount" in log for log in logs): return None + logger.info(f"๐Ÿ” Parsing token creation from logs for signature: {signature}") + # Look for event data in the logs (CreateEvent data!) + # We need to find the Program data that comes after "Instruction: Create" try: - for log in logs: - if "Program data:" in log: + create_instruction_found = False + program_data_entries = [] + + # First, collect all Program data entries and note when Create instruction happens + for i, log in enumerate(logs): + if "Program log: Instruction: Create" in log: + create_instruction_found = True + logger.info(f"๐Ÿ“ Found Create instruction at log index {i}") + elif "Program data:" in log: + # Extract base64 encoded event data + encoded_data = log.split("Program data: ")[1].strip() + program_data_entries.append((i, encoded_data, log)) + logger.info(f"๐Ÿ“Š Found Program data at log index {i}, length: {len(encoded_data)}") + + if not create_instruction_found: + logger.info("โŒ No Create instruction found in logs") + return None + + if not program_data_entries: + logger.info("โŒ No Program data entries found in logs") + return None + + logger.info(f"๐Ÿ” Found {len(program_data_entries)} Program data entries to check") + + # Try each Program data entry + for entry_idx, (log_idx, encoded_data, full_log) in enumerate(program_data_entries): + try: + logger.info(f"๐Ÿงช Trying Program data entry {entry_idx + 1}/{len(program_data_entries)} (log index {log_idx})") + + decoded_data = base64.b64decode(encoded_data) + + if len(decoded_data) < 8: + logger.info(f"โš ๏ธ Program data too short: {len(decoded_data)} bytes") + continue + + # Check discriminator from program data + discriminator = decoded_data[:8] + discriminator_int = struct.unpack(" bool: + """Check if verbose logging is enabled.""" + return getattr(self, '_verbose', False) + + @verbose.setter + def verbose(self, value: bool) -> None: + """Set verbose logging.""" + self._verbose = value \ No newline at end of file diff --git a/src/utils/idl_parser.py b/src/utils/idl_parser.py index ffcc272..2154b9d 100644 --- a/src/utils/idl_parser.py +++ b/src/utils/idl_parser.py @@ -163,44 +163,78 @@ class IDLParser: discriminator = event_data[:DISCRIMINATOR_SIZE] - # If event_name provided, validate it matches the discriminator - if event_name: - event_discriminators = self.get_event_discriminators() - if event_name not in event_discriminators: - if self.verbose: - print(f"Unknown event name: {event_name}") - return None - if event_discriminators[event_name] != discriminator: - if self.verbose: - print(f"Event discriminator mismatch for {event_name}") - return None - event_def = self.events[discriminator] - else: - # Try to find event by discriminator - if discriminator not in self.events: - if self.verbose: - print(f"Unknown event discriminator: {discriminator.hex()}") - return None - event_def = self.events[discriminator] + # Find event definition by discriminator in events section + if discriminator not in self.events: + if self.verbose: + print(f"Unknown event discriminator: {discriminator.hex()}") + return None + + event_def = self.events[discriminator] + + # If event_name provided, validate it matches + if event_name and event_def['name'] != event_name: + if self.verbose: + print(f"Event name mismatch: expected {event_name}, got {event_def['name']}") + return None + # Get the actual structure definition from types section + event_name_actual = event_def['name'] + if event_name_actual not in self.types: + if self.verbose: + print(f"Event type {event_name_actual} not found in types section") + return None + + type_def = self.types[event_name_actual] + event_type = type_def.get('type', {}) + # Decode event fields try: event_fields = {} data_part = event_data[DISCRIMINATOR_SIZE:] decode_offset = 0 - for field in event_def.get('fields', []): - value, decode_offset = self._decode_type(data_part, decode_offset, field['type']) - event_fields[field['name']] = value + if event_type.get('kind') != 'struct': + if self.verbose: + print(f"Event {event_name_actual} is not a struct type: {event_type.get('kind', 'NO KIND')}") + print(f"Available keys in type_def: {list(type_def.keys())}") + print(f"Event type structure: {event_type}") + return None + + # Decode each field in the struct + fields = event_type.get('fields', []) + if self.verbose: + print(f"Decoding {len(fields)} fields for event {event_name_actual}") + + for field in fields: + if self.verbose: + print(f"Decoding field: {field['name']} ({field['type']})") + + try: + value, decode_offset = self._decode_type(data_part, decode_offset, field['type']) + event_fields[field['name']] = value + + if self.verbose: + if field['type'] == 'string': + print(f" -> '{value}'") + elif field['type'] == 'pubkey': + print(f" -> {value}") + else: + print(f" -> {value}") + + except Exception as e: + if self.verbose: + print(f"Error decoding field {field['name']}: {e}") + # Don't return None here, continue with other fields + continue return { - 'event_name': event_def['name'], + 'event_name': event_name_actual, 'fields': event_fields } except Exception as e: if self.verbose: - print(f"โŒ Error decoding event {event_def['name']}: {e}") + print(f"โŒ Error decoding event {event_name_actual}: {e}") return None def find_event_in_logs(self, logs: list[str], target_event_name: str | None = None) -> dict[str, Any] | None: