popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

parsing wallets progress (wip)

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-02 18:16:30 +03:00
parent 0aaf0ec021
commit 97a61e4fce
1 changed files with 18 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
crypto/sol/app.py
View File

@ -76,45 +76,49 @@ async def get_token_balance(wallet_address, token_address):
class SolanaEncoder(json.JSONEncoder):
def default(self, obj):
if hasattr(obj, '__dict__'):
return obj.__dict__
return str(obj)
async def get_wallet_balances(wallet_address): async def get_wallet_balances(wallet_address):
balances = {} balances = {}
logging.info(f"Getting balances for wallet: {wallet_address}") logging.info(f"Getting balances for wallet: {wallet_address}")
opts = TokenAccountOpts(
program_id=Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"),
encoding="jsonParsed" # Ensure encoding is set to jsonParsed
)
try: try:
# Get all token accounts for the wallet response = await solana_client.get_token_accounts_by_owner_json_parsed(
response = await solana_client.get_token_accounts_by_owner(
Pubkey.from_string(wallet_address), Pubkey.from_string(wallet_address),
opts, opts=TokenAccountOpts(
program_id=Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
),
commitment=Confirmed commitment=Confirmed
) )
logging.debug(f"Raw response: {json.dumps(response, indent=2, cls=SolanaEncoder)}")
if response.value: if response.value:
for account in response.value: for account in response.value:
if 'parsed' in account.account.data and 'info' in account.account.data.parsed: if 'parsed' in account.account.data and 'info' in account.account.data['parsed']:
mint = account.account.data.parsed.info.mint mint = account.account.data['parsed']['info']['mint']
balance_response = await solana_client.get_token_account_balance(account.pubkey) balance_response = await solana_client.get_token_account_balance(account.pubkey)
if balance_response.value: if balance_response.value:
amount = float(balance_response.value.uiAmount) amount = float(balance_response.value.ui_amount)
balances[mint] = amount balances[mint] = amount
logging.debug(f"Balance for {mint}: {amount}") logging.debug(f"Balance for {mint}: {amount}")
else: else:
logging.warning(f"Unexpected data format for account: {account}") logging.warning(f"Unexpected data format for account: {account}")
sol_balance = await solana_client.get_balance(Pubkey.from_string(wallet_address)) sol_balance = await solana_client.get_balance(Pubkey.from_string(wallet_address))
if sol_balance.value: if sol_balance.value is not None:
balances['SOL'] = sol_balance.value / 1e9 balances['SOL'] = sol_balance.value / 1e9
else:
logging.warning(f"SOL balance response missing for wallet: {wallet_address}")
except Exception as e: except Exception as e:
logging.error(f"Error getting wallet balances: {str(e)}") logging.error(f"Error getting wallet balances: {str(e)}")
return balances return balances
async def get_non_zero_token_balances(wallet_address): async def get_non_zero_token_balances(wallet_address):
non_zero_balances = {} non_zero_balances = {}
logging.info(f"Getting non-zero balances for wallet: {wallet_address}") logging.info(f"Getting non-zero balances for wallet: {wallet_address}")