New Tool: Auto-Claiming SCOT Rewards (Thanks @herman-german!)

Following up on the multi-account reward claimer script I shared recently, user @herman-german reached out with a great suggestion. They liked the idea but mentioned it would be even cooler if it could automatically claim the various SCOT bot tokens that many Hive tribes use for their rewards.

Challenge accepted!

The Hurdle: Undocumented APIs

The first step was figuring out how to claim SCOT rewards programmatically. Unlike the main Hive rewards, there wasn't clear documentation readily available for the SCOT claim API endpoints. This meant diving into the browser's developer tools on the Hive-Engine site and trying to decipher the logic from the minified JavaScript that handles the claims there. It's not the prettiest sight, but sometimes you gotta do what you gotta do!

The Solution: A SCOT Claimer Script

After puzzling through the necessary API calls and JSON formats, I managed to put together a working Python script that specifically handles claiming pending SCOT rewards.

Similar to the HIVE/HBD claimer, this script:

1. Reads a list of accounts from a YAML configuration file (accounts.yaml by default).
2. Uses the posting key of the first account in the list.
3. Requires you to grant posting authority from the other accounts to that first main account (you can use my Hive Account Authority tool for this).
4. Connects to the SCOT API to fetch pending rewards for each account.
5. If rewards are found, it constructs and broadcasts the necessary custom_json operation (scot_claim_token) using the main account's posting authority to claim the tokens for the target account.
6. Includes options for --dry-run simulation and --debug logging.
7. Uses hive-nectar for the Hive connection and broadcasting.

The Code

It's another standalone script for now. I'm currently merging these claimer tools into a single repository with shared code so today's screenshot may not match the output of this code, but for now, here's the functional version for SCOT claims GitHub Gist:

#!/usr/bin/env -S uv run --quiet --script
# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "hive-nectar",
#     "pyyaml",
#     "requests",
# ]
#
# [tool.uv.sources]
# hive-nectar = { git = "https://github.com/thecrazygm/hive-nectar/" }
# ///

"""
Hive-Engine SCOT Token Reward Claimer
-------------------------------------
This script connects to the Hive blockchain and Hive-Engine sidechain to claim SCOT token rewards
for multiple accounts using the authority (posting key) of a single main account.

Features:
- Connects to Hive nodes using a Posting Key from environment, YAML config, or CLI.
- Fetches pending SCOT token rewards from the SCOT API for each account.
- Filters out tokens with zero pending rewards.
- Loops through accounts and claims rewards for each token with pending rewards.
- Uses posting authority of the main account.
- Provides informative logging and robust error handling.
- Supports dry-run mode to simulate claims without broadcasting.

Author: thecrazygm
"""

import argparse
import json
import logging
import os
import sys
from typing import Any, Dict, List

import requests
from nectar import Hive
from nectar.account import Account
from nectar.nodelist import NodeList

# ---------------------
# Logging configuration
# ---------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger(__name__)


def load_accounts_and_posting_key(
    accounts_path: str = None,
) -> tuple[List[str], str | None]:
    """
    Load accounts list and Posting Key from YAML file.
    Args:
        accounts_path (str, optional): Path to YAML config file. If None, tries accounts.yaml in current directory.
    Returns:
        tuple: (list of account names, Posting Key or None)
    """
    import yaml

    if accounts_path:
        try:
            with open(accounts_path, "r") as f:
                data = yaml.safe_load(f)
            logger.info(f"Loaded accounts and Posting Key from {accounts_path}")
            return data.get("accounts", []), data.get("posting_key")
        except Exception as e:
            logger.error(f"Failed to load accounts from {accounts_path}: {e}")
            sys.exit(1)
    # Try default accounts.yaml
    if os.path.exists("accounts.yaml"):
        try:
            with open("accounts.yaml", "r") as f:
                data = yaml.safe_load(f)
            logger.info("Loaded accounts and Posting Key from accounts.yaml")
            return data.get("accounts", []), data.get("posting_key")
        except Exception as e:
            logger.error(f"Failed to load accounts from accounts.yaml: {e}")
            sys.exit(1)
    logger.error(
        "No account list found. Please provide --accounts or create accounts.yaml in the current directory."
    )
    sys.exit(1)


