popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

parse transaction wip

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-05 19:15:00 +03:00
parent 1eeaec6e84
commit 42c290fbc6
1 changed files with 32 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
crypto/sol/app.py
View File

@ -619,41 +619,49 @@ async def process_log(log_result):
except Exception as e:
logging.error(f"Error processing log: {e}")
def parse_swap_logs(logs):
swap_details = {
"source_token_address": "",
"destination_token_address": "",
"amount_in": 0,
"amount_out": 0
}
total_amount_in = 0
total_amount_out = 0
source_token_address = ""
destination_token_address = ""
for log in logs:
if "SwapEvent" in log:
# Extract amounts from SwapEvent
parts = log.split("amount_in: ")[1].split(", amount_out: ")
swap_details["amount_in"] = int(parts[0])
swap_details["amount_out"] = int(parts[1].split(" ")[0])
parts = log.split("{ ")[1].strip(" }").split(", ")
event_details = {}
for part in parts:
key, value = part.split(": ")
event_details[key.strip()] = value.strip()
# Aggregate amounts
total_amount_in += int(event_details.get("amount_in", 0))
total_amount_out += int(event_details.get("amount_out", 0))
if "source_token_change:" in log:
# Extract source and destination token changes
# Extract final source and destination token addresses
changes = log.split(", ")
for change in changes:
for change in changes.Trim('Program log:'):
key, value = change.split(": ")
if key == "source_token_change":
swap_details["amount_in"] = int(value)
total_amount_in = int(value)
elif key == "destination_token_change":
swap_details["amount_out"] = int(value)
total_amount_out = int(value)
if "Program log:" in log and len(log.split()) == 2:
# Extract token addresses (assuming they are logged as single entries)
# Extract token addresses
token_address = log.split(": ")[1]
if not swap_details["source_token_address"]:
swap_details["source_token_address"] = token_address
elif not swap_details["destination_token_address"]:
swap_details["destination_token_address"] = token_address
return swap_details
if not source_token_address:
source_token_address = token_address
elif not destination_token_address:
destination_token_address = token_address
return {
"source_token_address": source_token_address,
"destination_token_address": destination_token_address,
"total_amount_in": total_amount_in,
"total_amount_out": total_amount_out,
}
async def on_logs(log):
logging.debug(f"Received log: {log}")