Python adapter class for the Insight block explorer software - written by Someguy123 at Privex - released under X11 / MIT license

insight.py

"""
This is a Python adapter class for the Insight block explorer software, which can query the Insight API for
various public block explorers, such as BitPay's multi-coin block explorer ( https://bitpay.com/insight/ ) ,
and Loshan's LiteCore LTC Insight explorer ( https://insight.litecore.io ).

This adapter was written by Chris (Someguy123) ( https://github.com/Someguy123 ), originally
for use in Privex's Payment Gateway ( https://pay.privex.io ), as well as
AnonSteem ( https://anon.steem.network ), and BeeAnon ( https://beeanon.com ).

License: X11 / MIT

Requirements:

    * Python 3.6+ (preferably 3.7+)
    * PyPi package ``privex-helpers`` ( ``python3.9 -m pip install -U privex-helpers`` )
    * If you're on Python <=3.6 , then you need PyPi package ``dataclasses`` ( ``python3.9 -m pip install -U dataclasses`` ),
      which is a package that tries to backport Python 3.7's native :mod:`.dataclasses` module to Python version's
      older than 3.7
    * An internet connection to make the queries.


If you're looking for somewhere to host the app you're writing using this adapter, consider
trying out Privex ( https://www.privex.io ) - a privacy focused web hosting provider, which
accepts a variety of crypto (LTC, DOGE, BTC, BCH, EOS, Hive, HBD, and possibly others at the
time you read this).

Privex offers both Virtual (VPS) and Dedicated Server's for low prices, with maximum privacy - requiring
only an email and a name (can be a pseudonym/username) to order from them.

"""
import random
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from decimal import Decimal
from collections import namedtuple
from typing import Callable, Dict, List, Optional
from privex.helpers import retry_on_err, DictDataClass
import requests
import logging

log = logging.getLogger(__name__)
__all__ = ['InsightConfig', "InsightConf", 'APINode', 'insight_apis', 'InsightAdapter']

InsightConf = InsightConfig = namedtuple('InsightConf', 'endpoint version')
sysrand = random.SystemRandom()
# insight_apis = {
#     "ltc": [
#         InsightConf("https://api.bitcore.io/api/LTC/mainnet", 2),
#         InsightConf("https://insight.litecore.io/api", 1)
#     ],
#     "doge": [InsightConf("https://api.bitcore.io/api/DOGE/mainnet", 2)],
#     "btc": [InsightConf("https://api.bitcore.io/api/BTC/mainnet", 2)],
#     "bch": [InsightConf("https://api.bitcore.io/api/BCH/mainnet", 2)],
# }


@dataclass
class APINode(DictDataClass):
    endpoint: str
    coin: str = None
    version: int = 2
    max_tries: int = 2
    retry_after: int = 60 * 5
    working: bool = True
    last_tried: datetime = field(default_factory=datetime.utcnow)
    tries: int = 0

    @property
    def should_retry(self) -> bool:
        """
        Returns a bool representing whether or not we've surpassed :attr:`.retry_after` seconds since :attr:`.last_tried` .
        
        If you need to know generally whether or not an API node is usable (is it working, or is it due for a retry?), then
        you should use the prop :meth:`.usable` instead.
        """
        return (self.last_tried + timedelta(seconds=self.retry_after)) <= datetime.utcnow()
    
    @property
    def usable(self) -> bool:
        """
        Returns ``True`` if the API is either working, or if we should retry it (it's been more than :attr:`.retry_after` seconds
        since :attr:`.last_tried` ). Otherwise returns ``False`` .
        """
        return self.working or self.should_retry
    
    def add_try(self, autofail=True, set_last=True) -> bool:
        """
        Adds a try to :attr:`.tries` , updates :attr:`.last_tried` , sets :attr:`.working` to False after hitting :attr:`.max_tries` ,
        and returns a :class:`.bool` to tell you whether or not you should continue retrying.
        
        If/when :attr:`.max_tries` is reached, then this method will return ``False`` (give up) instead of ``True`` (try again).
        
        :param bool autofail: Whether or not to set :attr:`.working` to ``False`` after reaching :attr:`.max_tries`
        :param bool set_last: Whether or not to update :attr:`.last_tried`
        :return bool try_again: Normally returns ``True`` (try again), But if/when :attr:`.max_tries` is reached, then
                                this method will return ``False`` (give up) instead
        """
        self.tries += 1
        if set_last: self.last_tried = datetime.utcnow()
        if self.tries >= self.max_tries:
            if autofail:
                self.working = False
            return False
        return True
    
    def set_working(self) -> "APINode":
        """Mark this node as working. Sets :attr:`.working` to True, resets :attr:`.tries` to 0, and updates :attr:`.last_tried` ."""
        self.working = True
        self.tries = 0
        self.last_tried = datetime.utcnow()
        return self
    
    def __str__(self) -> str: return self.endpoint
    
    def __int__(self) -> int: return self.version


