import dataclasses import datetime import email.utils import json import logging from decimal import Decimal from personal_data.data import DeduplicateMode, Scraper from .. import secrets logger = logging.getLogger(__name__) # mainnet: https://reader.partisiablockchain.com # testnet: https://node1.testnet.partisiablockchain.com HOSTNAME = 'reader.partisiablockchain.com' URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local' URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global' def shard_id_for_address(address: str) -> str: return 'shards/Shard2/' # TODO @dataclasses.dataclass(frozen=True) class MpcBalance(Scraper): dataset_name = 'defi_mpc_balance' deduplicate_mode = DeduplicateMode.ONLY_LATEST deduplicate_ignore_columns = ['account.update_time'] def get_json(self, url: str, data: dict) -> tuple[dict, datetime.datetime]: headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', } response = self.session.post(url, headers=headers, data=json.dumps(data)) response.raise_for_status() date_text = response.headers.get('last-modified') or response.headers.get( 'date', ) date = email.utils.parsedate_to_datetime(date_text) json_data = response.json() if json_data is None: msg = 'No result data for ' + url raise Exception(msg) return json_data, date def determine_coins(self) -> list[dict]: data: dict = {'path': []} url = URL_ACCOUNT_PLUGIN_GLOBAL.format( hostname=HOSTNAME, shard='', ) json_data, date = self.get_json(url, data=data) return json_data['coins']['coins'] def scrape(self): address = secrets.PBC_ACCOUNT_ADDRESS coins = self.determine_coins() url = URL_ACCOUNT_PLUGIN.format( hostname=HOSTNAME, shard=shard_id_for_address(address), ) data: dict = { 'path': [ {'type': 'field', 'name': 'accounts'}, {'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address}, ], } account_data, date = self.get_json(url, data=data) data_point = { 'account.address': address, 'account.update_time': date, } data_point['balance.MPC'] = Decimal(account_data['mpcTokens']) / 1000 for coin_idx, amount_data in enumerate(account_data['accountCoins']): coin_data = coins[coin_idx] byoc_balance = Decimal(amount_data['balance']) denominator = Decimal(coin_data['conversionRate']['denominator']) native_balance = byoc_balance / denominator data_point['balance.' + coin_data['symbol']] = native_balance del coin_idx, coin_data yield data_point