def get_posting_key(cli_posting_key: str = None, yaml_posting_key: str = None) -> str:
    """
    Retrieve and validate the Posting Key (private key) from CLI, YAML, or environment variables.
    Args:
        cli_posting_key (str, optional): Posting Key from command-line argument.
        yaml_posting_key (str, optional): Posting Key from YAML config.
    Returns:
        str: The Posting Key.
    """
    logger.debug(
        f"Attempting to retrieve Posting Key (cli_posting_key provided: {bool(cli_posting_key)}, yaml_posting_key provided: {bool(yaml_posting_key)})"
    )
    if cli_posting_key:
        logger.info("Using Posting Key from --posting_key argument.")
        posting_key = cli_posting_key
    elif yaml_posting_key:
        logger.info("Using Posting Key from YAML config file.")
        posting_key = yaml_posting_key
    else:
        posting_key = os.getenv("POSTING_KEY")
        if posting_key:
            logger.info("Using Posting Key from POSTING_KEY environment variable.")
    if not posting_key:
        logger.error(
            "Posting Key must be provided via --posting_key, YAML config, or POSTING_KEY env variable."
        )
        sys.exit(1)
    logger.debug("Posting Key successfully retrieved.")
    return posting_key


def connect_to_hive(posting_key: str) -> Hive:
    """
    Establish a connection to the Hive blockchain using the provided Posting Key.
    Automatically selects the best available Hive nodes.

    Args:
        posting_key (str): The Posting private key.

    Returns:
        Hive: A connected Hive blockchain instance.
    """
    try:
        logger.debug("Initializing NodeList and updating nodes...")
        nodelist = NodeList()
        nodelist.update_nodes()
        nodes = nodelist.get_hive_nodes()
        logger.debug(f"Selected Hive nodes: {nodes}")
        hive = Hive(keys=[posting_key], node=nodes)
        logger.debug("Hive instance created and connected.")
        return hive
    except Exception as e:
        logger.error(f"Failed to connect to Hive: {e}")
        sys.exit(1)


def format_token_amount(amount: float, precision: int) -> str:
    """
    Format a token amount with the correct precision.

    Args:
        amount (float): The token amount.
        precision (int): The token precision.

    Returns:
        str: Formatted token amount as a string.
    """
    format_str = f"{{:.{precision}f}}"
    return format_str.format(amount)


def get_scot_rewards(account_name: str) -> List[Dict[str, Any]]:
    """
    Fetch SCOT token rewards for a given account from the SCOT API.

    Args:
        account_name (str): The Hive account name to fetch rewards for.

    Returns:
        List[Dict[str, Any]]: List of tokens with pending rewards, properly formatted.
    """
    try:
        logger.debug(f"Fetching SCOT rewards for {account_name}")
        url = f"https://scot-api.hive-engine.com/@{account_name}"
        response = requests.get(url, params={"hive": 1}, timeout=10)
        response.raise_for_status()
        data = response.json()

        # Filter tokens with pending rewards > 0
        pending_rewards = []
        for symbol, token_data in data.items():
            raw_pending = token_data.get("pending_token", 0)
            if raw_pending > 0:
                precision = token_data.get("precision", 0)
                pending_amount = raw_pending / (10**precision)
                staked_amount = token_data.get("staked_tokens", 0) / (10**precision)
                pending_rewards.append(
                    {
                        "symbol": symbol,
                        "pending": format_token_amount(pending_amount, precision),
                        "staked": format_token_amount(staked_amount, precision),
                        "precision": precision,
                        "raw_pending": raw_pending,
                    }
                )
        return pending_rewards
    except Exception as e:
        logger.error(f"Error fetching SCOT rewards for {account_name}: {e}")
        return []


