Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Python code to parse the results of a Defi Kingdoms quest
#! /usr/bin/env python
from copy import deepcopy
import logging
import sys
from web3 import Web3
from web3.logs import DISCARD
from dfktools.quests.quest_core_v1 import CONTRACT_ADDRESS as QUEST_V1_CONTRACT_ADDRESS
from dfktools.quests.quest_core_v1 import ABI as QUEST_V1_ABI
from dfktools.quests.quest_core_v2 import CONTRACT_ADDRESS as QUEST_V2_CONTRACT_ADDRESS
from dfktools.quests.quest_core_v2 import ABI as QUEST_V2_ABI
import dfktools.quests.quest_v2 as quest_v2
import dfktools.quests.quest_v1 as quest_v1
from dfktools.dex import erc20
JEWEL_ADDR = erc20.symbol2address('JEWEL')
GOLD_ADDR = erc20.symbol2address('DFKGOLD')
ZERO_ADDRESS = '0x0000000000000000000000000000000000000000'
logger = logging.getLogger()
questV1 = quest_v1.Quest(RPC_SERVER, logger)
questV2 = quest_v2.Quest(RPC_SERVER, logger)
# =========
# = TYPES =
# =========
from typing import Dict
HexAddress = str
QuestResult = Dict
TxReceipt = Dict
def main():
# example tx if you need one
# tx_id = "0x3c99384a4895e9a89f3ee241409650876c9d8128e52a7cd93d5cacca6ccd219a"
tx_id = sys.argv[1]
except Exception as e:
print('Usage: QUEST_TX_HASH')
results = get_quest_results(tx_id)
table = quest_reward_table(results)
def get_quest_results(tx_id: HexAddress) -> QuestResult:
w3 = Web3(Web3.HTTPProvider(RPC_SERVER))
tx_receipt = w3.eth.get_transaction_receipt(tx_id)
results = parse_ended_quest_receipt(tx_receipt)
return results
def parse_ended_quest_receipt(tx_receipt:TxReceipt) -> QuestResult:
result: QuestResult = {}
hero_dicts: Dict = {}
if not tx_receipt:
return result
contract_address =
is_quest_v2 = (contract_address == QUEST_V2_CONTRACT_ADDRESS)
if is_quest_v2:
contract_abi = QUEST_V2_ABI
amount_key, reward_key = 'amount', 'reward'
contract_abi = QUEST_V1_ABI
amount_key, reward_key = 'itemQuantity', 'rewardItem'
w3 = Web3(Web3.HTTPProvider(RPC_SERVER))
contract_address = Web3.toChecksumAddress(contract_address)
contract = w3.eth.contract(contract_address, abi=contract_abi)
# NOTE: calling processReceipt without 'errors = web3.logs.DISCARD' caused
# lots of MismatchedABI errors. Seems like we get the data we need without it, though
if is_quest_v2:
quest_reward =, errors=DISCARD)
quest_reward =, errors=DISCARD)
quest_skillup =, errors=DISCARD)
quest_xp =, errors=DISCARD)
default_dict = {'hero_id':0, 'rewards':[], 'skill_up':0, 'xp':0}
# NOTE: In v2 quests, we can't count on all of these events being
# triggered in each quest, so we have to make sure that each hero_dicts[hero_id]
# dict is instantiated separately; that's why the `hero_dicts.setdefault()` code
# is duplicated for each event.
# (Note deepcopy, so that the 'rewards' array isn't shared between copies)
# - Athiriyya 07 May 2022
for qr in quest_reward:
hero_id = qr.args.heroId
hero_dicts.setdefault(hero_id, deepcopy(default_dict))
hero_dicts[hero_id]['hero_id'] = hero_id
reward_address = qr.args[reward_key]
reward_str = label_for_reward(reward_address, qr.args[amount_key])
# NOTE: Looks like there are a lot of items with null addresses; not sure what's
# happening there. Ignore for now 05 April 2022
if reward_address != ZERO_ADDRESS:
for qs in quest_skillup:
hero_id = qs.args.heroId
hero_dicts.setdefault(hero_id, deepcopy(default_dict))
hero_dicts[hero_id]['hero_id'] = hero_id
hero_dicts[hero_id]['skill_up'] += qs.args.skillUp
for qx in quest_xp:
hero_id = qx.args.heroId
hero_dicts.setdefault(hero_id, deepcopy(default_dict))
hero_dicts[hero_id]['hero_id'] = hero_id
hero_dicts[hero_id]['xp'] += qx.args.xpEarned
# each entry in quest_xp contains this info; only set it if it hasn't
# been set yet
if not result:
result['quest_id'] = qx.args.questId
result['address'] = qx.args.player
result['transaction_hash'] = qx.transactionHash.hex()
result['rewards'] = list(hero_dicts.values())
# TODO: We'd also like to know:
# - quest type (jewel mining, gold mining, foraging, etc)
# - liquidity pool id for gardening quests
# - time quest ended: maybe derivable from quest_reward.blockNumber?
return result
def label_for_reward(reward_address:HexAddress, item_quantity:str) -> str:
item = erc20.address2item(reward_address)
if item:
obj_name = item[2]
obj_name = f'Unknown item ({reward_address})'
if reward_address == GOLD_ADDR:
amount = float(item_quantity) / 1e3
amount_str = '%0.3f'%amount
elif reward_address == JEWEL_ADDR:
amount = float(item_quantity) / 1e18
amount_str = '%0.3f'%amount
amount = float(item_quantity)
amount_str = '%.0f'%amount
label = f'{amount_str} {obj_name}'
return label
def quest_reward_table(quest_result:QuestResult) -> str:
from prettytable import PrettyTable
except ImportError as e:
msg = f'\nUnable to import the "prettytable" module. Install with `pip install prettytable` to print quest results.\n'
return msg
# make a table for all the rewards given by a quest.
columns = ['quest_id', 'quest_type', 'pool_id', 'hero_id', 'xp', 'skill_up', 'rewards']
no_rewards_cols = columns[:-1]
empty_row = ['' for c in no_rewards_cols]
table = PrettyTable(field_names=columns)
rows = []
for hero_dict in quest_result['rewards']:
rewards_list = hero_dict['rewards']
r = [hero_dict.get(k,'') for k in no_rewards_cols]
r.append(rewards_list[0] if rewards_list else '')
for item in rewards_list[1:]:
rows.append(empty_row + [item])
# Some info is only in the top level of the quest_result dictionary; put
# that in the first line
top_level_fields = [quest_result.get(k, '') for k in columns[:3]]
rows[0][:3] = top_level_fields
return table.get_string()
if __name__ == '__main__':
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment