@ -1,7 +1,7 @@
import trio # type: ignore
import httpx
from os import getenv
from typing import Optional , Dict , AsyncGenerator
from requests import post
from . base import InvoiceResponse , PaymentResponse , PaymentStatus , Wallet
@ -12,8 +12,9 @@ class LntxbotWallet(Wallet):
def __init__ ( self ) :
endpoint = getenv ( " LNTXBOT_API_ENDPOINT " )
self . endpoint = endpoint [ : - 1 ] if endpoint . endswith ( " / " ) else endpoint
self . auth_admin = { " Authorization " : f " Basic { getenv ( ' LNTXBOT_ADMIN_KEY ' ) } " }
self . auth_invoice = { " Authorization " : f " Basic { getenv ( ' LNTXBOT_INVOICE_KEY ' ) } " }
key = getenv ( " LNTXBOT_KEY " ) or getenv ( " LNTXBOT_ADMIN_KEY " ) or getenv ( " LNTXBOT_INVOICE_KEY " )
self . auth = { " Authorization " : f " Basic { key } " }
def create_invoice (
self , amount : int , memo : Optional [ str ] = None , description_hash : Optional [ bytes ] = None
@ -24,44 +25,47 @@ class LntxbotWallet(Wallet):
else :
data [ " memo " ] = memo or " "
r = post (
r = httpx . post (
url = f " { self . endpoint } /addinvoice " ,
headers = self . auth_invoice ,
headers = self . auth ,
json = data ,
)
ok , checking_id , payment_request , error_message = r . ok , None , None , None
if r . ok :
data = r . json ( )
checking_id , payment_request = data [ " payment_hash " ] , data [ " pay_req " ]
if " error " in data and data [ " error " ] :
ok = False
if r . is_error :
try :
data = r . json ( )
error_message = data [ " message " ]
except :
error_message = r . text
pass
return InvoiceResponse ( ok , checking_id , payment_request , error_message )
return InvoiceResponse ( False , None , None , error_message )
data = r . json ( )
return InvoiceResponse ( True , data [ " payment_hash " ] , data [ " pay_req " ] , None )
def pay_invoice ( self , bolt11 : str ) - > PaymentResponse :
r = post ( url = f " { self . endpoint } /payinvoice " , headers = self . auth_admin , json = { " invoice " : bolt11 } )
ok , checking_id , fee_msat , error_message = r . ok , None , 0 , None
r = httpx . post ( url = f " { self . endpoint } /payinvoice " , headers = self . auth , json = { " invoice " : bolt11 } )
if r . ok :
data = r . json ( )
if r . is_error :
try :
data = r . json ( )
error_message = data [ " message " ]
except :
error_message = r . text
pass
if " payment_hash " in data :
checking_id , fee_msat = data [ " decoded " ] [ " payment_hash " ] , data [ " fee_msat " ]
elif " error " in data and data [ " error " ] :
ok , error_message = False , data [ " message " ]
return PaymentResponse ( False , None , 0 , error_message )
return PaymentResponse ( ok , checking_id , fee_msat , error_message )
data = r . json ( )
return PaymentResponse ( True , data [ " decoded " ] [ " payment_hash " ] , data [ " fee_msat " ] , None )
def get_invoice_status ( self , checking_id : str ) - > PaymentStatus :
r = post ( url = f " { self . endpoint } /invoicestatus/ { checking_id } ?wait=false " , headers = self . auth_invoice )
if not r . ok or " error " in r . json ( ) :
return PaymentStatus ( None )
r = httpx . post ( url = f " { self . endpoint } /invoicestatus/ { checking_id } ?wait=false " , headers = self . auth )
data = r . json ( )
if r . is_error or " error " in data :
return PaymentStatus ( None )
if " preimage " not in data :
return PaymentStatus ( False )
@ -69,13 +73,14 @@ class LntxbotWallet(Wallet):
return PaymentStatus ( True )
def get_payment_status ( self , checking_id : str ) - > PaymentStatus :
r = post ( url = f " { self . endpoint } /paymentstatus/ { checking_id } " , headers = self . auth_invoice )
r = httpx . post ( url = f " { self . endpoint } /paymentstatus/ { checking_id } " , headers = self . auth )
if not r . ok or " error " in r . json ( ) :
data = r . json ( )
if r . is_error or " error " in data :
return PaymentStatus ( None )
statuses = { " complete " : True , " failed " : False , " pending " : None , " unknown " : None }
return PaymentStatus ( statuses [ r . json ( ) . get ( " status " , " unknown " ) ] )
return PaymentStatus ( statuses [ data . get ( " status " , " unknown " ) ] )
async def paid_invoices_stream ( self ) - > AsyncGenerator [ str , None ] :
print ( " lntxbot does not support paid invoices stream yet " )
fiatjaf
4 years ago