diff --git a/blockapi/test/v2/api/nft/test_unisat.py b/blockapi/test/v2/api/nft/test_unisat.py index f0b01774..288519d8 100644 --- a/blockapi/test/v2/api/nft/test_unisat.py +++ b/blockapi/test/v2/api/nft/test_unisat.py @@ -24,92 +24,6 @@ test_nft_type = BtcNftType.COLLECTION -def test_parse_nfts(requests_mock, unisat_client, inscription_data): - """Test basic NFT parsing with valid data""" - requests_mock.get( - f"{unisat_client.api_options.base_url}v1/indexer/address/{nfts_test_address}/inscription-data", - text=inscription_data, - ) - - result = unisat_client.fetch_nfts(nfts_test_address) - assert not result.errors, f"Fetch errors: {result.errors}" - - parsed = unisat_client.parse_nfts(result) - assert not parsed.errors, f"Parse errors: {parsed.errors}" - assert len(parsed.data) == 2 - - # Test first NFT - nft1 = parsed.data[0] - assert ( - nft1.ident - == "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" - ) - assert nft1.collection == "ordinals" - assert nft1.collection_name == "Bitcoin Ordinals" - assert ( - nft1.contract - == "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" - ) - assert nft1.standard == "ordinals" - assert nft1.name == "Ordinal #12345" - assert nft1.amount == 1 - assert nft1.blockchain == Blockchain.BITCOIN - assert nft1.asset_type == AssetType.AVAILABLE - - # Test second NFT - nft2 = parsed.data[1] - assert ( - nft2.ident - == "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" - ) - assert nft2.collection == "ordinals" - assert nft2.collection_name == "Bitcoin Ordinals" - assert ( - nft2.contract - == "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" - ) - assert nft2.standard == "ordinals" - assert nft2.name == "Ordinal #12346" - assert nft2.amount == 1 - assert nft2.blockchain == Blockchain.BITCOIN - assert nft2.asset_type == AssetType.AVAILABLE - - -def test_parse_nfts_edge_cases( - requests_mock, unisat_client, inscription_data_edge_cases -): - """Test NFT parsing with various edge cases""" - requests_mock.get( - f"{unisat_client.api_options.base_url}v1/indexer/address/{nfts_test_address}/inscription-data", - text=inscription_data_edge_cases, - ) - - result = unisat_client.fetch_nfts(nfts_test_address) - assert not result.errors, f"Fetch errors: {result.errors}" - - parsed = unisat_client.parse_nfts(result) - assert not parsed.errors, f"Parse errors: {parsed.errors}" - # Should only parse the last inscription as it's the only one with all required fields - assert len(parsed.data) == 1 - - nft = parsed.data[0] - assert ( - nft.ident - == "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" - ) - assert nft.collection == "ordinals" - assert nft.collection_name == "Bitcoin Ordinals" - assert ( - nft.contract - == "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" - ) - assert nft.standard == "ordinals" - assert nft.name == "Ordinal #2" - assert nft.amount == 1 - assert nft.blockchain == Blockchain.BITCOIN - assert nft.asset_type == AssetType.AVAILABLE - - def test_fetch_collection(requests_mock, unisat_client, collection_stats): requests_mock.post( f"{unisat_client.api_options.base_url}v3/market/collection/auction/collection_statistic", diff --git a/blockapi/v2/api/nft/unisat.py b/blockapi/v2/api/nft/unisat.py index 8080d573..a5a1136d 100644 --- a/blockapi/v2/api/nft/unisat.py +++ b/blockapi/v2/api/nft/unisat.py @@ -1,5 +1,5 @@ import logging -from typing import Optional, Dict, Generator +from typing import Optional, Dict, Generator, Tuple from enum import Enum from datetime import datetime @@ -41,6 +41,9 @@ class UnisatApi(BlockchainApi, INftParser, INftProvider): coin = COIN_BTC + DEFAULT_CID = "uncategorized-ordinals" + DEFAULT_CNAME = "Uncategorized Ordinals" + api_options = ApiOptions( blockchain=Blockchain.BITCOIN, base_url='https://open-api.unisat.io/', @@ -52,6 +55,7 @@ class UnisatApi(BlockchainApi, INftParser, INftProvider): 'get_listings': 'v3/market/collection/auction/list', 'get_offers': 'v3/market/collection/auction/actions', 'get_collection_stats': 'v3/market/collection/auction/collection_statistic', + 'get_collection_summary': '/v3/market/collection/auction/collection_summary', } def __init__( @@ -77,6 +81,8 @@ def __init__( } self.limit = limit + self._collection_map: Dict[str, Tuple[str, str]] | None = None + def fetch_nfts(self, address: str) -> FetchResult: """ Fetch NFTs (inscriptions) owned by the address @@ -96,13 +102,18 @@ def fetch_nfts(self, address: str) -> FetchResult: params = {'size': self.limit, 'cursor': 0} try: - return self.get_data( + result = self.get_data( 'get_nfts', headers=self.headers, params=params, address=address, extra=dict(address=address), ) + + if self._collection_map is None: + self._collection_map = self._build_collection_map(address) + + return result except (HTTPError, ValueError, TypeError) as e: logger.error(f"Error fetching NFTs for address {address}: {str(e)}") return FetchResult(errors=[str(e)]) @@ -133,6 +144,10 @@ def parse_nfts(self, fetch_result: FetchResult) -> ParseResult: errors.append("No data in API response") return ParseResult(data=[], errors=errors) + if self._collection_map is None: + errors.append("Collection map is not initialized.") + return ParseResult(errors=[errors]) + for nft in self._yield_parsed_nfts(inner_data): data.append(nft) @@ -163,10 +178,15 @@ def _yield_parsed_nfts(self, data: Dict) -> Generator[NftToken, None, None]: logger.warning(f"Missing required fields in UTXO data: {utxo}") continue + iid = item["inscriptionId"] + cid, cname = self._collection_map.get( + iid, (self.DEFAULT_CID, self.DEFAULT_CNAME) + ) + yield NftToken.from_api( - ident=item["inscriptionId"], - collection="ordinals", - collection_name="Bitcoin Ordinals", + ident=iid, + collection=cid, + collection_name=cname, contract=utxo["txid"], standard="ordinals", name=f"Ordinal #{item['inscriptionNumber']}", @@ -186,6 +206,37 @@ def _yield_parsed_nfts(self, data: Dict) -> Generator[NftToken, None, None]: logger.warning(f"Error parsing NFT item {item}: {str(e)}") continue + def _build_collection_map(self, address: str) -> Dict[str, Tuple[str, str]]: + """Return a mapping using a single collection_summary call. + + Raises + ------ + ValueError + If UniSat returns a non‑zero `code` (e.g. -119 = address invalid). + """ + try: + resp = self.post( + "get_collection_summary", + json={"address": address}, + headers=self.headers, + ) + except Exception as exc: + raise ValueError(f"UniSat request failed: {exc}") from exc + + if not isinstance(resp, dict): + raise ValueError("Unexpected response object from UniSat") + + if resp.get("code", -1) != 0: + raise ValueError(f"Unisat error {resp.get('code')}: {resp.get('msg')}") + + mapping: Dict[str, Tuple[str, str]] = {} + for col in resp.get("data", {}).get("list", []): + cid = col.get("collectionId", self.DEFAULT_CID) + name = col.get("name", self.DEFAULT_CNAME) + for iid in col.get("ids", []): + mapping[iid] = (cid, name) + return mapping + def fetch_collection(self, collection: str) -> FetchResult: """Fetch collection data from Unisat API.""" try: @@ -212,8 +263,23 @@ def parse_collection(self, fetch_result: FetchResult) -> ParseResult: Returns: ParseResult containing parsed collection data """ - if not fetch_result or not fetch_result.data: - return ParseResult(errors=fetch_result.errors if fetch_result else None) + if not fetch_result: + return ParseResult(errors=["Empty response from UniSat"]) + + unisat_response = fetch_result.data + code = unisat_response.get("code", 0) + + # ----- dummy-result case --------------------------------------- + if ( + code == -1 + and unisat_response.get("msg") == "Internal Server Error" + and unisat_response.get("data") is None + ): + return self._dummy_result() + + # ----- any other UniSat error ------------------------------------------ + if code != 0: + return ParseResult(errors=fetch_result.errors) stats = fetch_result.data.get("data", {}) if not stats: @@ -263,6 +329,33 @@ def parse_collection(self, fetch_result: FetchResult) -> ParseResult: return ParseResult(data=[collection], errors=fetch_result.errors) + def _dummy_result(self) -> ParseResult: + dummy_stats = NftCollectionTotalStats.from_api( + volume="0", + sales_count="0", + owners_count="0", + market_cap="0", + floor_price="0", + average_price="0", + coin=self.coin, + ) + dummy_col = NftCollection.from_api( + ident=self.DEFAULT_CID, + name=self.DEFAULT_CNAME, + contracts=[ + ContractInfo.from_api( + blockchain=Blockchain.BITCOIN, address=self.DEFAULT_CID + ) + ], + image=None, + is_disabled=False, + is_nsfw=False, + blockchain=Blockchain.BITCOIN, + total_stats=dummy_stats, + volumes=NftVolumes.from_api(coin=self.coin), + ) + return ParseResult(data=[dummy_col]) + def fetch_listings( self, nft_type: BtcNftType = BtcNftType.COLLECTION,