Commit b26a3340 authored by Moul's avatar Moul

[feat] #22: Display transactions history in a table

parent 67771283
......@@ -20,6 +20,7 @@ along with Silkaj. If not, see <>.
from click import group, help_option, version_option, option, pass_context
from silkaj.tx import send_transaction
from silkaj.tx_history import transaction_history
from import cmd_amount
from silkaj.cert import send_certification
from silkaj.commands import (
......@@ -88,6 +89,7 @@ cli.add_command(cmd_amount)
Copyright 2016-2019 Maël Azimi <>
Silkaj 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.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with Silkaj. If not, see <>.
from click import command, argument, option, echo_via_pager, get_terminal_size
from texttable import Texttable
from operator import itemgetter, neg, eq, ne
from time import time
from duniterpy.api.bma.tx import history
from duniterpy.documents.transaction import Transaction
from silkaj.network_tools import ClientInstance
from import convert_time, coroutine
from silkaj.crypto_tools import check_public_key
from silkaj.wot import identity_of, identities_from_pubkeys
from import get_amount_from_pubkey, amount_in_current_base, UDValue
from import CurrencySymbol
@command("history", help="Display transaction history")
@option("--uids", "-u", is_flag=True, help="Display uids")
async def transaction_history(pubkey, uids):
if not check_public_key(pubkey, True):
client = ClientInstance().client
ud_value = await UDValue().ud_value
currency_symbol = await CurrencySymbol().symbol
header = await generate_header(pubkey, currency_symbol, ud_value)
received_txs, sent_txs = list(), list()
await get_transactions_history(client, pubkey, received_txs, sent_txs)
remove_duplicate_txs(received_txs, sent_txs)
txs_list = await generate_table(
received_txs, sent_txs, pubkey, ud_value, currency_symbol, uids
table = Texttable(max_width=get_terminal_size()[0])
await client.close()
echo_via_pager(header + table.draw())
async def generate_header(pubkey, currency_symbol, ud_value):
idty = await identity_of(pubkey)
idty = dict([("uid", "")])
balance = await get_amount_from_pubkey(pubkey)
return "Transactions history from: {uid} {pubkey}\n\
Current balance: {balance} {currency}, {balance_ud} UD {currency} on the {date}\n\
balance=balance[1] / 100,
balance_ud=round(balance[1] / ud_value, 2),
date=convert_time(time(), "all"),
async def get_transactions_history(client, pubkey, received_txs, sent_txs):
Get transaction history
Store txs in Transaction object
tx_history = await client(history, pubkey)
currency = tx_history["currency"]
for received in tx_history["history"]["received"]:
received_txs.append(Transaction.from_bma_history(currency, received))
for sent in tx_history["history"]["sent"]:
sent_txs.append(Transaction.from_bma_history(currency, sent))
def remove_duplicate_txs(received_txs, sent_txs):
Remove duplicate transactions from history
Remove received tx which contains output back return
that we don’t want to displayed
A copy of received_txs is necessary to remove elements
for received_tx in list(received_txs):
if received_tx in sent_txs:
async def generate_table(
received_txs, sent_txs, pubkey, ud_value, currency_symbol, uids
Generate information in a list of lists for texttabe
Merge received and sent txs
Insert table header at the end not to disturb its generation
Sort txs temporarily
received_txs_table, sent_txs_table = list(), list()
await parse_received_tx(received_txs_table, received_txs, pubkey, ud_value, uids)
await parse_sent_tx(sent_txs_table, sent_txs, pubkey, ud_value, uids)
txs_table = received_txs_table + sent_txs_table
table_titles = [
"Amounts {}".format(currency_symbol),
"Amounts UD{}".format(currency_symbol),
txs_table.insert(0, table_titles)
txs_table.sort(key=itemgetter(0), reverse=True)
return txs_table
async def parse_received_tx(received_txs_table, received_txs, pubkey, ud_value, uids):
Extract issuers’ pubkeys
Get identities from pubkeys
Convert time into human format
Assign identities
Get amounts and assign amounts and amounts_ud
Append comment
issuers = list()
for received_tx in received_txs:
for issuer in received_tx.issuers:
identities = await identities_from_pubkeys(issuers, uids)
for received_tx in received_txs:
tx_list = list()
tx_list.append(convert_time(received_tx.time, "all"))
for i, issuer in enumerate(received_tx.issuers):
tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
issuer, identities
amounts = tx_amount(received_tx, pubkey, received_func)[0]
tx_list.append(amounts / 100)
tx_list.append(amounts / ud_value)
async def parse_sent_tx(sent_txs_table, sent_txs, pubkey, ud_value, uids):
Extract recipients’ pubkeys from outputs
Get identities from pubkeys
Convert time into human format
Store "Total" and total amounts according to the number of outputs
If not output back return:
Assign amounts, amounts_ud, identities, and comment
pubkeys = list()
for sent_tx in sent_txs:
outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
for output in outputs:
if output_available(output.condition, ne, pubkey):
identities = await identities_from_pubkeys(pubkeys, uids)
for sent_tx in sent_txs:
tx_list = list()
tx_list.append(convert_time(sent_tx.time, "all"))
total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
if len(outputs) > 1:
amounts = str(total_amount / 100)
amounts_ud = str(round(total_amount / ud_value, 2))
amounts = str()
amounts_ud = str()
for i, output in enumerate(outputs):
if output_available(output.condition, ne, pubkey):
amounts += prefix(None, outputs, i) + str(
neg(amount_in_current_base(output)) / 100
amounts_ud += prefix(None, outputs, i) + str(
round(neg(amount_in_current_base(output)) / ud_value, 2)
tx_list[1] += prefix(tx_list[1], outputs, 0) + assign_idty_from_pubkey(
output.condition.left.pubkey, identities
def tx_amount(tx, pubkey, function):
Determine transaction amount from output sources
amount = 0
outputs = list()
for output in tx.outputs:
if output_available(output.condition, ne, pubkey):
amount += function(output, pubkey)
return amount, outputs
def received_func(output, pubkey):
if output_available(output.condition, eq, pubkey):
return amount_in_current_base(output)
return 0
def sent_func(output, pubkey):
if output_available(output.condition, ne, pubkey):
return neg(amount_in_current_base(output))
return 0
def output_available(condition, comparison, value):
Check if output source is available
Currently only handle simple SIG condition
XHX, CLTV, CSV should be handled when present in the blockchain
if hasattr(condition.left, "pubkey"):
return comparison(condition.left.pubkey, value)
return False
def assign_idty_from_pubkey(pubkey, identities):
idty = pubkey[:18]
for identity in identities:
if pubkey == identity["pubkey"]:
idty = identity["uid"] + " - " + pubkey[:13]
return idty
def prefix(tx_addresses, outputs, occurence):
Pretty print with texttable
Break line when several values in a cell
Handle Total line in case of multi-output txs
if tx_addresses == "Total":
return "\n"
if not outputs:
return ""
return "\n" if occurence + len(outputs) > 1 else ""
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment