Coverage for silkaj/money/history.py: 83%
153 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-22 12:04 +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 pendulum
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, CENT_MULT_TO_UNIT
30from silkaj.money import tools as mt
31from silkaj.network import client_instance
32from silkaj.public_key import (
33 check_pubkey_format,
34 gen_pubkey_checksum,
35 validate_checksum,
36)
37from silkaj.tools import get_currency_symbol
38from silkaj.tui import Table
39from silkaj.wot import tools as wt
42@click.command("history", help="History of wallet money movements")
43@click.argument("pubkey")
44@click.option("--uids", "-u", is_flag=True, help="Display identities username")
45@click.option(
46 "--full-pubkey", "-f", is_flag=True, help="Display full-length public keys"
47)
48@click.option(
49 "--csv-file",
50 "--csv",
51 type=click.Path(exists=False, writable=True, dir_okay=False, path_type=Path),
52 help="Write in specified file name in CSV (Comma-separated values) format the history of money movements",
53)
54def transaction_history(
55 pubkey: str,
56 uids: bool,
57 full_pubkey: bool,
58 csv_file: Optional[Path],
59) -> None:
60 if csv_file:
61 full_pubkey = True
63 if check_pubkey_format(pubkey):
64 pubkey = validate_checksum(pubkey)
66 client = client_instance()
67 ud_value = mt.get_ud_value()
68 currency_symbol = get_currency_symbol()
70 received_txs, sent_txs = [], [] # type: list[Transaction], list[Transaction]
71 get_transactions_history(client, pubkey, received_txs, sent_txs)
72 remove_duplicate_txs(received_txs, sent_txs)
74 txs_list = generate_txs_list(
75 received_txs,
76 sent_txs,
77 pubkey,
78 ud_value,
79 currency_symbol,
80 uids,
81 full_pubkey,
82 )
83 table_headers = [
84 "Date",
85 "Issuers/Recipients",
86 f"Amounts {currency_symbol}",
87 f"Amounts UD{currency_symbol}",
88 "Reference",
89 ]
90 if csv_file:
91 if csv_file.is_file():
92 click.confirm(f"{csv_file} exists, would you like to erase it?", abort=True)
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 = mt.get_amount_from_pubkey(pubkey)
111 balance_ud = round(balance[1] / ud_value, 2)
112 date = pendulum.now().format(ALL)
113 return f"Transactions history from: {idty['uid']} {gen_pubkey_checksum(pubkey)}\n\
114Current balance: {balance[1] / CENT_MULT_TO_UNIT} {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 reference/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(
206 pendulum.from_timestamp(received_tx.time, tz="local").format(ALL_DIGITAL)
207 )
208 tx_list.append("")
209 for i, issuer in enumerate(received_tx.issuers):
210 tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
211 issuer,
212 identities,
213 full_pubkey,
214 )
215 amounts = tx_amount(received_tx, pubkey, received_func)[0]
216 tx_list.append(amounts / CENT_MULT_TO_UNIT)
217 tx_list.append(round(amounts / ud_value, 2))
218 tx_list.append(received_tx.comment)
219 received_txs_table.append(tx_list)
222def parse_sent_tx(
223 sent_txs_table: list[Transaction],
224 sent_txs: list[Transaction],
225 pubkey: str,
226 ud_value: int,
227 uids: bool,
228 full_pubkey: bool,
229) -> None:
230 """
231 Extract recipients` pubkeys from outputs
232 Get identities from pubkeys
233 Convert time into human format
234 Store "Total" and total amounts according to the number of outputs
235 If not output back return:
236 Assign amounts, amounts_ud, identities, and comment
237 """
238 pubkeys = []
239 for sent_tx in sent_txs:
240 outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
241 for output in outputs:
242 if output_available(output.condition, ne, pubkey):
243 pubkeys.append(output.condition.left.pubkey)
245 identities = wt.identities_from_pubkeys(pubkeys, uids)
246 for sent_tx in sent_txs:
247 tx_list = []
248 tx_list.append(
249 pendulum.from_timestamp(sent_tx.time, tz="local").format(ALL_DIGITAL)
250 )
252 total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
253 if len(outputs) > 1:
254 tx_list.append("Total")
255 amounts = str(total_amount / CENT_MULT_TO_UNIT)
256 amounts_ud = str(round(total_amount / ud_value, 2))
257 else:
258 tx_list.append("")
259 amounts = ""
260 amounts_ud = ""
262 for i, output in enumerate(outputs):
263 if output_available(output.condition, ne, pubkey):
264 amounts += prefix(None, outputs, i) + str(
265 neg(mt.amount_in_current_base(output)) / CENT_MULT_TO_UNIT,
266 )
267 amounts_ud += prefix(None, outputs, i) + str(
268 round(neg(mt.amount_in_current_base(output)) / ud_value, 2),
269 )
270 tx_list[1] += prefix(tx_list[1], outputs, 0) + assign_idty_from_pubkey(
271 output.condition.left.pubkey,
272 identities,
273 full_pubkey,
274 )
275 tx_list.append(amounts)
276 tx_list.append(amounts_ud)
277 tx_list.append(sent_tx.comment)
278 sent_txs_table.append(tx_list)
281def tx_amount(
282 tx: list[Transaction],
283 pubkey: str,
284 function: Any,
285) -> tuple[int, list[OutputSource]]:
286 """
287 Determine transaction amount from output sources
288 """
289 amount = 0
290 outputs = []
291 for output in tx.outputs: # type: ignore[attr-defined]
292 if output_available(output.condition, ne, pubkey):
293 outputs.append(output)
294 amount += function(output, pubkey)
295 return amount, outputs
298def received_func(output: OutputSource, pubkey: str) -> int:
299 if output_available(output.condition, eq, pubkey):
300 return mt.amount_in_current_base(output)
301 return 0
304def sent_func(output: OutputSource, pubkey: str) -> int:
305 if output_available(output.condition, ne, pubkey):
306 return neg(mt.amount_in_current_base(output))
307 return 0
310def output_available(condition: Condition, comparison: Any, value: str) -> bool:
311 """
312 Check if output source is available
313 Currently only handle simple SIG condition
314 XHX, CLTV, CSV should be handled when present in the blockchain
315 """
316 if hasattr(condition.left, "pubkey"):
317 return comparison(condition.left.pubkey, value)
318 return False
321def assign_idty_from_pubkey(pubkey: str, identities: list, full_pubkey: bool) -> str:
322 idty = gen_pubkey_checksum(pubkey, short=not full_pubkey)
323 for identity in identities:
324 if pubkey == identity["pubkey"]:
325 pubkey_mod = gen_pubkey_checksum(pubkey, short=not full_pubkey)
326 idty = f"{identity['uid']} - {pubkey_mod}"
327 return idty
330def prefix(
331 tx_addresses: Optional[str],
332 outputs: Optional[list[OutputSource]],
333 occurence: int,
334) -> str:
335 """
336 Pretty print with texttable
337 Break line when several values in a cell
339 Received tx case, 'outputs' is not defined, then add a breakline
340 between the pubkeys except for the first occurence for multi-sig support
342 Sent tx case, handle "Total" line in case of multi-output txs
343 In case of multiple outputs, there is a "Total" on the top,
344 where there must be a breakline
345 """
347 if not outputs:
348 return "\n" if occurence > 0 else ""
350 if tx_addresses == "Total":
351 return "\n"
352 return "\n" if len(outputs) > 1 else ""