Simple Hive App on a $7 SBC 3: Block Stream and Discord Webhook

Welcome to part 3 of this series! Sorry for the long delay between posts. Check out part 1 and part 2 if you want to catch up.

In this post we'll create a HiveCondenser class to keep things clean and organized. Then we'll stream blocks to look for specific transactions and use a Discord webhook to send a message when needed. Let's dive right in!

HiveCondenser Class

Most of this code was in the main script in part 2. It makes sense to move it to a separate file though, which will really make things neat in the main script.
Here is the new HiveCondenser class:

import datetime
from time import sleep
from hivesign import sign_trx

class HiveCondenser():
    def __init__(self, session, node):
        self.session = session
        self.node = node
        self.id = 1
    
    @property
    def unow_ts(self):
        return datetime.datetime.now(datetime.timezone.utc).timestamp()
    
    def get_props(self, expiration=60):
        result = self.do_condenser('get_dynamic_global_properties')
        props = {}
        props['head_block_num'] = result['head_block_number']
        props['ref_block_num'] = props['head_block_num'] & 0xFFFF
        props['ref_block_prefix'] = int.from_bytes(bytes.fromhex(result['head_block_id'][8:16]), 'little')
        e = datetime.datetime.fromisoformat(result['time']) + datetime.timedelta(seconds=expiration)
        props['expiration'] = e.isoformat()
        return props
    
    def contruct_trx(self, operations, key):
        props = get_props()
        trx = {
            'expiration':props['expiration'],
            'ref_block_num':props['ref_block_num'],
            'ref_block_prefix':props['ref_block_prefix'],
            'operations':operations,
            'extensions':[],
            'signatures':[]
        }
        return sign_trx(trx, key)
    
    def do_condenser(self, method, params=[]):
        if not isinstance(params, list):
            params = [params]
        data = {'jsonrpc':'2.0', 'method':f'condenser_api.{method}', 'params':params, 'id':self.id}
        with self.session.post(self.node, json=data) as r:
            self.id += 1
            return r.json().get('result')
    
    def get_block(self, block):
        return self.do_condenser('get_block', block)
    
    def stream_blocks(self, start_block=None):
        if not start_block:
            props = self.get_props()
            start = props['head_block_num']
        block_num = start
        while True:
            block = self.get_block(block_num)
            if block is None:
                sleep(3)
            else:
                yield block
                block_num += 1
                block_ts = datetime.datetime.fromisoformat(block['timestamp'] + '+00:00').timestamp()
                dif = self.unow_ts - block_ts
                wait = 3.5 - dif
                if (wait > 0):
                    sleep(wait)
    
    def broadcast(self, operations, key):
        trx, trx_id = contruct_trx(operations, key)
        self.do_condenser('broadcast_transaction', trx)
        return [trx, trx_id]

The new item of note is the stream_blocks generator function that will let us easily check each fresh block for whatever we're looking for.

Main Script: Block Stream and Discord Message

I decided to just check for custom json operations with the dcity id and send them to Discord via a webhook. Not entirely practical, but it's a start.
Here's the new main script:

import rapidjson as json
from requests import Session
from discord import Webhook, RequestsWebhookAdapter
from hivecondenser import HiveCondenser

def main():
    with Session() as session:
        condenser = HiveCondenser(session, 'https://api.deathwing.me')
        webhook = Webhook.from_url(url=webhook_url, adapter=RequestsWebhookAdapter(session))
        for block in condenser.stream_blocks():
            for trx in block['transactions']:
                for type, op in trx['operations']:
                    if (type != 'custom_json'): continue
                    if (op['id'] != 'dcity'): continue
                    d = {}
                    d['block_num'] = trx['block_num']
                    d['transaction_num'] = trx['transaction_num']
                    d['transaction_id'] = trx['transaction_id']
                    d['account'] = op['required_auths'][0] if op['required_auths'] else op['required_posting_auths'][0]
                    d['id'] = op['id']
                    d['json'] = json.loads(op['json'])
                    webhook.send(f'```{json.dumps(d, indent=4)}```', username='Orange Pi')

if __name__ == '__main__':
    main()

And of course some picture evidence of it working:

I'll try to keep it running for a few days at least.Here's an invite to the channel if you're really bored: https://discord.gg/3d7z9grg53

Next Steps

Next time we'll do some new stuff (give me ideas pls 😅). At some point things might get asynchronous.
I'm welcome to any questions, comments, or suggestions. Thanks for reading!