def claim_scot_rewards_for_all_accounts(
    hive: Hive, accounts: List[str], main_account_name: str, dry_run: bool = False
) -> None:
    """
    Claim SCOT token rewards for all accounts in the list using the authority of the main account.
    This function loops through accounts and claims any pending SCOT rewards.

    Args:
        hive (Hive): The connected Hive blockchain instance.
        accounts (List[str]): List of account names to claim rewards for.
        main_account_name (str): The account whose Posting Key is used for authority.
        dry_run (bool): If True, simulate claims without broadcasting.
    """
    logger.debug(
        f"Instantiating main account object for authority: {main_account_name}"
    )
    main_account = Account(main_account_name, blockchain_instance=hive)
    logger.debug(f"Account list to process: {accounts}")

    for account_name in accounts:
        try:
            logger.debug(f"Processing account: {account_name}")
            rewards = get_scot_rewards(account_name)
            if not rewards:
                logger.info(f"[{account_name}] No SCOT token rewards to claim.")
                continue

            for token in rewards:
                logger.info(
                    f"[{account_name}] {token['symbol']} rewards to claim: {token['pending']}"
                )

            if dry_run:
                for token in rewards:
                    logger.info(
                        f"[DRY RUN] Would claim {token['pending']} {token['symbol']} for {account_name} using authority of {main_account_name}."
                    )
                continue

            # Prepare tokens data for custom_json operation
            tokens_to_claim = [{"symbol": token["symbol"]} for token in rewards]

            custom_json_data = {
                "id": "scot_claim_token",
                "required_posting_auths": [main_account_name],
                "required_auths": [],
                "json": json.dumps(tokens_to_claim),
            }
            logger.debug(
                f"Broadcasting custom_json operation for {account_name}: {custom_json_data}"
            )

            tx = hive.custom_json(**custom_json_data)

            logger.info(
                f"[{account_name}] SCOT rewards claimed successfully using authority of {main_account_name}."
            )
            logger.debug(f"Transaction details: {tx}")

        except Exception as e:
            import traceback

            logger.error(
                f"Error claiming SCOT rewards for {account_name} using {main_account_name}: {type(e).__name__}: {e}"
            )
            logger.error(traceback.format_exc())


def main() -> None:
    """
    Main entry point for the Hive-Engine SCOT Token Reward Claimer script.
    Parses command-line arguments, loads Posting Key, connects to Hive,
    and claims SCOT rewards for all accounts in the config.
    """
    parser = argparse.ArgumentParser(
        description="Claim Hive-Engine SCOT token rewards for multiple accounts."
    )
    parser.add_argument(
        "-w",
        "--posting_key",
        type=str,
        default=None,
        help="Posting Key (private key) for authority account. If omitted, uses POSTING_KEY env variable or YAML config.",
    )
    parser.add_argument(
        "-a",
        "--accounts",
        type=str,
        default=None,
        help="Path to YAML file with accounts and/or Posting Key. If omitted, uses accounts.yaml if available.",
    )
    parser.add_argument(
        "-n",
        "--dry-run",
        action="store_true",
        help="Simulate reward claims without broadcasting transactions.",
    )
    parser.add_argument(
        "-d", "--debug", action="store_true", help="Enable debug logging."
    )
    args = parser.parse_args()

    logger.debug(f"Parsed CLI arguments: {args}")

    if args.debug:
        logger.setLevel(logging.DEBUG)
        logger.debug("Debug logging enabled.")

    accounts, yaml_posting_key = load_accounts_and_posting_key(args.accounts)
    posting_key = get_posting_key(args.posting_key, yaml_posting_key)
    logger.debug("Connecting to Hive blockchain...")
    hive = connect_to_hive(posting_key)
    main_account_name = accounts[0]
    logger.debug(f"Using main authority account: {main_account_name}")
    claim_scot_rewards_for_all_accounts(
        hive, accounts, main_account_name, dry_run=args.dry_run
    )


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        sys.exit(1)

Anyway, that's the story behind the SCOT claimer script. It was a fun little challenge digging into undocumented territory, and it solves a specific need that came up thanks to community feedback. Hopefully, it's useful for others who manage multiple tribe accounts and want to automate their reward claims!

As always,
Michael Garcia a.k.a. TheCrazyGM