A script to fetch wax mining events on all lands owned by a specified address, to help with distributing rewards or aggregating statistics.

fetch_wax_mining_events.py

# Wax api Alienworlds Mining aggregator by Vyryn. Licensed under the GNU AGPLv3.
# https://www.gnu.org/licenses/agpl-3.0.en.html
import asyncio
import json
from datetime import datetime
# pip install aiohttp
import aiohttp
# pip install parsedatetime
import parsedatetime as pdt

# Change this to the account that holds your lands
landowner = 'monkeymining'
# Change this to the earliest date/time to collect data from. Most plaintext interpretations should work,
# such as "an hour ago" or "yesterday", but I recommend using absolute dates like 1/20/2021.
start_time = 'ten minutes ago'
# Change this to the latest date/time to collect data from.
end_time = 'now'
# This command will display the top this many miners in the console. The whole list will be put into the output file
# below
top_n = 10
# The name of the output file
filename = 'temp_file.json'
# If set to False, the script will not save data to an external file and you'll only have the displayed list.
save_to_file = True

# You can change this, though there's probably no reason to
wax_mining_api = 'https://api.waxsweden.org/v2/history/get_actions?acount=m.federation&sort=1&filter=m.federation' \
                 ':logmine'


def dt(raw: str) -> datetime:
    """Converts a datetime formatted string into a datetime"""
    raw = raw.replace('T', ' ')
    try:
        return datetime.strptime(raw, '%Y-%m-%d %H:%M:%S.%f')
    except ValueError:
        return datetime.strptime(raw, '%Y-%m-%d %H:%M:%S')


async def topminers(top: int = 10, start: str = 'yesterday', end: str = 'today', save_data: bool = True):
    """Gets the top miners for the chosen time period"""
    miners = dict()
    cal = pdt.Calendar()
    now_ = datetime.utcnow()
    after = str(cal.parseDT(start, now_)[0]).replace(' ', 'T')
    original_after = after
    before = str(cal.parseDT(end, now_)[0]).replace(' ', 'T')
    print('Fetching data.')
    remaining_seconds = 1E10
    counter = 0
    async with aiohttp.ClientSession() as session:
        while remaining_seconds > 10:
            counter += 1
            api = wax_mining_api + f'&limit=1000&after={after}&before={before}'
            try:
                async with session.get(api) as resp:
                    if resp.status != 200:
                        return print(f"API request returned error code {resp.status}.")
                    response = await resp.json()
                    actions = [i['act']['data'] for i in response['actions']]
                    earliest = response['actions'][0]['timestamp']
                    latest = response['actions'][-1]['timestamp']
                    update = f'Fetched mining data between {earliest} and {latest}, still need until {before}'
                    print(update)
                    remaining_seconds = (dt(before) - dt(latest)).total_seconds()
                    after = f'{latest[:-1]}{int(latest[-1]) + 1}'
            except aiohttp.ServerDisconnectedError:
                await asyncio.sleep(1)
                continue
            except aiohttp.ClientPayloadError:
                await asyncio.sleep(1)
                continue
            except aiohttp.ClientOSError:
                await asyncio.sleep(1)
                continue
            for action in actions:
                if action['landowner'] != landowner:  # only monkey lands
                    continue
                amount = float(action['bounty'].replace(' TLM', ''))
                if action['miner'] not in miners:  # Set base TLM for new miners
                    miners[action['miner']] = 0
                miners[action['miner']] += amount
            if earliest == latest:
                break

            if counter % 10 == 0:
                got_miner_data = sorted(miners.items(), key=lambda x: x[1], reverse=True)
                i = 0
                update += f'\nMining data so far:\n'
                for item in got_miner_data:
                    miner, amount = item
                    i += 1
                    if i <= top:
                        update += f'{i}) {miner} - {amount:.4f} TLM\n'
                print(update)
            await asyncio.sleep(0.0001)

    to_send = f'Top {top_n} miners between {original_after} and {before}:\n'
    i = 0
    got_miner_data = sorted(miners.items(), key=lambda x: x[1], reverse=True)
    for item in got_miner_data:
        miner, amount = item
        i += 1
        if i <= top:
            to_send += f'{i}) {miner} - {amount:.4f} TLM\n'
    with open(filename, 'w+') as f:
        json.dump(got_miner_data, f, indent=4)
    total = sum(miners.values())
    print(f"Total TLM mined in that period is {total:.4f} by {len(got_miner_data)} "
          f"different miners.\n")
    if save_data:
        print(f"Find the full data in '{filename}' in the local folder.")
    print(to_send)


loop = asyncio.get_event_loop()
results = loop.run_until_complete(topminers(top_n, start_time, end_time, save_to_file))