Coverage for silkaj/money/history.py: 83%
151 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-20 12:29 +0000
1# Copyright 2016-2025 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/>.
16import csv
17from operator import eq, itemgetter, ne, neg
18from pathlib import Path
19from typing import Any, Optional
20from urllib.error import HTTPError
22import arrow
23import rich_click as click
24from duniterpy.api.bma.tx import history
25from duniterpy.api.client import Client
26from duniterpy.documents.transaction import OutputSource, Transaction
27from duniterpy.grammars.output import Condition
29from silkaj.constants import ALL, ALL_DIGITAL
30from silkaj.money.tools import (
31 amount_in_current_base,
32 get_amount_from_pubkey,
33 get_ud_value,
34)
35from silkaj.network import client_instance
36from silkaj.public_key import (
37 check_pubkey_format,
38 gen_pubkey_checksum,
39 validate_checksum,
40)
41from silkaj.tools import get_currency_symbol
42from silkaj.tui import Table
43from silkaj.wot import tools as wt
46@click.command("history", help="Display transaction history")
47@click.argument("pubkey")
48@click.option("--uids", "-u", is_flag=True, help="Display uids")
49@click.option("--full-pubkey", "-f", is_flag=True, help="Display full-length pubkeys")
50@click.option(
51 "--csv-file",
52 "--csv",
53 type=click.Path(exists=False, writable=True, dir_okay=False, path_type=Path),
54 help="Write in specified file name in CSV (Comma-separated values) format the history of money movements",
55)
56def transaction_history(
57 pubkey: str,
58 uids: bool,
59 full_pubkey: bool,
60 csv_file: Optional[Path],
61) -> None:
62 if csv_file:
63 full_pubkey = True
65 if check_pubkey_format(pubkey):
66 pubkey = validate_checksum(pubkey)
68 client = client_instance()
69 ud_value = get_ud_value()
70 currency_symbol = get_currency_symbol()
72 received_txs, sent_txs = [], [] # type: list[Transaction], list[Transaction]
73 get_transactions_history(client, pubkey, received_txs, sent_txs)
74 remove_duplicate_txs(received_txs, sent_txs)
76 txs_list = generate_txs_list(
77 received_txs,
78 sent_txs,
79 pubkey,
80 ud_value,
81 currency_symbol,
82 uids,
83 full_pubkey,
84 )
85 table_headers = [
86 "Date",
87 "Issuers/Recipients",
88 f"Amounts {currency_symbol}",
89 f"Amounts UD{currency_symbol}",
90 "Comment",
91 ]
92 if csv_file:
93 txs_list.insert(0, table_headers)
94 with csv_file.open("w", encoding="utf-8") as f:
95 writer = csv.writer(f)
96 writer.writerows(txs_list)
97 click.echo(f"{csv_file} file successfully saved!")
98 else:
99 table = Table()
100 table.fill_rows(txs_list, table_headers)
101 header = generate_header(pubkey, currency_symbol, ud_value)
102 click.echo_via_pager(header + table.draw())
105def generate_header(pubkey: str, currency_symbol: str, ud_value: int) -> str:
106 try:
107 idty = wt.identity_of(pubkey)
108 except HTTPError:
109 idty = {"uid": ""}
110 balance = get_amount_from_pubkey(pubkey)
111 balance_ud = round(balance[1] / ud_value, 2)
112 date = arrow.now().format(ALL)
113 return f'Transactions history from: {idty["uid"]} {gen_pubkey_checksum(pubkey)}\n\
114Current balance: {balance[1] / 100} {currency_symbol}, {balance_ud} UD {currency_symbol} on {date}\n'
117def get_transactions_history(
118 client: Client,
119 pubkey: str,
120 received_txs: list,
121 sent_txs: list,
122) -> None:
123 """
124 Get transaction history
125 Store txs in Transaction object
126 """
127 tx_history = client(history, pubkey)
128 currency = tx_history["currency"]
130 for received in tx_history["history"]["received"]:
131 received_txs.append(Transaction.from_bma_history(received, currency))
132 for sent in tx_history["history"]["sent"]:
133 sent_txs.append(Transaction.from_bma_history(sent, currency))
136def remove_duplicate_txs(received_txs: list, sent_txs: list) -> None:
137 """
138 Remove duplicate transactions from history
139 Remove received tx which contains output back return
140 that we don`t want to displayed
141 A copy of received_txs is necessary to remove elements
142 """
143 for received_tx in list(received_txs):
144 if received_tx in sent_txs:
145 received_txs.remove(received_tx)
148def generate_txs_list(
149 received_txs: list[Transaction],
150 sent_txs: list[Transaction],
151 pubkey: str,
152 ud_value: int,
153 currency_symbol: str,
154 uids: bool,
155 full_pubkey: bool,
156) -> list:
157 """
158 Generate information in a list of lists for texttable
159 Merge received and sent txs
160 Sort txs temporarily
161 """
163 received_txs_list, sent_txs_list = (
164 [],
165 [],
166 ) # type: list[Transaction], list[Transaction]
167 parse_received_tx(
168 received_txs_list,
169 received_txs,
170 pubkey,
171 ud_value,
172 uids,
173 full_pubkey,
174 )
175 parse_sent_tx(sent_txs_list, sent_txs, pubkey, ud_value, uids, full_pubkey)
176 txs_list = received_txs_list + sent_txs_list
178 txs_list.sort(key=itemgetter(0), reverse=True)
179 return txs_list
182def parse_received_tx(
183 received_txs_table: list[Transaction],
184 received_txs: list[Transaction],
185 pubkey: str,
186 ud_value: int,
187 uids: bool,
188 full_pubkey: bool,
189) -> None:
190 """
191 Extract issuers` pubkeys
192 Get identities from pubkeys
193 Convert time into human format
194 Assign identities
195 Get amounts and assign amounts and amounts_ud
196 Append comment
197 """
198 issuers = []
199 for received_tx in received_txs:
200 for issuer in received_tx.issuers:
201 issuers.append(issuer)
202 identities = wt.identities_from_pubkeys(issuers, uids)
203 for received_tx in received_txs:
204 tx_list = []
205 tx_list.append(arrow.get(received_tx.time).to("local").format(ALL_DIGITAL))
206 tx_list.append("")
207 for i, issuer in enumerate(received_tx.issuers):
208 tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
209 issuer,
210 identities,
211 full_pubkey,
212 )
213 amounts = tx_amount(received_tx, pubkey, received_func)[0]
214 tx_list.append(amounts / 100)
215 tx_list.append(amounts / ud_value)
216 tx_list.append(received_tx.comment)
217 received_txs_table.append(tx_list)
220def parse_sent_tx(
221 sent_txs_table: list[Transaction],
222 sent_txs: list[Transaction],
223 pubkey: str,
224 ud_value: int,
225 uids: bool,
226 full_pubkey: bool,
227) -> None:
228 """
229 Extract recipients` pubkeys from outputs
230 Get identities from pubkeys
231 Convert time into human format
232 Store "Total" and total amounts according to the number of outputs
233 If not output back return:
234 Assign amounts, amounts_ud, identities, and comment
235 """
236 pubkeys = []
237 for sent_tx in sent_txs:
238 outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
239 for output in outputs:
240 if output_available(output.condition, ne, pubkey):
241 pubkeys.append(output.condition.left.pubkey)
243 identities = wt.identities_from_pubkeys(pubkeys, uids)
244 for sent_tx in sent_txs:
245 tx_list = []
246 tx_list.append(arrow.get(sent_tx.time).to("local").format(ALL_DIGITAL))
248 total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
249 if len(outputs) > 1:
250 tx_list.append("Total")
251 amounts = str(total_amount / 100)
252 amounts_ud = str(round(total_amount / ud_value, 2))
253 else:
254 tx_list.append("")
255 amounts = ""
256 amounts_ud = ""
258 for i, output in enumerate(outputs):
259 if output_available(output.condition, ne, pubkey):
260 amounts += prefix(None, outputs, i) + str(
261 neg(amount_in_current_base(output)) / 100,
262 )
263 amounts_ud += prefix(None, outputs, i) + str(
264 round(neg(amount_in_current_base(output)) / ud_value, 2),
265 )
266 tx_list[1] += prefix(tx_list[1], outputs, 0) + assign_idty_from_pubkey(
267 output.condition.left.pubkey,
268 identities,
269 full_pubkey,
270 )
271 tx_list.append(amounts)
272 tx_list.append(amounts_ud)
273 tx_list.append(sent_tx.comment)
274 sent_txs_table.append(tx_list)
277def tx_amount(
278 tx: list[Transaction],
279 pubkey: str,
280 function: Any,
281) -> tuple[int, list[OutputSource]]:
282 """
283 Determine transaction amount from output sources
284 """
285 amount = 0
286 outputs = []
287 for output in tx.outputs: # type: ignore[attr-defined]
288 if output_available(output.condition, ne, pubkey):
289 outputs.append(output)
290 amount += function(output, pubkey)
291 return amount, outputs
294def received_func(output: OutputSource, pubkey: str) -> int:
295 if output_available(output.condition, eq, pubkey):
296 return amount_in_current_base(output)
297 return 0
300def sent_func(output: OutputSource, pubkey: str) -> int:
301 if output_available(output.condition, ne, pubkey):
302 return neg(amount_in_current_base(output))
303 return 0
306def output_available(condition: Condition, comparison: Any, value: str) -> bool:
307 """
308 Check if output source is available
309 Currently only handle simple SIG condition
310 XHX, CLTV, CSV should be handled when present in the blockchain
311 """
312 if hasattr(condition.left, "pubkey"):
313 return comparison(condition.left.pubkey, value)
314 return False
317def assign_idty_from_pubkey(pubkey: str, identities: list, full_pubkey: bool) -> str:
318 idty = gen_pubkey_checksum(pubkey, short=not full_pubkey)
319 for identity in identities:
320 if pubkey == identity["pubkey"]:
321 pubkey_mod = gen_pubkey_checksum(pubkey, short=not full_pubkey)
322 idty = f'{identity["uid"]} - {pubkey_mod}'
323 return idty
326def prefix(
327 tx_addresses: Optional[str],
328 outputs: Optional[list[OutputSource]],
329 occurence: int,
330) -> str:
331 """
332 Pretty print with texttable
333 Break line when several values in a cell
335 Received tx case, 'outputs' is not defined, then add a breakline
336 between the pubkeys except for the first occurence for multi-sig support
338 Sent tx case, handle "Total" line in case of multi-output txs
339 In case of multiple outputs, there is a "Total" on the top,
340 where there must be a breakline
341 """
343 if not outputs:
344 return "\n" if occurence > 0 else ""
346 if tx_addresses == "Total":
347 return "\n"
348 return "\n" if len(outputs) > 1 else ""