|
| 1 | +from typing import Optional |
| 2 | + |
| 3 | +import aiohttp |
| 4 | +from aiohttp import ClientResponseError |
| 5 | +from aleph_message.models import Chain |
| 6 | + |
| 7 | +from aleph.sdk.conf import settings |
| 8 | +from aleph.sdk.query.filters import PostFilter |
| 9 | +from aleph.sdk.query.responses import Post, PostsResponse |
| 10 | +from aleph.sdk.types import Voucher, VoucherMetadata |
| 11 | + |
| 12 | + |
| 13 | +class Vouchers: |
| 14 | + """ |
| 15 | + This service is made to fetch voucher (SOL / EVM) |
| 16 | + """ |
| 17 | + |
| 18 | + def __init__(self, client): |
| 19 | + self._client = client |
| 20 | + |
| 21 | + # Utils |
| 22 | + def _resolve_address(self, address: str) -> str: |
| 23 | + return address # Not Authenticated client so address need to be given |
| 24 | + |
| 25 | + async def _fetch_voucher_update(self): |
| 26 | + """ |
| 27 | + Fetch the latest EVM voucher update (unfiltered). |
| 28 | + """ |
| 29 | + |
| 30 | + post_filter = PostFilter( |
| 31 | + types=["vouchers-update"], addresses=[settings.VOUCHER_ORIGIN_ADDRESS] |
| 32 | + ) |
| 33 | + vouchers_post: PostsResponse = await self._client.get_posts( |
| 34 | + post_filter=post_filter, page_size=1 |
| 35 | + ) |
| 36 | + |
| 37 | + if not vouchers_post.posts: |
| 38 | + return [] |
| 39 | + |
| 40 | + message_post: Post = vouchers_post.posts[0] |
| 41 | + |
| 42 | + nft_vouchers = message_post.content.get("nft_vouchers", {}) |
| 43 | + return list(nft_vouchers.items()) # [(voucher_id, voucher_data)] |
| 44 | + |
| 45 | + async def _fetch_solana_voucher_list(self): |
| 46 | + """ |
| 47 | + Fetch full Solana voucher registry (unfiltered). |
| 48 | + """ |
| 49 | + try: |
| 50 | + async with aiohttp.ClientSession() as session: |
| 51 | + async with session.get(settings.VOUCHER_SOL_REGISTRY) as resp: |
| 52 | + resp.raise_for_status() |
| 53 | + return await resp.json() |
| 54 | + except ClientResponseError: |
| 55 | + return {} |
| 56 | + |
| 57 | + async def fetch_voucher_metadata( |
| 58 | + self, metadata_id: str |
| 59 | + ) -> Optional[VoucherMetadata]: |
| 60 | + """ |
| 61 | + Fetch metadata for a given voucher. |
| 62 | + """ |
| 63 | + url = f"https://claim.twentysix.cloud/sbt/metadata/{metadata_id}.json" |
| 64 | + try: |
| 65 | + async with aiohttp.ClientSession() as session: |
| 66 | + async with session.get(url) as resp: |
| 67 | + resp.raise_for_status() |
| 68 | + data = await resp.json() |
| 69 | + return VoucherMetadata.model_validate(data) |
| 70 | + except ClientResponseError: |
| 71 | + return None |
| 72 | + |
| 73 | + async def get_solana_vouchers(self, address: str) -> list[Voucher]: |
| 74 | + """ |
| 75 | + Fetch Solana vouchers for a specific address. |
| 76 | + """ |
| 77 | + resolved_address = self._resolve_address(address=address) |
| 78 | + vouchers: list[Voucher] = [] |
| 79 | + |
| 80 | + registry_data = await self._fetch_solana_voucher_list() |
| 81 | + |
| 82 | + claimed_tickets = registry_data.get("claimed_tickets", {}) |
| 83 | + batches = registry_data.get("batches", {}) |
| 84 | + |
| 85 | + for ticket_hash, ticket_data in claimed_tickets.items(): |
| 86 | + claimer = ticket_data.get("claimer") |
| 87 | + if claimer != resolved_address: |
| 88 | + continue |
| 89 | + |
| 90 | + batch_id = ticket_data.get("batch_id") |
| 91 | + metadata_id = None |
| 92 | + |
| 93 | + if str(batch_id) in batches: |
| 94 | + metadata_id = batches[str(batch_id)].get("metadata_id") |
| 95 | + |
| 96 | + if metadata_id: |
| 97 | + metadata = await self.fetch_voucher_metadata(metadata_id) |
| 98 | + if metadata: |
| 99 | + voucher = Voucher( |
| 100 | + id=ticket_hash, |
| 101 | + metadata_id=metadata_id, |
| 102 | + name=metadata.name, |
| 103 | + description=metadata.description, |
| 104 | + external_url=metadata.external_url, |
| 105 | + image=metadata.image, |
| 106 | + icon=metadata.icon, |
| 107 | + attributes=metadata.attributes, |
| 108 | + ) |
| 109 | + vouchers.append(voucher) |
| 110 | + |
| 111 | + return vouchers |
| 112 | + |
| 113 | + async def get_evm_vouchers(self, address: str) -> list[Voucher]: |
| 114 | + """ |
| 115 | + Retrieve vouchers specific to EVM chains for a specific address. |
| 116 | + """ |
| 117 | + resolved_address = self._resolve_address(address=address) |
| 118 | + vouchers: list[Voucher] = [] |
| 119 | + |
| 120 | + nft_vouchers = await self._fetch_voucher_update() |
| 121 | + for voucher_id, voucher_data in nft_vouchers: |
| 122 | + if voucher_data.get("claimer") != resolved_address: |
| 123 | + continue |
| 124 | + |
| 125 | + metadata_id = voucher_data.get("metadata_id") |
| 126 | + metadata = await self.fetch_voucher_metadata(metadata_id) |
| 127 | + if not metadata: |
| 128 | + continue |
| 129 | + |
| 130 | + voucher = Voucher( |
| 131 | + id=voucher_id, |
| 132 | + metadata_id=metadata_id, |
| 133 | + name=metadata.name, |
| 134 | + description=metadata.description, |
| 135 | + external_url=metadata.external_url, |
| 136 | + image=metadata.image, |
| 137 | + icon=metadata.icon, |
| 138 | + attributes=metadata.attributes, |
| 139 | + ) |
| 140 | + vouchers.append(voucher) |
| 141 | + return vouchers |
| 142 | + |
| 143 | + async def fetch_vouchers_by_chain(self, chain: Chain, address: str): |
| 144 | + if chain == Chain.SOL: |
| 145 | + return await self.get_solana_vouchers(address=address) |
| 146 | + else: |
| 147 | + return await self.get_evm_vouchers(address=address) |
| 148 | + |
| 149 | + async def get_vouchers(self, address: str) -> list[Voucher]: |
| 150 | + """ |
| 151 | + Retrieve all vouchers for the account / specific adress, across EVM and Solana chains. |
| 152 | + """ |
| 153 | + vouchers = [] |
| 154 | + |
| 155 | + # Get EVM vouchers |
| 156 | + if address.startswith("0x") and len(address) == 42: |
| 157 | + evm_vouchers = await self.get_evm_vouchers(address=address) |
| 158 | + vouchers.extend(evm_vouchers) |
| 159 | + else: |
| 160 | + # Get Solana vouchers |
| 161 | + solana_vouchers = await self.get_solana_vouchers(address=address) |
| 162 | + vouchers.extend(solana_vouchers) |
| 163 | + |
| 164 | + return vouchers |
0 commit comments