Coverage for silkaj/money/history.py: 83%
138 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 18:49 +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 for issuer in received_tx.issuers:
205 tx_list = []
206 tx_list.append(
207 pendulum.from_timestamp(received_tx.time, tz="local").format(
208 ALL_DIGITAL
209 )
210 )
211 tx_list.append(assign_idty_from_pubkey(issuer, identities, full_pubkey))
212 amounts = tx_amount(received_tx, pubkey, received_func)[0]
213 tx_list.append(amounts / CENT_MULT_TO_UNIT)
214 tx_list.append(round(amounts / ud_value, 2))
215 tx_list.append(received_tx.comment)
216 received_txs_table.append(tx_list)
219def parse_sent_tx(
220 sent_txs_table: list[Transaction],
221 sent_txs: list[Transaction],
222 pubkey: str,
223 ud_value: int,
224 uids: bool,
225 full_pubkey: bool,
226) -> None:
227 """
228 Extract recipients` pubkeys from outputs
229 Get identities from pubkeys
230 Convert time into human format
231 If not output back return:
232 Assign amounts, amounts_ud, identities, and comment
233 """
234 pubkeys = []
235 for sent_tx in sent_txs:
236 outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
237 for output in outputs:
238 if output_available(output.condition, ne, pubkey):
239 pubkeys.append(output.condition.left.pubkey)
241 identities = wt.identities_from_pubkeys(pubkeys, uids)
242 for sent_tx in sent_txs:
243 total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
244 for output in outputs:
245 if output_available(output.condition, ne, pubkey):
246 tx_list = []
247 tx_list.append(
248 pendulum.from_timestamp(sent_tx.time, tz="local").format(
249 ALL_DIGITAL
250 )
251 )
252 wallet = assign_idty_from_pubkey(
253 output.condition.left.pubkey, identities, full_pubkey
254 )
255 tx_list.append(wallet)
256 tx_list.append(
257 neg(mt.amount_in_current_base(output)) / CENT_MULT_TO_UNIT,
258 )
259 tx_list.append(
260 round(neg(mt.amount_in_current_base(output)) / ud_value, 2),
261 )
262 tx_list.append(sent_tx.comment)
263 sent_txs_table.append(tx_list)
266def tx_amount(
267 tx: list[Transaction],
268 pubkey: str,
269 function: Any,
270) -> tuple[int, list[OutputSource]]:
271 """
272 Determine transaction amount from output sources
273 """
274 amount = 0
275 outputs = []
276 for output in tx.outputs: # type: ignore[attr-defined]
277 if output_available(output.condition, ne, pubkey):
278 outputs.append(output)
279 amount += function(output, pubkey)
280 return amount, outputs
283def received_func(output: OutputSource, pubkey: str) -> int:
284 if output_available(output.condition, eq, pubkey):
285 return mt.amount_in_current_base(output)
286 return 0
289def sent_func(output: OutputSource, pubkey: str) -> int:
290 if output_available(output.condition, ne, pubkey):
291 return neg(mt.amount_in_current_base(output))
292 return 0
295def output_available(condition: Condition, comparison: Any, value: str) -> bool:
296 """
297 Check if output source is available
298 Currently only handle simple SIG condition
299 XHX, CLTV, CSV should be handled when present in the blockchain
300 """
301 if hasattr(condition.left, "pubkey"):
302 return comparison(condition.left.pubkey, value)
303 return False
306def assign_idty_from_pubkey(pubkey: str, identities: list, full_pubkey: bool) -> str:
307 idty = gen_pubkey_checksum(pubkey, short=not full_pubkey)
308 for identity in identities:
309 if pubkey == identity["pubkey"]:
310 pubkey_mod = gen_pubkey_checksum(pubkey, short=not full_pubkey)
311 idty = f"{identity['uid']} - {pubkey_mod}"
312 return idty