From b26a3340c39fba8a9d05bb92f0e1cafd53674099 Mon Sep 17 00:00:00 2001
From: Moul <moul@moul.re>
Date: Sat, 11 May 2019 15:56:34 +0200
Subject: [PATCH] [feat] #22: Display transactions history in a table
---
silkaj/cli_tools.py | 2 +
silkaj/tx_history.py | 261 +++++++++++++++++++++++++++++++++++++++++++
2 files changed, 263 insertions(+)
create mode 100644 silkaj/tx_history.py
diff --git a/silkaj/cli_tools.py b/silkaj/cli_tools.py
index 8d5e14d2..dbc987c0 100644
--- a/silkaj/cli_tools.py
+++ b/silkaj/cli_tools.py
@@ -20,6 +20,7 @@ along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
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 silkaj.money import cmd_amount
from silkaj.cert import send_certification
from silkaj.commands import (
@@ -88,6 +89,7 @@ cli.add_command(cmd_amount)
cli.add_command(list_blocks)
cli.add_command(send_certification)
cli.add_command(difficulties)
+cli.add_command(transaction_history)
cli.add_command(id_pubkey_correspondence)
cli.add_command(currency_info)
cli.add_command(license_command)
diff --git a/silkaj/tx_history.py b/silkaj/tx_history.py
new file mode 100644
index 00000000..dfeffc3a
--- /dev/null
+++ b/silkaj/tx_history.py
@@ -0,0 +1,261 @@
+"""
+Copyright 2016-2019 Maël Azimi <m.a@moul.re>
+
+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
+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 Silkaj. If not, see <https://www.gnu.org/licenses/>.
+"""
+
+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 silkaj.tools import convert_time, coroutine
+from silkaj.crypto_tools import check_public_key
+from silkaj.wot import identity_of, identities_from_pubkeys
+from silkaj.money import get_amount_from_pubkey, amount_in_current_base, UDValue
+from silkaj.tools import CurrencySymbol
+
+
+@command("history", help="Display transaction history")
+@argument("pubkey")
+@option("--uids", "-u", is_flag=True, help="Display uids")
+@coroutine
+async def transaction_history(pubkey, uids):
+ if not check_public_key(pubkey, True):
+ return
+
+ 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])
+ table.add_rows(txs_list)
+ await client.close()
+ echo_via_pager(header + table.draw())
+
+
+async def generate_header(pubkey, currency_symbol, ud_value):
+ try:
+ idty = await identity_of(pubkey)
+ except:
+ 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\
+".format(
+ uid=idty["uid"],
+ pubkey=pubkey,
+ currency=currency_symbol,
+ 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:
+ received_txs.remove(received_tx)
+
+
+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 = [
+ "Date",
+ "Issuers/Recipients",
+ "Amounts {}".format(currency_symbol),
+ "Amounts UD{}".format(currency_symbol),
+ "Comment",
+ ]
+ 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:
+ issuers.append(issuer)
+ 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"))
+ tx_list.append(str())
+ 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)
+ tx_list.append(received_tx.comment)
+ received_txs_table.append(tx_list)
+
+
+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):
+ pubkeys.append(output.condition.left.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:
+ tx_list.append("Total")
+ amounts = str(total_amount / 100)
+ amounts_ud = str(round(total_amount / ud_value, 2))
+ else:
+ tx_list.append(str())
+ 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
+ )
+ tx_list.append(amounts)
+ tx_list.append(amounts_ud)
+ tx_list.append(sent_tx.comment)
+ sent_txs_table.append(tx_list)
+
+
+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):
+ outputs.append(output)
+ 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)
+ else:
+ 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 ""
--
GitLab