mirror of
https://github.com/lnbits/lnbits-legend.git
synced 2024-11-20 02:28:10 +01:00
161 lines
4.1 KiB
Python
161 lines
4.1 KiB
Python
import json
|
|
import hmac
|
|
import hashlib
|
|
from ecdsa import SECP256k1, SigningKey # type: ignore
|
|
from typing import List, NamedTuple, Optional, Dict
|
|
from sqlite3 import Row
|
|
|
|
from lnbits.settings import WALLET
|
|
|
|
|
|
class User(NamedTuple):
|
|
id: str
|
|
email: str
|
|
extensions: List[str] = []
|
|
wallets: List["Wallet"] = []
|
|
password: Optional[str] = None
|
|
|
|
@property
|
|
def wallet_ids(self) -> List[str]:
|
|
return [wallet.id for wallet in self.wallets]
|
|
|
|
def get_wallet(self, wallet_id: str) -> Optional["Wallet"]:
|
|
w = [wallet for wallet in self.wallets if wallet.id == wallet_id]
|
|
return w[0] if w else None
|
|
|
|
|
|
class Wallet(NamedTuple):
|
|
id: str
|
|
name: str
|
|
user: str
|
|
adminkey: str
|
|
inkey: str
|
|
balance_msat: int
|
|
|
|
@property
|
|
def balance(self) -> int:
|
|
return self.balance_msat // 1000
|
|
|
|
def lnurlauth_key(self, domain: str) -> SigningKey:
|
|
hashing_key = hashlib.sha256(self.id.encode("utf-8")).digest()
|
|
linking_key = hmac.digest(hashing_key, domain.encode("utf-8"), "sha256")
|
|
|
|
return SigningKey.from_string(
|
|
linking_key,
|
|
curve=SECP256k1,
|
|
hashfunc=hashlib.sha256,
|
|
)
|
|
|
|
async def get_payment(self, payment_hash: str) -> Optional["Payment"]:
|
|
from .crud import get_wallet_payment
|
|
|
|
return await get_wallet_payment(self.id, payment_hash)
|
|
|
|
async def get_payments(
|
|
self,
|
|
*,
|
|
complete: bool = True,
|
|
pending: bool = False,
|
|
outgoing: bool = True,
|
|
incoming: bool = True,
|
|
exclude_uncheckable: bool = False,
|
|
) -> List["Payment"]:
|
|
from .crud import get_payments
|
|
|
|
return await get_payments(
|
|
wallet_id=self.id,
|
|
complete=complete,
|
|
pending=pending,
|
|
outgoing=outgoing,
|
|
incoming=incoming,
|
|
exclude_uncheckable=exclude_uncheckable,
|
|
)
|
|
|
|
|
|
class Payment(NamedTuple):
|
|
checking_id: str
|
|
pending: bool
|
|
amount: int
|
|
fee: int
|
|
memo: str
|
|
time: int
|
|
bolt11: str
|
|
preimage: str
|
|
payment_hash: str
|
|
extra: Dict
|
|
wallet_id: str
|
|
webhook: str
|
|
webhook_status: int
|
|
|
|
@classmethod
|
|
def from_row(cls, row: Row):
|
|
return cls(
|
|
checking_id=row["checking_id"],
|
|
payment_hash=row["hash"] or "0" * 64,
|
|
bolt11=row["bolt11"] or "",
|
|
preimage=row["preimage"] or "0" * 64,
|
|
extra=json.loads(row["extra"] or "{}"),
|
|
pending=row["pending"],
|
|
amount=row["amount"],
|
|
fee=row["fee"],
|
|
memo=row["memo"],
|
|
time=row["time"],
|
|
wallet_id=row["wallet"],
|
|
webhook=row["webhook"],
|
|
webhook_status=row["webhook_status"],
|
|
)
|
|
|
|
@property
|
|
def tag(self) -> Optional[str]:
|
|
return self.extra.get("tag")
|
|
|
|
@property
|
|
def msat(self) -> int:
|
|
return self.amount
|
|
|
|
@property
|
|
def sat(self) -> int:
|
|
return self.amount // 1000
|
|
|
|
@property
|
|
def is_in(self) -> bool:
|
|
return self.amount > 0
|
|
|
|
@property
|
|
def is_out(self) -> bool:
|
|
return self.amount < 0
|
|
|
|
@property
|
|
def is_uncheckable(self) -> bool:
|
|
return self.checking_id.startswith("temp_") or self.checking_id.startswith(
|
|
"internal_"
|
|
)
|
|
|
|
async def set_pending(self, pending: bool) -> None:
|
|
from .crud import update_payment_status
|
|
|
|
await update_payment_status(self.checking_id, pending)
|
|
|
|
async def check_pending(self) -> None:
|
|
if self.is_uncheckable:
|
|
return
|
|
|
|
if self.is_out:
|
|
status = await WALLET.get_payment_status(self.checking_id)
|
|
else:
|
|
status = await WALLET.get_invoice_status(self.checking_id)
|
|
|
|
print(
|
|
f" - checking '{'in' if self.is_in else 'out'}' {self.checking_id}: {status}"
|
|
)
|
|
|
|
if self.is_out and status.failed:
|
|
await self.delete()
|
|
elif not status.pending:
|
|
await self.set_pending(status.pending)
|
|
|
|
async def delete(self) -> None:
|
|
from .crud import delete_payment
|
|
|
|
await delete_payment(self.checking_id)
|