如何使用 Python 获取所有 Hive 帐户委托人

Answer 1

由于 Hive 是一个区块链，从技术上讲，任何人都应该可以访问所有信息，并且可以使用官方 Hive API 轻松访问。我们可以使用condenser_api.* 来获取所有使用get_account_history method 的帐户交易。

JSON RPC 结果可以通过方法的参数进一步缩小：

• 帐户：字符串
• 开始：int（-1 表示反向历史记录或任何正数）
• 限制：整数（最多 1000）
• operation_filter_low：int（可选）
• operation_filter_high：int（可选）

这里棘手的部分是Operation IDs 的 128 位掩码，它可以来不是这样对一些新开发人员来说很简单。

只需检查上面operations.hpp 链接中的操作 ID，并将操作 ID 用作位字符串的零长度。对于下面的示例，它是 40+1，因为我们正在执行 ljust.(total_string_length, character) 然后将位字符串转换回整数。

operation_id = 40  # delegate_vesting_shares_operation
operation_filter_low = int("1".ljust(operation_id+1, "0"), 2)

将结果值用于operation_filter_low 参数并在发送请求之前继续格式化有效负载。最好创建一个实用程序函数来预格式化参数、有效负载和 API 调用以增加可重用性。

# formatting the parameters
username = "hivehealth"
start = -1
limit = 1000
params = [username, start, limit, operation_filter_low]

# preparing the payload
request_id = 1
method = "condenser_api.get_account_history"
payload = { "jsonrpc": "2.0", "method": method, "params": params, "id": request_id}

创建 API 调用函数。检查所有 Hive Witness Nodes 并创建一个节点列表以循环，然后在完成请求操作后中断。

NODES = ["api.hive.blog"]
def _send_request(session, headers, payload, timeout=60):
    response = None
    for node in NODES:
        try:
            url = "https://" + node
            if None not in auth:
                response = session.post(url,
                                        headers=headers,
                                        json=payload,
                                        timeout=timeout)
                break
        except:
            print(f"\nNode API '{node}' is offline, trying the next node....")
    if response is None:
        return []
    
    response.raise_for_status()
    try:
        response = json.loads(response.content.decode("utf-8"))
        if "result" in response:
            return response["result"]
    except:
        pass
    print(f"\nNode API response from '{node}' is unusable....")
    return []

该函数将返回一个列表列表，并且可以使用循环访问。

[[4, {'trx_id': '1c453a97c10daff1c27161ec6f9d4f4566b54896', 'block': 55496700, 'trx_in_block': 42, 'op_in_trx': 0, 'virtual_op': 0, 'timestamp': '2021-07-09T16:37:39', 'op': ['delegate_vesting_shares', {'delegator': 'giftgiver', 'delegatee': 'checkit', 'vesting_shares': '30000.000000 VESTS'}]}]]

要运行它，下面是示例代码：

import json
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def delegations():
    """ Get all Hive inward delegations. """
    
    # Prepare session
    session = requests.Session()
    retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[502, 503, 504])
    session.mount("https://", HTTPAdapter(max_retries=retries))
    headers = { "User-Agent": f"CustomAPI v0.0.9", "content-type": "application/json; charset=utf-8" }

    # formatting the parameters
    username = "hivehealth"
    start = -1
    limit = 1000
    params = [username, start, limit, operation_filter_low]

    # preparing the payload
    request_id = 1
    method = "condenser_api.get_account_history"
    payload = { "jsonrpc": "2.0", "method": method, "params": params, "id": request_id}

    transactions = send_request(session, headers, payload, timeout)

    ## customize this part as needed
    delegators = {}
    for transaction in transactions:
        delegation = transaction[1]["op"]
        delegator = delegation[1]["delegator"]
        if delegator not in delegators:
            delegators[delegator] = {}
        timestamp = transaction[1]["timestamp"]
        # can optionally be { delegator : vesting }
        # but changes in delegation can be important at times
        delegators[delegator][timestamp] = float(delegation[1]["vesting_shares"].split(" ")[0])
    return delegators

就是这样，您现在可以使用上面的代码对帐户历史提出更窄的请求。只要确保优化代码并根据需要使其他变量动态化。