forked from LBRYCommunity/lbry-sdk
136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
import json
|
|
|
|
from copy import deepcopy
|
|
from lbrynet.conf import CURRENCIES
|
|
from lbrynet.core import utils
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
BITTREX_FEE = 0.0025
|
|
|
|
# Metadata version
|
|
SOURCE_TYPES = ['lbry_sd_hash', 'url', 'btih']
|
|
NAME_ALLOWED_CHARSET = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0987654321-'
|
|
BASE_METADATA_FIELDS = ['title', 'description', 'author', 'language', 'license', 'content-type', 'sources']
|
|
OPTIONAL_METADATA_FIELDS = ['thumbnail', 'preview', 'fee', 'contact', 'pubkey']
|
|
|
|
MV001 = "0.0.1"
|
|
MV002 = "0.0.2"
|
|
CURRENT_METADATA_VERSION = MV002
|
|
|
|
METADATA_REVISIONS = {}
|
|
METADATA_REVISIONS[MV001] = {'required': BASE_METADATA_FIELDS, 'optional': OPTIONAL_METADATA_FIELDS}
|
|
METADATA_REVISIONS[MV002] = {'required': ['nsfw', 'ver'], 'optional': ['license_url']}
|
|
|
|
# Fee version
|
|
BASE_FEE_FIELDS = ['amount', 'address']
|
|
|
|
FV001 = "0.0.1"
|
|
CURRENT_FEE_REVISION = FV001
|
|
|
|
FEE_REVISIONS = {}
|
|
FEE_REVISIONS[FV001] = {'required': BASE_FEE_FIELDS, 'optional': []}
|
|
|
|
|
|
def verify_name_characters(name):
|
|
for c in name:
|
|
assert c in NAME_ALLOWED_CHARSET, "Invalid character"
|
|
return True
|
|
|
|
|
|
class LBRYFeeValidator(dict):
|
|
def __init__(self, fee_dict):
|
|
dict.__init__(self)
|
|
assert len(fee_dict) == 1
|
|
self.fee_version = None
|
|
self.currency_symbol = None
|
|
|
|
fee_to_load = deepcopy(fee_dict)
|
|
|
|
for currency in fee_dict:
|
|
self._verify_fee(currency, fee_to_load)
|
|
|
|
self.amount = self._get_amount()
|
|
self.address = self[self.currency_symbol]['address']
|
|
|
|
def _get_amount(self):
|
|
amt = self[self.currency_symbol]['amount']
|
|
if isinstance(amt, float):
|
|
return amt
|
|
else:
|
|
try:
|
|
return float(amt)
|
|
except TypeError:
|
|
log.error('Failed to convert %s to float', amt)
|
|
raise
|
|
|
|
def _verify_fee(self, currency, f):
|
|
# str in case someone made a claim with a wierd fee
|
|
assert currency in CURRENCIES, "Unsupported currency: %s" % str(currency)
|
|
self.currency_symbol = currency
|
|
self.update({currency: {}})
|
|
for version in FEE_REVISIONS:
|
|
self._load_revision(version, f)
|
|
if not f:
|
|
self.fee_version = version
|
|
break
|
|
assert f[self.currency_symbol] == {}, "Unknown fee keys: %s" % json.dumps(f.keys())
|
|
|
|
def _load_revision(self, version, f):
|
|
for k in FEE_REVISIONS[version]['required']:
|
|
assert k in f[self.currency_symbol], "Missing required fee field: %s" % k
|
|
self[self.currency_symbol].update({k: f[self.currency_symbol].pop(k)})
|
|
for k in FEE_REVISIONS[version]['optional']:
|
|
if k in f[self.currency_symbol]:
|
|
self[self.currency_symbol].update({k: f[self.currency_symbol].pop(k)})
|
|
|
|
|
|
class Metadata(dict):
|
|
@classmethod
|
|
def load_from_hex(cls, metadata):
|
|
return cls(json.loads(metadata.decode('hex')))
|
|
|
|
def __init__(self, metadata):
|
|
dict.__init__(self)
|
|
self.meta_version = None
|
|
metadata_to_load = deepcopy(metadata)
|
|
|
|
self._verify_sources(metadata_to_load)
|
|
self._verify_metadata(metadata_to_load)
|
|
|
|
def _load_revision(self, version, metadata):
|
|
for k in METADATA_REVISIONS[version]['required']:
|
|
assert k in metadata, "Missing required metadata field: %s" % k
|
|
self.update({k: metadata.pop(k)})
|
|
for k in METADATA_REVISIONS[version]['optional']:
|
|
if k == 'fee':
|
|
self._load_fee(metadata)
|
|
elif k in metadata:
|
|
self.update({k: metadata.pop(k)})
|
|
|
|
def _load_fee(self, metadata):
|
|
if 'fee' in metadata:
|
|
self['fee'] = LBRYFeeValidator(metadata.pop('fee'))
|
|
|
|
def _verify_sources(self, metadata):
|
|
assert "sources" in metadata, "No sources given"
|
|
for source in metadata['sources']:
|
|
assert source in SOURCE_TYPES, "Unknown source type"
|
|
|
|
def _verify_metadata(self, metadata):
|
|
for version in METADATA_REVISIONS:
|
|
self._load_revision(version, metadata)
|
|
if not metadata:
|
|
self.meta_version = version
|
|
if utils.version_is_greater_than(self.meta_version, "0.0.1"):
|
|
assert self.meta_version == self['ver'], "version mismatch"
|
|
break
|
|
assert metadata == {}, "Unknown metadata keys: %s" % json.dumps(metadata.keys())
|
|
|
|
def serialize(self):
|
|
return json.dumps(self).encode("hex")
|
|
|
|
def as_json(self):
|
|
return json.dumps(self)
|