status.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  1. #!/usr/bin/env python3
  2. #
  3. # MMGen Wallet, a terminal-based cryptocurrency wallet
  4. # Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
  5. # Licensed under the GNU General Public License, Version 3:
  6. # https://www.gnu.org/licenses
  7. # Public project repositories:
  8. # https://github.com/mmgen/mmgen-wallet
  9. # https://gitlab.com/mmgen/mmgen-wallet
  10. """
  11. proto.btc.tx.status: Bitcoin transaction status class
  12. """
  13. import time
  14. from ....tx import status as TxBase
  15. from ....util import msg, suf, die
  16. from ....util2 import format_elapsed_hr
  17. class Status(TxBase.Status):
  18. async def display(self, *, usr_req=False, return_exit_val=False):
  19. def do_exit(retval, message):
  20. if return_exit_val:
  21. msg(message)
  22. return retval
  23. else:
  24. die(retval, message)
  25. tx = self.tx
  26. class r:
  27. pass
  28. async def is_in_wallet():
  29. try:
  30. ret = await tx.rpc.icall(
  31. 'gettransaction',
  32. txid = tx.coin_txid,
  33. include_watchonly = True,
  34. verbose = False)
  35. except:
  36. return False
  37. if ret.get('confirmations', 0) > 0:
  38. r.confs = ret['confirmations']
  39. return True
  40. else:
  41. return False
  42. async def is_in_utxos():
  43. try:
  44. return 'txid' in await tx.rpc.call('getrawtransaction', tx.coin_txid, True)
  45. except:
  46. return False
  47. async def is_in_mempool():
  48. try:
  49. await tx.rpc.call('getmempoolentry', tx.coin_txid)
  50. return True
  51. except:
  52. return False
  53. async def is_replaced():
  54. if await is_in_mempool():
  55. return False
  56. try:
  57. ret = await tx.rpc.icall(
  58. 'gettransaction',
  59. txid = tx.coin_txid,
  60. include_watchonly = True,
  61. verbose = False)
  62. except:
  63. return False
  64. else:
  65. if 'bip125-replaceable' in ret and ret.get('confirmations', 1) <= 0:
  66. r.replacing_confs = -ret['confirmations']
  67. r.replacing_txs = ret['walletconflicts']
  68. return True
  69. else:
  70. return False
  71. if await is_in_mempool():
  72. if usr_req:
  73. d = await tx.rpc.icall(
  74. 'gettransaction',
  75. txid = tx.coin_txid,
  76. include_watchonly = True,
  77. verbose = False)
  78. rep = ('' if d.get('bip125-replaceable') == 'yes' else 'NOT ') + 'replaceable'
  79. t = d['timereceived']
  80. if tx.cfg.quiet:
  81. msg('Transaction is in mempool')
  82. else:
  83. msg(f'TX status: in mempool, {rep}')
  84. msg('Sent {} ({})'.format(time.strftime('%c', time.gmtime(t)), format_elapsed_hr(t)))
  85. else:
  86. msg('Warning: transaction is in mempool!')
  87. elif await is_in_wallet():
  88. return do_exit(0, f'Transaction has {r.confs} confirmation{suf(r.confs)}')
  89. elif await is_in_utxos():
  90. return do_exit(4, 'ERROR: transaction is in the blockchain (but not in the tracking wallet)!')
  91. elif await is_replaced():
  92. msg('Transaction has been replaced')
  93. msg('Replacement transaction ' + (
  94. f'has {r.replacing_confs} confirmation{suf(r.replacing_confs)}'
  95. if r.replacing_confs else
  96. 'is in mempool'))
  97. if not tx.cfg.quiet:
  98. msg('Replacing transactions:')
  99. d = []
  100. for txid in r.replacing_txs:
  101. try:
  102. d.append(await tx.rpc.call('getmempoolentry', txid))
  103. except:
  104. d.append({})
  105. for txid, mp_entry in zip(r.replacing_txs, d):
  106. msg(f' {txid}' + (' in mempool' if 'height' in mp_entry else ''))
  107. return do_exit(0, '')