When Neopets Metaverse sold NFTs, we would watch the sales happen.

solana_watcher_cog.py

from __future__ import annotations

import asyncio
import contextlib
from typing import Any, Dict, List, NamedTuple, Optional

import discord
from cogs.utils.constants import GuildChannels, GuildEmojis, GuildIDs
from cogs.utils.funcs import Plural
from discord.ext import commands, tasks
from neobot import NeoBot
from solana.rpc.async_api import AsyncClient

ONE_SOL_IN_LAMPERTS = 1_000_000_000


class SolanaAccount(NamedTuple):
    address: str
    title: str
    color: discord.Colour
    emoji: discord.PartialEmoji


SCAMMER = SolanaAccount( # stopped keeping track of this one
    address="AtB6N8htedeCEJQR14V537bHPSaBwyGBuJBhK3K6FHtz",
    title="The Scam Solana Address",
    color=discord.Colour.dark_grey(),
    emoji=GuildEmojis.NEOPETS_RIP,
)

METAVERSE = SolanaAccount(
    address="Fwdp7bSAA1G4EsDn6DCkAuKSBRAJp7BjHutQptzQtzUG",
    title="Neopets Metaverse",
    color=discord.Colour.blue(),
    emoji=GuildEmojis.NEOPETS_MADNIGEL,
)

NFT_HELL = 890338997136654368


class SolanaWatcher(commands.Cog):
    def __init__(self, bot: NeoBot):
        self.bot = bot
        self.solana_watcher.start()

        self.previous_signature: Dict[str, Optional[str]] = {}

    @tasks.loop(seconds=15)
    async def solana_watcher(self):
        await self.bot.wait_until_ready()

        # for account in [SCAMMER, METAVERSE]:  # stopped keeping track of the scam wallet
        for account in [METAVERSE]:
            asyncio.get_event_loop().create_task(self.process_account(account))

    async def process_account(self, account: SolanaAccount) -> None:
        async with AsyncClient("https://api.mainnet-beta.solana.com") as client:
            previous_signature = self.previous_signature.get(account.address)
            data = await client.get_signatures_for_address(
                account.address,
                until=previous_signature,
                limit=1 if previous_signature is None else None,
            )  # if this is the first time, just grab the newest

            # reverse so the newest one is last
            for transaction in data["result"][::-1]:
                if transaction["err"] is not None:
                    continue

                signature = transaction["signature"]

                txn = await client.get_confirmed_transaction(signature)

                if previous_signature is None:
                    self.previous_signature[account.address] = signature
                    continue

                # the payload
                result: Dict[str, Any] = txn["result"]

                # metadata of the transaction (the good stuff)
                meta: Dict[str, Any] = result["meta"]

                # let's skip any errored transactions
                if meta["err"] is not None:
                    continue

                # post-transaction token balances
                post_token_balances: List[Dict[str, Any]] = meta["postTokenBalances"]

                # pre-transaction token balances
                # if the buyer has never purchased this token before, it will be an empty list
                pre_token_balances: List[Dict[str, Any]] = meta["preTokenBalances"]

                # all addresses involved in the transaction
                account_keys: List[str] = result["transaction"]["message"][
                    "accountKeys"
                ]

                # the address of the account doing the purchase
                purchaser_address = account_keys[0]

                # the index of the metaverse wallet in this transaction, to track the income
                metaverse_wallet_index = account_keys.index(METAVERSE.address)

                # a list of the balances of the addresses involved in the transaction, before the transaction goes through
                pre_txn_balances: List[int] = meta["preBalances"]

                # a list of the balances of the addresses involved in the transaction, after the transaction goes through
                post_txn_balances: List[int] = meta["postBalances"]

                bought = 0
                post_token_balance = int(
                    post_token_balances[0]["uiTokenAmount"]["uiAmountString"]
                )
                if len(pre_token_balances):
                    before = int(
                        pre_token_balances[0]["uiTokenAmount"]["uiAmountString"]
                    )
                    bought = post_token_balance - before
                else:
                    bought = post_token_balance

                sol_spent = (
                    post_txn_balances[metaverse_wallet_index]
                    - pre_txn_balances[metaverse_wallet_index]
                ) / ONE_SOL_IN_LAMPERTS

                metaverse_balance = (
                    post_txn_balances[metaverse_wallet_index] / ONE_SOL_IN_LAMPERTS
                )

                if guild := self.bot.get_guild(
                    GuildIDs.SUBREDDIT_SERVER
                    if self.bot.debugging is False
                    else GuildIDs.NEOPETSBOT_TEST_SERVER
                ):
                    nft_hell = guild.get_channel_or_thread(
                        NFT_HELL
                        if self.bot.debugging is False
                        else GuildChannels.TEST_DICESPAM
                    )
                    embed = discord.Embed(color=account.color)
                    embed.description = f"[This address](https://explorer.solana.com/address/{purchaser_address})\njust spent **{sol_spent} $SOL** on **{Plural(bought):token}**,\nthey now have **{Plural(post_token_balance):token}**."
                    embed.set_footer(
                        text=f"Metaverse's total $SOL: {metaverse_balance:,}",
                        icon_url=account.emoji.url,
                    )

                    msg = await nft_hell.send(embed=embed)  # type: ignore

                    self.previous_signature[account.address] = signature

                    with contextlib.suppress(Exception):
                        await msg.add_reaction(account.emoji)  # type: ignore

    def cog_unload(self):
        self.solana_watcher.cancel()


def setup(bot: NeoBot):
    bot.add_cog(SolanaWatcher(bot))