"""
**Copyright**::
+===================================================+
| © 2019 Privex Inc. |
| https://www.privex.io |
+===================================================+
| |
| CryptoToken Converter |
| |
| Core Developer(s): |
| |
| (+) Chris (@someguy123) [Privex] |
| |
+===================================================+
"""
import pytz
import logging
from decimal import Decimal
from typing import Generator, Iterable, List
from datetime import datetime
from django.core.cache import cache
from django.utils import timezone
from payments.coin_handlers.base.BaseLoader import BaseLoader
from payments.coin_handlers.Bitshares.BitsharesMixin import BitsharesMixin
from payments.models import Coin
from steemengine.helpers import empty
from bitshares.account import Account
from bitshares.amount import Amount
from bitshares.asset import Asset
from bitsharesbase.memo import decode_memo
from bitsharesbase.account import PrivateKey, PublicKey
log = logging.getLogger(__name__)
[docs]class BitsharesLoader(BaseLoader, BitsharesMixin):
"""
This class handles loading transactions for the **Bitshares** network, and can support any token
on Bitshares.
**Copyright**::
+===================================================+
| © 2019 Privex Inc. |
| https://www.privex.io |
+===================================================+
| |
| CryptoToken Converter |
| |
| Core Developer(s): |
| |
| (+) Chris (@someguy123) [Privex] |
| |
+===================================================+
"""
provides = [] # type: List[str]
"""
This attribute is automatically generated by scanning for :class:`models.Coin` s with the type ``bitshares``.
This saves us from hard coding specific coin symbols. See __init__.py for populating code.
"""
def __init__(self, symbols):
super().__init__(symbols=symbols)
self.tx_count = 1000
self.loaded = False
[docs] def clean_txs(self, account: Account, symbol: str, transactions: Iterable[dict]) -> Generator[dict, None, None]:
"""
Filters a list of transactions by the receiving account, yields dict's conforming with
:class:`payments.models.Deposit`
:param str account: The 'to' account to filter by
:param str symbol: The symbol of the token being filtered
:param list<dict> transactions: A list<dict> of transactions to filter
:return: A generator yielding ``dict`` s conforming to :class:`payments.models.Deposit`
"""
for tx in transactions:
try:
data = tx['op'][1] # unwrap the transaction structure to get at the data within
if data['to'] != account['id'] or data['from'] == account['id']:
continue
# cache asset data for 5 mins, so we aren't spamming the RPC node for data
amount_info = data['amount']
asset_id = amount_info['asset_id']
asset_key = 'btsasset:%s' % (asset_id,)
asset = cache.get(asset_key, default=None)
if asset is None:
asset_obj = self.get_asset_obj(asset_id)
if asset_obj is not None:
asset = { 'symbol' : asset_obj.symbol, 'precision' : asset_obj.precision }
cache.set(asset_key, asset, 300)
else:
continue
if asset['symbol'] != symbol:
continue
raw_amount = Decimal(int(amount_info['amount']))
transfer_quantity = raw_amount / (10 ** asset['precision'])
# cache account data for 5 mins, so we aren't spamming the RPC node for data
account_key = 'btsacc:%s' % (data['from'],)
from_account_name = cache.get(account_key, default=data['from'])
if from_account_name == data['from']:
from_account = self.get_account_obj(data['from'])
if from_account is not None:
from_account_name = from_account.name
cache.set(account_key, from_account_name, 300)
else:
log.exception('From account not found for transaction %s', tx)
# decrypt the transaction memo
memo_msg = ''
if 'memo' in data:
memo = data['memo']
try:
memokey = self.get_private_key(account.name, 'memo')
privkey = PrivateKey(memokey)
pubkey = PublicKey(memo['from'], prefix='BTS')
memo_msg = decode_memo(privkey, pubkey, memo['nonce'], memo['message'])
except Exception as e:
memo_msg = '--cannot decode memo--'
log.exception('Error decoding memo %s, got exception %s', memo['message'], e)
# fetch timestamp from the block containing this transaction
# (a hugely inefficient lookup but unfortunately no other way to do this)
tx_datetime = datetime.fromtimestamp(self.get_block_timestamp(tx['block_num']))
tx_datetime = timezone.make_aware(tx_datetime, pytz.UTC)
clean_tx = dict(
txid=tx['id'], coin=self.coins[symbol].symbol, tx_timestamp=tx_datetime,
from_account=from_account_name, to_account=account.name, memo=memo_msg,
amount=transfer_quantity
)
yield clean_tx
except Exception as e:
log.exception('Error parsing transaction data. Skipping this TX. tx = %s, exception = %s', tx, e)
continue
[docs] def list_txs(self, batch=0) -> Generator[dict, None, None]:
"""
Get transactions for all coins in `self.coins` where the 'to' field matches coin.our_account
If :meth:`.load()` hasn't been ran already, it will automatically call self.load()
:return: Generator yielding dicts that conform to :class:`models.Deposit`
"""
if not self.loaded:
self.load()
for symbol, c in self.coins.items():
try:
acc_name = self.coins[symbol].our_account
acc = self.get_account_obj(acc_name)
if acc is None:
log.error('Account %s not found while loading transactions for coin %s. Skipping for now.', acc_name, c)
continue
# history returns a generator with automatic batching, so we don't have to worry about batches.
txs = acc.history(only_ops = ['transfer'], limit=self.tx_count)
yield from self.clean_txs(symbol=symbol, transactions=txs, account=acc)
except:
log.exception('Error while loading transactions for coin %s. Skipping for now.', c)
continue
[docs] def load(self, tx_count=1000):
log.info('Loading Bitshares transactions...')
self.tx_count = tx_count
for symbol, coin in self.coins.items():
if empty(coin.our_account):
log.warning('The coin %s does not have `our_account` set. Refusing to load transactions.', coin)
del self.coins[symbol]
self.symbols = [s for s in self.symbols if s != symbol]
self.loaded = True