Coverage for silkaj/money/history.py: 74%
140 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-17 17:03 +0000
1# Copyright 2016-2024 Maël Azimi <m.a@moul.re>
2#
3# Silkaj is free software: you can redistribute it and/or modify
4# it under the terms of the GNU Affero General Public License as published by
5# the Free Software Foundation, either version 3 of the License, or
6# (at your option) any later version.
7#
8# Silkaj is distributed in the hope that it will be useful,
9# but WITHOUT ANY WARRANTY; without even the implied warranty of
10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11# GNU Affero General Public License for more details.
12#
13# You should have received a copy of the GNU Affero General Public License
14# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
16from operator import eq, itemgetter, ne, neg
17from typing import Any, List, Optional, Tuple
18from urllib.error import HTTPError
20import rich_click as click
21from duniterpy.api.bma.tx import history
22from duniterpy.api.client import Client
23from duniterpy.documents.transaction import OutputSource, Transaction
24from duniterpy.grammars.output import Condition
25from pendulum import from_timestamp, now
27from silkaj.constants import ALL, ALL_DIGITAL
28from silkaj.money.tools import (
29 amount_in_current_base,
30 get_amount_from_pubkey,
31 get_ud_value,
32)
33from silkaj.network import client_instance
34from silkaj.public_key import (
35 check_pubkey_format,
36 gen_pubkey_checksum,
37 validate_checksum,
38)
39from silkaj.tools import get_currency_symbol
40from silkaj.tui import Table
41from silkaj.wot import tools as wt
44@click.command("history", help="Display transaction history")
45@click.argument("pubkey")
46@click.option("--uids", "-u", is_flag=True, help="Display uids")
47@click.option("--full-pubkey", "-f", is_flag=True, help="Display full-length pubkeys")
48def transaction_history(pubkey: str, uids: bool, full_pubkey: bool) -> None:
49 if check_pubkey_format(pubkey):
50 pubkey = validate_checksum(pubkey)
52 client = client_instance()
53 ud_value = get_ud_value()
54 currency_symbol = get_currency_symbol()
56 header = generate_header(pubkey, currency_symbol, ud_value)
57 received_txs, sent_txs = [], [] # type: List[Transaction], List[Transaction]
58 get_transactions_history(client, pubkey, received_txs, sent_txs)
59 remove_duplicate_txs(received_txs, sent_txs)
61 txs_list = generate_txs_list(
62 received_txs,
63 sent_txs,
64 pubkey,
65 ud_value,
66 currency_symbol,
67 uids,
68 full_pubkey,
69 )
70 table_headers = [
71 "Date",
72 "Issuers/Recipients",
73 f"Amounts {currency_symbol}",
74 f"Amounts UD{currency_symbol}",
75 "Comment",
76 ]
77 table = Table()
78 table.fill_rows(txs_list, table_headers)
79 click.echo_via_pager(header + table.draw())
82def generate_header(pubkey: str, currency_symbol: str, ud_value: int) -> str:
83 try:
84 idty = wt.identity_of(pubkey)
85 except HTTPError:
86 idty = {"uid": ""}
87 balance = get_amount_from_pubkey(pubkey)
88 balance_ud = round(balance[1] / ud_value, 2)
89 date = now().format(ALL)
90 return f'Transactions history from: {idty["uid"]} {gen_pubkey_checksum(pubkey)} \
91 \n\
92Current balance: {balance[1] / 100} {currency_symbol}, {balance_ud} UD {currency_symbol} on {date}\n'
95def get_transactions_history(
96 client: Client,
97 pubkey: str,
98 received_txs: List,
99 sent_txs: List,
100) -> None:
101 """
102 Get transaction history
103 Store txs in Transaction object
104 """
105 tx_history = client(history, pubkey)
106 currency = tx_history["currency"]
108 for received in tx_history["history"]["received"]:
109 received_txs.append(Transaction.from_bma_history(received, currency))
110 for sent in tx_history["history"]["sent"]:
111 sent_txs.append(Transaction.from_bma_history(sent, currency))
114def remove_duplicate_txs(received_txs: List, sent_txs: List) -> None:
115 """
116 Remove duplicate transactions from history
117 Remove received tx which contains output back return
118 that we don`t want to displayed
119 A copy of received_txs is necessary to remove elements
120 """
121 for received_tx in list(received_txs):
122 if received_tx in sent_txs:
123 received_txs.remove(received_tx)
126def generate_txs_list(
127 received_txs: List[Transaction],
128 sent_txs: List[Transaction],
129 pubkey: str,
130 ud_value: int,
131 currency_symbol: str,
132 uids: bool,
133 full_pubkey: bool,
134) -> List:
135 """
136 Generate information in a list of lists for texttable
137 Merge received and sent txs
138 Sort txs temporarily
139 """
141 received_txs_list, sent_txs_list = (
142 [],
143 [],
144 ) # type: List[Transaction], List[Transaction]
145 parse_received_tx(
146 received_txs_list,
147 received_txs,
148 pubkey,
149 ud_value,
150 uids,
151 full_pubkey,
152 )
153 parse_sent_tx(sent_txs_list, sent_txs, pubkey, ud_value, uids, full_pubkey)
154 txs_list = received_txs_list + sent_txs_list
156 txs_list.sort(key=itemgetter(0), reverse=True)
157 return txs_list
160def parse_received_tx(
161 received_txs_table: List[Transaction],
162 received_txs: List[Transaction],
163 pubkey: str,
164 ud_value: int,
165 uids: bool,
166 full_pubkey: bool,
167) -> None:
168 """
169 Extract issuers` pubkeys
170 Get identities from pubkeys
171 Convert time into human format
172 Assign identities
173 Get amounts and assign amounts and amounts_ud
174 Append comment
175 """
176 issuers = []
177 for received_tx in received_txs:
178 for issuer in received_tx.issuers:
179 issuers.append(issuer)
180 identities = wt.identities_from_pubkeys(issuers, uids)
181 for received_tx in received_txs:
182 tx_list = []
183 tx_list.append(from_timestamp(received_tx.time, tz="local").format(ALL_DIGITAL))
184 tx_list.append("")
185 for i, issuer in enumerate(received_tx.issuers):
186 tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
187 issuer,
188 identities,
189 full_pubkey,
190 )
191 amounts = tx_amount(received_tx, pubkey, received_func)[0]
192 tx_list.append(amounts / 100)
193 tx_list.append(amounts / ud_value)
194 tx_list.append(received_tx.comment)
195 received_txs_table.append(tx_list)
198def parse_sent_tx(
199 sent_txs_table: List[Transaction],
200 sent_txs: List[Transaction],
201 pubkey: str,
202 ud_value: int,
203 uids: bool,
204 full_pubkey: bool,
205) -> None:
206 """
207 Extract recipients` pubkeys from outputs
208 Get identities from pubkeys
209 Convert time into human format
210 Store "Total" and total amounts according to the number of outputs
211 If not output back return:
212 Assign amounts, amounts_ud, identities, and comment
213 """
214 pubkeys = []
215 for sent_tx in sent_txs:
216 outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
217 for output in outputs:
218 if output_available(output.condition, ne, pubkey):
219 pubkeys.append(output.condition.left.pubkey)
221 identities = wt.identities_from_pubkeys(pubkeys, uids)
222 for sent_tx in sent_txs:
223 tx_list = []
224 tx_list.append(from_timestamp(sent_tx.time, tz="local").format(ALL_DIGITAL))
226 total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
227 if len(outputs) > 1:
228 tx_list.append("Total")
229 amounts = str(total_amount / 100)
230 amounts_ud = str(round(total_amount / ud_value, 2))
231 else:
232 tx_list.append("")
233 amounts = ""
234 amounts_ud = ""
236 for i, output in enumerate(outputs):
237 if output_available(output.condition, ne, pubkey):
238 amounts += prefix(None, outputs, i) + str(
239 neg(amount_in_current_base(output)) / 100,
240 )
241 amounts_ud += prefix(None, outputs, i) + str(
242 round(neg(amount_in_current_base(output)) / ud_value, 2),
243 )
244 tx_list[1] += prefix(tx_list[1], outputs, 0) + assign_idty_from_pubkey(
245 output.condition.left.pubkey,
246 identities,
247 full_pubkey,
248 )
249 tx_list.append(amounts)
250 tx_list.append(amounts_ud)
251 tx_list.append(sent_tx.comment)
252 sent_txs_table.append(tx_list)
255def tx_amount(
256 tx: List[Transaction],
257 pubkey: str,
258 function: Any,
259) -> Tuple[int, List[OutputSource]]:
260 """
261 Determine transaction amount from output sources
262 """
263 amount = 0
264 outputs = []
265 for output in tx.outputs: # type: ignore[attr-defined]
266 if output_available(output.condition, ne, pubkey):
267 outputs.append(output)
268 amount += function(output, pubkey)
269 return amount, outputs
272def received_func(output: OutputSource, pubkey: str) -> int:
273 if output_available(output.condition, eq, pubkey):
274 return amount_in_current_base(output)
275 return 0
278def sent_func(output: OutputSource, pubkey: str) -> int:
279 if output_available(output.condition, ne, pubkey):
280 return neg(amount_in_current_base(output))
281 return 0
284def output_available(condition: Condition, comparison: Any, value: str) -> bool:
285 """
286 Check if output source is available
287 Currently only handle simple SIG condition
288 XHX, CLTV, CSV should be handled when present in the blockchain
289 """
290 if hasattr(condition.left, "pubkey"):
291 return comparison(condition.left.pubkey, value)
292 return False
295def assign_idty_from_pubkey(pubkey: str, identities: List, full_pubkey: bool) -> str:
296 idty = gen_pubkey_checksum(pubkey, short=not full_pubkey)
297 for identity in identities:
298 if pubkey == identity["pubkey"]:
299 pubkey_mod = gen_pubkey_checksum(pubkey, short=not full_pubkey)
300 idty = f'{identity["uid"]} - {pubkey_mod}'
301 return idty
304def prefix(
305 tx_addresses: Optional[str],
306 outputs: Optional[List[OutputSource]],
307 occurence: int,
308) -> str:
309 """
310 Pretty print with texttable
311 Break line when several values in a cell
313 Received tx case, 'outputs' is not defined, then add a breakline
314 between the pubkeys except for the first occurence for multi-sig support
316 Sent tx case, handle "Total" line in case of multi-output txs
317 In case of multiple outputs, there is a "Total" on the top,
318 where there must be a breakline
319 """
321 if not outputs:
322 return "\n" if occurence > 0 else ""
324 if tx_addresses == "Total":
325 return "\n"
326 return "\n" if len(outputs) > 1 else ""