cic-internal-integration/apps/cic-ussd/cic_ussd/account/statement.py
2021-10-07 15:12:35 +00:00

112 lines
4.0 KiB
Python

# standard imports
import datetime
import logging
from typing import Optional
# external imports
import celery
from chainlib.hash import strip_0x
from cic_eth.api import Api
# local import
from cic_ussd.account.chain import Chain
from cic_ussd.account.transaction import from_wei
from cic_ussd.cache import cache_data_key, get_cached_data
from cic_ussd.translation import translation_for
logg = logging.getLogger(__name__)
def filter_statement_transactions(transaction_list: list) -> list:
"""This function parses a transaction list and removes all transactions that entail interactions with the
zero address as the source transaction.
:param transaction_list: Array containing transaction objects.
:type transaction_list: list
:return: Transactions exclusive of the zero address transactions.
:rtype: list
"""
return [tx for tx in transaction_list if tx.get('source_token') != '0x0000000000000000000000000000000000000000']
def generate(querying_party: str, queue: Optional[str], transaction: dict):
"""
:param querying_party:
:type querying_party:
:param queue:
:type queue:
:param transaction:
:type transaction:
:return:
:rtype:
"""
s_generate_statement = celery.signature(
'cic_ussd.tasks.processor.generate_statement', [querying_party, transaction], queue=queue
)
s_generate_statement.apply_async()
def get_cached_statement(blockchain_address: str) -> bytes:
"""This function retrieves an account's cached record of a specific number of transactions in chronological order.
:param blockchain_address: Bytes representation of the hex value of an account's blockchain address.
:type blockchain_address: bytes
:return: Account's transactions statements.
:rtype: str
"""
identifier = bytes.fromhex(strip_0x(blockchain_address))
key = cache_data_key(identifier=identifier, salt=':cic.statement')
return get_cached_data(key=key)
def parse_statement_transactions(statement: list):
"""This function extracts information for transaction objects loaded from the redis cache and structures the data in
a format that is appropriate for the ussd interface.
:param statement: A list of transaction objects.
:type statement: list
:return:
:rtype:
"""
parsed_transactions = []
for transaction in statement:
action_tag = transaction.get('action_tag')
amount = from_wei(transaction.get('token_value'))
direction_tag = transaction.get('direction_tag')
token_symbol = transaction.get('token_symbol')
metadata_id = transaction.get('metadata_id')
timestamp = datetime.datetime.now().strftime('%d/%m/%y, %H:%M')
transaction_repr = f'{action_tag} {amount} {token_symbol} {direction_tag} {metadata_id} {timestamp}'
parsed_transactions.append(transaction_repr)
return parsed_transactions
def query_statement(blockchain_address: str, limit: int = 9):
"""This function queries cic-eth for a set of chronologically ordered number of transactions associated with
an account.
:param blockchain_address: Ethereum address associated with an account.
:type blockchain_address: str, 0x-hex
:param limit: Number of transactions to be returned.
:type limit: int
"""
logg.debug(f'retrieving statement for address: {blockchain_address}')
chain_str = Chain.spec.__str__()
cic_eth_api = Api(
chain_str=chain_str,
callback_queue='cic-ussd',
callback_task='cic_ussd.tasks.callback_handler.statement_callback',
callback_param=blockchain_address
)
cic_eth_api.list(address=blockchain_address, limit=limit)
def statement_transaction_set(preferred_language: str, transaction_reprs: list):
"""
:param preferred_language:
:type preferred_language:
:param transaction_reprs:
:type transaction_reprs:
:return:
:rtype:
"""
if not transaction_reprs:
return translation_for('helpers.no_transaction_history', preferred_language)
return ''.join(f'{transaction_repr}\n' for transaction_repr in transaction_reprs)