insight_apis: Dict[str, List[APINode]] = {
    "ltc":  [
        APINode("https://api.bitcore.io/api/LTC/mainnet", 'ltc', 2),
        APINode("https://insight.litecore.io/api", 'ltc', 1)
    ],
    "doge": [APINode("https://api.bitcore.io/api/DOGE/mainnet", 'doge', 2)],
    "btc":  [APINode("https://api.bitcore.io/api/BTC/mainnet", 'btc', 2)],
    "bch":  [APINode("https://api.bitcore.io/api/BCH/mainnet", 'bch', 2)],
}


class InsightAdapter:
    """
    This is a Python adapter class for the Insight block explorer software, which can query the Insight API for
    various public block explorers, such as BitPay's multi-coin block explorer ( https://bitpay.com/insight/ ) ,
    and Loshan's LiteCore LTC Insight explorer ( https://insight.litecore.io ).
    
    Basic Usage::
    
        >>> b = InsightAdapter('ltc')
        >>> bal = b.get_balance('LVXXmgcVYBZAuiJM3V99uG48o3yG89h2Ph')
        >>> repr(bal)
        Decimal('0.34642876')
        >>> print(bal)
        0.34642876
        
    This adapter was written by Chris (Someguy123) ( https://github.com/Someguy123 ), originally
    for use in Privex's Payment Gateway ( https://pay.privex.io ), as well as
    AnonSteem ( https://anon.steem.network ), and BeeAnon ( https://beeanon.com ).
    
    License: X11 / MIT
    
    If you're looking for somewhere to host the app you're writing using this adapter, consider
    trying out Privex ( https://www.privex.io ) - a privacy focused web hosting provider, which
    accepts a variety of crypto (LTC, DOGE, BTC, BCH, EOS, Hive, HBD, and possibly others at the
    time you read this).
    
    Privex offers both Virtual (VPS) and Dedicated Server's for low prices, with maximum privacy - requiring
    only an email and a name (can be a pseudonym/username) to order from them.
    """
    version_map: Dict[int, Callable[[str], Decimal]]
    confirmations: int
    rsess: requests.Session
    coin: str
    insight_apis: Dict[str, List[APINode]] = insight_apis
    current_node: APINode
    
    def __init__(self, coin: str, confirmations: int = 0):
        """
        InsightAdapter offers various APIs to Insight.litecore

        Example usage:

        >>> b = InsightAdapter('LTC')
        >>> print(b.get_balance('LVXXmgcVYBZAuiJM3V99uG48o3yG89h2Ph'))
            20.1234

        :param str coin: The short code for a given coin, e.g. ``btc`` , ``ltc`` , ``doge`` , ``bch``
        :param int confirmations: integer minimum confirmations needed for balance to be trusted.
        """
        self.confirmations = confirmations
        self.coin = coin.lower()
        self.rsess = requests.session()
        self.version_map = {
            1: self.get_balance_v1,
            2: self.get_balance_v2
        }
        """
        A dictionary mapping API versions to their appropriate handling methods.
        """
        self.current_node = sysrand.choice(self.working_nodes)

    @property
    def all_nodes(self) -> List[APINode]: return insight_apis[self.coin]
    
    @property
    def working_nodes(self) -> List[APINode]: return [n for n in self.all_nodes if n.usable]

    @property
    def node_index(self) -> Optional[int]:
        return self.all_nodes.index(self.current_node) if self.current_node else None
    
    @node_index.setter
    def node_index(self, value: int):
        self.current_node = self.all_nodes[value]
    
    @property
    def api_url(self) -> str: return self.current_node.endpoint.rstrip('/')

    @property
    def api_version(self) -> int: return int(self.current_node.version)

    def new_node(self, save=True, usable=True) -> APINode:
        """
        Pick a random node, optionally save it to :attr:`.current_node` , and return it.
        
        :param bool save: (def: True) Whether or not to save the picked node to :attr:`.current_node`
        :param bool usable: (def: True) Whether to pick from only usable/working nodes (True), or pick from ALL nodes (False).
        :return APINode node: The node that was picked.
        """
        node = sysrand.choice(self.working_nodes if usable else self.all_nodes)
        if save: self.current_node = node
        return node
    
    def next_node(self, usable=True) -> APINode:
        """
        Select the next node incrementally, save it to :attr:`.current_node` , and return it.
        
        Unlike :meth:`.new_node` which selects a random node, this method simply **increments the node index** (or returns to 0
        if at the end of the list)
        
        After the node is picked, similar to new_node, it optionally checks if it's usable, and then saves it to :attr:`.current_node`

        :param bool usable: (def: True) Whether to select only usable/working nodes (True), or select ANY node (False).
        :return APINode node: The node that was picked.
        """
        log.info(f"next_node(usable={usable}) was called. Switching from current node (idx: {self.node_index}): {self.current_node!r}")
        if (self.node_index + 1) >= len(self.all_nodes):
            self.node_index = 0
        else:
            self.node_index += 1

        if usable and len(self.working_nodes) > 0:
            if not self.current_node.usable:
                log.info(f"Next Insight node (idx {self.node_index}) {self.current_node!r} is NOT usable."
                         f"Switching to the node after this.")
                return self.next_node(usable=usable)
        log.warning(f"Switched to next Insight node (idx: {self.node_index}): {self.current_node!r}")
        return self.current_node

    @retry_on_err()
    def get_balance_v1(self, address: str) -> Decimal:
        """
        Insight v1 API simply returns an very simple response - the integer number of satoshis - no JSON or XML etc., just
        the raw number as the response.
        
        :param str address:        The address to lookup
        :return Decimal balance:   The balance of the address
        """
        api_url = f"{self.api_url}/addr/{address}/balance"
        try:
            log.debug(f"Getting balance for coin '{self.coin}' address '{address}' from API node: {self.current_node}")
            log.debug(f"Full API URL: {api_url}")
            res = self.rsess.get(api_url)
            res.raise_for_status()
            data = res.text
            dres = Decimal(str(data)) / Decimal(10 ** 8)
            self.current_node.set_working()
            return dres
        except Exception as e:
            # Retry the current node up to max_tries. If add_try returns False, then the current node has exceeded max_tries,
            # thus we should switch to the next usable node.
            if not self.current_node.add_try():
                log.warning(f"Insight API node '{self.current_node!r}' has exceeded max_tries during lookup of address {address!r}. "
                            f"Switching to next node.")
                new_node = self.next_node()
                log.warning(f"Switched to next node: {new_node}")
            raise e
    
    @retry_on_err()
    def get_balance_v2(self, address: str) -> Decimal:
        """
        Insight v2 API returns JSON instead of a flat number, and has a slightly different endpoint, but still returns an integer
        number of satoshis inside of the JSON.

        :param str address:        The address to lookup
        :return Decimal balance:   The balance of the address
        """
        api_url = f"{self.api_url}/address/{address}/balance"
        try:
            log.debug(f"Getting balance for coin '{self.coin}' address '{address}' from API node: {self.current_node}")
            log.debug(f"Full API URL: {api_url}")
            res = self.rsess.get(api_url)
            res.raise_for_status()
            data = res.json()
            data = data['confirmed'] if self.confirmations > 0 else data['balance']
            dres = Decimal(str(data)) / Decimal(10 ** 8)
            self.current_node.set_working()
            return dres
        except Exception as e:
            # Retry the current node up to max_tries. If add_try returns False, then the current node has exceeded max_tries,
            # thus we should switch to the next usable node.
            if not self.current_node.add_try():
                log.warning(f"Insight API node '{self.current_node!r}' has exceeded max_tries during lookup of address {address!r}. "
                            f"Switching to next node.")
                new_node = self.next_node()
                log.warning(f"Switched to next node: {new_node}")
            raise e
    
    def get_balance(self, address: str) -> Decimal:
        """
        Gets the balance for an address from the Insight API

        :param address: String or list<str> of self.coin addresses
        :return Decimal balance: The balance of the address as a Decimal
        """
        # Strip 'bitcoincash:' from the start of any BCH addresses passed to this method.
        if self.coin == 'bch' and 'bitcoincash' in address:
            address = address.replace('bitcoincash:', '')
        # Grab the appropriate get_balance method depending on which api_version we need to use.
        get_bal = self.version_map[self.api_version]
        return get_bal(address.strip())