Python code to parse the results of a Defi Kingdoms quest

parse_dfk_quest_results.py

#! /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'

GAS_PRICE_GWEI = 30
TX_TIMEOUT = 30
RPC_SERVER = 'https://api.harmony.one'

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():
    try:
        # example tx if you need one
        # tx_id = "0x3c99384a4895e9a89f3ee241409650876c9d8128e52a7cd93d5cacca6ccd219a"
        tx_id = sys.argv[1]
    except Exception as e:
        print('Usage: parse_quest_results.py QUEST_TX_HASH')
        sys.exit(1)
    results = get_quest_results(tx_id)
    print(results)

    table = quest_reward_table(results)
    print(table)


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 = tx_receipt.to
    is_quest_v2 = (contract_address == QUEST_V2_CONTRACT_ADDRESS)

    if is_quest_v2:
        contract_abi = QUEST_V2_ABI
        amount_key, reward_key = 'amount', 'reward'
    else:
        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 = contract.events.RewardMinted().processReceipt(tx_receipt, errors=DISCARD)
    else:
        quest_reward = contract.events.QuestReward().processReceipt(tx_receipt, errors=DISCARD)
    quest_skillup = contract.events.QuestSkillUp().processReceipt(tx_receipt, errors=DISCARD)
    quest_xp = contract.events.QuestXP().processReceipt(tx_receipt, 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:
            hero_dicts[hero_id]['rewards'].append(reward_str)
        
    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]
    else: 
        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
    else:
        amount = float(item_quantity)
        amount_str = '%.0f'%amount
    label = f'{amount_str} {obj_name}'
    return label

def quest_reward_table(quest_result:QuestResult) -> str:
    try:
        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 '')
        rows.append(r)
        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

    table.add_rows(rows)
    return table.get_string()
  
if __name__ == '__main__':
    main()