import struct
from binascii import hexlify, unhexlify
from typing import List, Iterable, Optional
from torba.client.basetransaction import BaseTransaction, BaseInput, BaseOutput
from torba.client.hash import hash160
from lbrynet.schema.decode import smart_decode
from lbrynet.schema.claim import ClaimDict
from lbrynet.extras.wallet.account import Account
from lbrynet.extras.wallet.script import InputScript, OutputScript
class Input(BaseInput):
script: InputScript
script_class = InputScript
class Output(BaseOutput):
script: OutputScript
script_class = OutputScript
__slots__ = '_claim_dict', 'channel', 'private_key'
def __init__(self, *args, channel: Optional['Output'] = None,
private_key: Optional[str] = None, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._claim_dict = None
self.channel = channel
self.private_key = private_key
def update_annotations(self, annotated):
super().update_annotations(annotated)
self.channel = annotated.channel if annotated else None
self.private_key = annotated.private_key if annotated else None
def get_fee(self, ledger):
name_fee = 0
if self.script.is_claim_name:
name_fee = len(self.script.values['claim_name']) * ledger.fee_per_name_char
return max(name_fee, super().get_fee(ledger))
@property
def is_claim(self) -> bool:
return self.script.is_claim_name or self.script.is_update_claim
@property
def claim_id(self) -> str:
if self.script.is_claim_name:
claim_id = hash160(self.tx_ref.hash + struct.pack('>I', self.position))
elif self.script.is_update_claim or self.script.is_support_claim:
claim_id = self.script.values['claim_id']
else:
raise ValueError('No claim_id associated.')
return hexlify(claim_id[::-1]).decode()
@property
def claim_name(self) -> str:
if self.script.is_claim_involved:
return self.script.values['claim_name'].decode()
raise ValueError('No claim_name associated.')
@property
def claim(self) -> ClaimDict:
if self.is_claim:
return smart_decode(self.script.values['claim'])
raise ValueError('Only claim name and claim update have the claim payload.')
@property
def claim_dict(self) -> dict:
if self._claim_dict is None:
self._claim_dict = self.claim.claim_dict
return self._claim_dict
@property
def permanent_url(self) -> str:
if self.script.is_claim_involved:
if self.channel is not None:
return "{}#{}/{}".format(
self.channel.claim_name,
self.channel.claim_id,
self.claim_name
)
return f"{self.claim_name}#{self.claim_id}"
raise ValueError('No claim associated.')
@property
def has_private_key(self):
return self.private_key is not None
@classmethod
def pay_claim_name_pubkey_hash(
cls, amount: int, claim_name: str, claim: bytes, pubkey_hash: bytes) -> 'Output':
script = cls.script_class.pay_claim_name_pubkey_hash(
claim_name.encode(), claim, pubkey_hash)
return cls(amount, script)
@classmethod
def purchase_claim_pubkey_hash(cls, amount: int, claim_id: str, pubkey_hash: bytes) -> 'Output':
script = cls.script_class.purchase_claim_pubkey_hash(unhexlify(claim_id)[::-1], pubkey_hash)
return cls(amount, script)
@classmethod
def pay_update_claim_pubkey_hash(
cls, amount: int, claim_name: str, claim_id: str, claim: bytes, pubkey_hash: bytes) -> 'Output':
script = cls.script_class.pay_update_claim_pubkey_hash(
claim_name.encode(), unhexlify(claim_id)[::-1], claim, pubkey_hash)
return cls(amount, script)
@classmethod
def pay_support_pubkey_hash(cls, amount: int, claim_name: str, claim_id: str, pubkey_hash: bytes) -> 'Output':
script = cls.script_class.pay_support_pubkey_hash(claim_name.encode(), unhexlify(claim_id)[::-1], pubkey_hash)
return cls(amount, script)
class Transaction(BaseTransaction):
input_class = Input
output_class = Output
@classmethod
def pay(cls, amount: int, address: bytes, funding_accounts: List[Account], change_account: Account):
ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account)
output = Output.pay_pubkey_hash(amount, ledger.address_to_hash160(address))
return cls.create([], [output], funding_accounts, change_account)
@classmethod
def claim(cls, name: str, meta: ClaimDict, amount: int, holding_address: bytes,
funding_accounts: List[Account], change_account: Account):
ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account)
claim_output = Output.pay_claim_name_pubkey_hash(
amount, name, meta.serialized, ledger.address_to_hash160(holding_address)
)
return cls.create([], [claim_output], funding_accounts, change_account)
@classmethod
def purchase(cls, claim: Output, amount: int, merchant_address: bytes,
funding_accounts: List[Account], change_account: Account):
ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account)
claim_output = Output.purchase_claim_pubkey_hash(
amount, claim.claim_id, ledger.address_to_hash160(merchant_address)
)
return cls.create([], [claim_output], funding_accounts, change_account)
@classmethod
def update(cls, previous_claim: Output, meta: ClaimDict, amount: int, holding_address: bytes,
funding_accounts: List[Account], change_account: Account):
ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account)
updated_claim = Output.pay_update_claim_pubkey_hash(
amount, previous_claim.claim_name, previous_claim.claim_id,
meta.serialized, ledger.address_to_hash160(holding_address)
)
return cls.create([Input.spend(previous_claim)], [updated_claim], funding_accounts, change_account)
@classmethod
def support(cls, claim_name: str, claim_id: str, amount: int, holding_address: bytes,
funding_accounts: List[Account], change_account: Account):
ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account)
output = Output.pay_support_pubkey_hash(
amount, claim_name, claim_id, ledger.address_to_hash160(holding_address)
)
return cls.create([], [output], funding_accounts, change_account)
@classmethod
def abandon(cls, claims: Iterable[Output], funding_accounts: Iterable[Account], change_account: Account):
return cls.create([Input.spend(txo) for txo in claims], [], funding_accounts, change_account)
@property
def my_inputs(self):
for txi in self.inputs:
if txi.txo_ref.txo is not None and txi.txo_ref.txo.is_my_account:
yield txi
def _filter_my_outputs(self, f):
for txo in self.outputs:
if txo.is_my_account and f(txo.script):
yield txo
def _filter_other_outputs(self, f):
for txo in self.outputs:
if not txo.is_my_account and f(txo.script):
yield txo
@property
def my_claim_outputs(self):
return self._filter_my_outputs(lambda s: s.is_claim_name)
@property
def my_update_outputs(self):
return self._filter_my_outputs(lambda s: s.is_update_claim)
@property
def my_support_outputs(self):
return self._filter_my_outputs(lambda s: s.is_support_claim)
@property
def other_support_outputs(self):
return self._filter_other_outputs(lambda s: s.is_support_claim)
@property
def my_abandon_outputs(self):
for txi in self.inputs:
abandon = txi.txo_ref.txo
if abandon is not None and abandon.is_my_account and abandon.script.is_claim_involved:
is_update = False
if abandon.script.is_claim_name or abandon.script.is_update_claim:
for update in self.my_update_outputs:
if abandon.claim_id == update.claim_id:
is_update = True
break
if not is_update:
yield abandon