Source code for bitcoinlib.services.chainso

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    Chain.so client
#    © 2017-2022 October - 1200 Web Development <http://1200wd.com/>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import logging
from datetime import datetime
from bitcoinlib.main import MAX_TRANSACTIONS
from bitcoinlib.services.baseclient import BaseClient, ClientError
from bitcoinlib.transactions import Transaction


_logger = logging.getLogger(__name__)

PROVIDERNAME = 'chainso'


[docs] class ChainSo(BaseClient): def __init__(self, network, base_url, denominator, *args): super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)
[docs] def compose_request(self, function, data='', parameter='', variables=None, method='get'): url_path = function url_path += '/' + self.provider_coin_id if data: url_path += '/' + data if parameter: url_path += '/' + parameter if variables is None: variables = {} if self.api_key: variables.update({'api_key': self.api_key}) return self.request(url_path, variables, method)
[docs] def sendrawtransaction(self, rawtx): res = self.compose_request('send_tx', variables={'tx_hex': rawtx}, method='post') return { 'txid': '' if 'data' not in res else res['data']['txid'], 'response_dict': res }
[docs] def getbalance(self, addresslist): balance = 0.0 for address in addresslist: res = self.compose_request('get_address_balance', address) balance += float(res['data']['confirmed_balance']) + float(res['data']['unconfirmed_balance']) return int(balance * self.units)
[docs] def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS): txs = [] lasttx = after_txid res = self.compose_request('get_tx_unspent', address, lasttx) if res['status'] != 'success': pass for tx in res['data']['txs'][:limit]: txs.append({ 'address': address, 'txid': tx['txid'], 'confirmations': tx['confirmations'], 'output_n': -1 if 'output_no' not in tx else tx['output_no'], 'input_n': -1 if 'input_no' not in tx else tx['input_no'], 'block_height': None, 'fee': None, 'size': 0, 'value': int(round(float(tx['value']) * self.units, 0)), 'script': tx['script_hex'], 'date': datetime.utcfromtimestamp(tx['time']), }) if len(txs) >= 1000: _logger.warning("ChainSo: transaction list has been truncated, and thus is incomplete") return txs
[docs] def getrawtransaction(self, txid): res = self.compose_request('get_tx', txid) return res['data']['tx_hex']
[docs] def gettransaction(self, txid, block_height=None): res = self.compose_request('get_tx', txid) tx = res['data'] rawtx = tx['tx_hex'] t = Transaction.parse_hex(rawtx, strict=self.strict, network=self.network) input_total = 0 output_total = 0 if not t.coinbase: for n, i in enumerate(t.inputs): i.value = int(round(float(tx['inputs'][n]['value']) * self.units, 0)) input_total += i.value for o in t.outputs: o.spent = None output_total += o.value if not t.block_height and tx['confirmations']: t.block_height = self.getblock(tx['blockhash'], False, 1, 1)['height'] t.block_hash = tx['blockhash'] t.rawtx = bytes.fromhex(rawtx) t.size = tx['size'] t.network = self.network t.locktime = tx['locktime'] t.input_total = input_total t.output_total = output_total t.fee = 0 if t.input_total: t.fee = t.input_total - t.output_total t.confirmations = tx['confirmations'] if tx['confirmations']: t.status = 'confirmed' t.date = datetime.utcfromtimestamp(tx['time']) else: t.status = 'unconfirmed' t.date = None return t
[docs] def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS): txs = [] res1 = self.compose_request('get_tx_received', address, after_txid) if res1['status'] != 'success': raise ClientError("Chainso get_tx_received request unsuccessful, status: %s" % res1['status']) res2 = self.compose_request('get_tx_spent', address, after_txid) if res2['status'] != 'success': raise ClientError("Chainso get_tx_spent request unsuccessful, status: %s" % res2['status']) res = res1['data']['txs'] + res2['data']['txs'] res = sorted(res, key=lambda x: x['time']) tx_conf = [] for t in res: tt = (t['confirmations'], t['txid']) if tt not in tx_conf: tx_conf.append(tt) for tx in tx_conf[:limit]: t = self.gettransaction(tx[1]) txs.append(t) return txs
[docs] def blockcount(self): return self.compose_request('get_info')['data']['blocks']
[docs] def mempool(self, txid): res = self.compose_request('is_tx_confirmed', txid) if res['status'] == 'success' and res['data']['confirmations'] == 0: return [txid] return []
[docs] def getblock(self, blockid, parse_transactions, page, limit): if limit > 5: limit = 5 bd = self.compose_request('get_block', str(blockid))['data'] if parse_transactions: txs = [] for txid in bd['txs'][(page-1)*limit:page*limit]: # try: txs.append(self.gettransaction(txid, block_height=bd['block_no'])) # except Exception as e: # raise ClientError("Could not parse tx %s with error %s" % (txid, e)) else: txs = bd['txs'] n_txs = len(bd['txs']) block = { 'bits': None, 'depth': bd['confirmations'], 'block_hash': bd['blockhash'], 'height': bd['block_no'], 'merkle_root': bd['merkleroot'], 'nonce': None, 'prev_block': bd['previous_blockhash'], 'time': bd['time'], 'tx_count': n_txs, 'txs': txs, 'version': b'', 'page': page, 'pages': None if not limit else int(n_txs // limit) + (n_txs % limit > 0), 'limit': limit } return block
# def getrawblock(self, blockid): # def isspent(self, txid, output_n):
[docs] def getinfo(self): info = self.compose_request('get_info')['data'] return { 'blockcount': info['blocks'], 'chain': info['name'], 'difficulty': int(float(info['mining_difficulty'])), 'hashrate': int(float(info['hashrate'])), 'mempool_size': int(info['unconfirmed_txs']), }