fix(unisat): cursor fix

* fix(unisat): cursor fix

* fix(unisat): cursor fix

* fix(unisat): cursor fix

* fix(unisat): cursor fix

* fix(unisat): cursor fix

* fix(unisat): cursor fix

Author: SimSef

blockapi/test/v2/api/nft/test_unisat.py

@@ -53,7 +53,6 @@ def test_parse_nfts(requests_mock, unisat_client, inscription_data):
     assert nft1.standard == "ordinals"
     assert nft1.name == "Ordinal #12345"
     assert nft1.amount == 1
-    assert nft1.updated_time == 1672531200
     assert nft1.blockchain == Blockchain.BITCOIN
     assert nft1.asset_type == AssetType.AVAILABLE
 
@@ -72,7 +71,6 @@ def test_parse_nfts(requests_mock, unisat_client, inscription_data):
     assert nft2.standard == "ordinals"
     assert nft2.name == "Ordinal #12346"
     assert nft2.amount == 1
-    assert nft2.updated_time == 1672531300
     assert nft2.blockchain == Blockchain.BITCOIN
     assert nft2.asset_type == AssetType.AVAILABLE
 
@@ -108,7 +106,6 @@ def test_parse_nfts_edge_cases(
     assert nft.standard == "ordinals"
     assert nft.name == "Ordinal #2"
     assert nft.amount == 1
-    assert nft.updated_time == 1234567890
     assert nft.blockchain == Blockchain.BITCOIN
     assert nft.asset_type == AssetType.AVAILABLE
 

blockapi/v2/api/nft/unisat.py

@@ -54,7 +54,12 @@ class UnisatApi(BlockchainApi, INftParser, INftProvider):
         'get_collection_stats': 'v3/market/collection/auction/collection_statistic',
     }
 
-    def __init__(self, api_key: str, sleep_provider: Optional[ISleepProvider] = None):
+    def __init__(
+        self,
+        api_key: str,
+        sleep_provider: Optional[ISleepProvider] = None,
+        limit: Optional[int] = 10000,
+    ):
         """
         Initialize the Unisat API client
 
@@ -70,17 +75,14 @@ def __init__(self, api_key: str, sleep_provider: Optional[ISleepProvider] = None
             'Authorization': f'Bearer {api_key}',
             'Content-Type': 'application/json',
         }
+        self.limit = limit
 
-    def fetch_nfts(
-        self, address: str, cursor: Optional[int] = None, size: int = 16
-    ) -> FetchResult:
+    def fetch_nfts(self, address: str) -> FetchResult:
         """
         Fetch NFTs (inscriptions) owned by the address
 
         Args:
             address: BTC address to fetch NFTs for
-            cursor: Pagination cursor (offset)
-            size: Number of items to return per page (default: 16)
 
         Returns:
             FetchResult containing the NFT data
@@ -91,9 +93,7 @@ def fetch_nfts(
         if not address:
             raise ValueError("Address is required")
 
-        params = {'size': size}
-        if cursor is not None:
-            params['cursor'] = cursor
+        params = {'size': self.limit, 'cursor': 0}
 
         try:
             return self.get_data(
@@ -116,76 +116,74 @@ def parse_nfts(self, fetch_result: FetchResult) -> ParseResult:
         """Parse NFT data from API response"""
         errors = []
         data = []
-        cursor = None
 
         if not fetch_result.data:
             errors.append("No data in fetch result")
             return ParseResult(data=[], errors=errors)
 
+        if isinstance(fetch_result.data, dict) and "code" in fetch_result.data:
+            api_code = fetch_result.data["code"]
+            if api_code != 0:
+                api_msg = fetch_result.data.get("msg", "Unknown error")
+                errors.append(f"Unisat error {api_code}: {api_msg}")
+                return ParseResult(data=[], errors=errors)
+
         inner_data = fetch_result.data.get("data", {})
         if not inner_data:
             errors.append("No data in API response")
             return ParseResult(data=[], errors=errors)
 
-        cursor = (
-            str(inner_data.get("cursor"))
-            if inner_data.get("cursor") is not None
-            else None
-        )
-
         for nft in self._yield_parsed_nfts(inner_data):
             data.append(nft)
 
-        return ParseResult(data=data, errors=errors, cursor=cursor)
+        return ParseResult(data=data, errors=errors)
 
     def _yield_parsed_nfts(self, data: Dict) -> Generator[NftToken, None, None]:
-        """Yield parsed NFT tokens from API response data"""
+        """Yield parsed NFT tokens from Unisat API response"""
         if not data or "inscription" not in data:
+            logger.warning("No NFT data found in response")
             return
 
         for item in data["inscription"]:
             try:
                 if not all(
                     k in item
-                    for k in [
+                    for k in (
                         "inscriptionId",
                         "inscriptionNumber",
                         "timestamp",
                         "utxo",
-                    ]
+                    )
                 ):
                     logger.warning(f"Missing required fields in NFT data: {item}")
                     continue
 
                 utxo = item["utxo"]
-                if not all(k in utxo for k in ["txid", "address"]):
+                if not all(k in utxo for k in ("txid", "address")):
                     logger.warning(f"Missing required fields in UTXO data: {utxo}")
                     continue
 
-                inscription_number = str(item["inscriptionNumber"])
-                timestamp = str(item["timestamp"])
-
-                yield NftToken(
+                yield NftToken.from_api(
                     ident=item["inscriptionId"],
                     collection="ordinals",
                     collection_name="Bitcoin Ordinals",
                     contract=utxo["txid"],
                     standard="ordinals",
-                    name=f"Ordinal #{inscription_number}",
+                    name=f"Ordinal #{item['inscriptionNumber']}",
                     description="",
                     amount=1,
                     image_url="",
                     metadata_url=None,
-                    metadata={},
-                    updated_time=int(timestamp),
+                    updated_time=str(item["timestamp"]),
                     is_disabled=False,
                     is_nsfw=False,
                     blockchain=Blockchain.BITCOIN,
                     asset_type=AssetType.AVAILABLE,
                     market_url=None,
                 )
+
             except Exception as e:
-                logger.warning(f"Error parsing NFT item {item}: {e}")
+                logger.warning(f"Error parsing NFT item {item}: {str(e)}")
                 continue
 
     def fetch_collection(self, collection: str) -> FetchResult:
@@ -269,8 +267,7 @@ def fetch_listings(
         self,
         nft_type: BtcNftType = BtcNftType.COLLECTION,
         collection: Optional[str] = None,
-        cursor: Optional[str] = None,
-        limit: int = 100,
+        limit: int = 499,
         address: Optional[str] = None,
         tick: Optional[str] = None,
         min_price: Optional[int] = None,
@@ -293,7 +290,6 @@ def fetch_listings(
         Args:
             nft_type: Type of NFT (brc20, domain, collection, arc20, runes)
             collection: Collection ID (slug), optional
-            cursor: Pagination cursor (offset, 'start' parameter)
             limit: Number of items per page
             address: Filter by address
             tick: Filter by tick (for BRC20)
@@ -319,8 +315,6 @@ def fetch_listings(
             nft_type.value if isinstance(nft_type, BtcNftType) else str(nft_type)
         )
 
-        start = int(cursor) if cursor else 0
-
         filter_dict = {"nftType": nft_type_str}
 
         if collection:
@@ -355,10 +349,16 @@ def fetch_listings(
         sort_dict = {}
         sort_dict[sort_by] = sort_order
 
+        if limit >= 500:
+            logger.warning(
+                f"Unisat API limit is 500. You tried to fetch {limit} items. Truncating to 499."
+            )
+            limit = 499
+
         request_body = {
             "filter": filter_dict,
             "sort": sort_dict,
-            "start": start,
+            "start": 0,
             "limit": limit,
         }
 
@@ -394,12 +394,9 @@ def parse_listings(self, fetch_result: FetchResult) -> ParseResult:
             return ParseResult(errors=fetch_result.errors)
 
         items = inner_data.get("list", [])
-        timestamp = inner_data.get("timestamp")
-        cursor = str(timestamp) if timestamp else None
 
         return ParseResult(
             data=list(self._yield_parsed_listings(items)),
-            cursor=cursor,
             errors=fetch_result.errors,
         )
 
@@ -457,8 +454,7 @@ def fetch_offers(
         tick: Optional[str] = None,
         domain_type: Optional[str] = None,
         collection: Optional[str] = None,
-        cursor: Optional[str] = None,
-        limit: int = 100,
+        limit: int = 499,
     ) -> FetchResult:
         """
         Fetch listing events (historical or recent) in a collection.
@@ -471,8 +467,7 @@ def fetch_offers(
             tick: Filter by tick (for BRC20)
             domain_type: Filter by domain type
             collection: Collection ID to filter by
-            cursor: Pagination cursor (offset, 'start' parameter)
-            limit: Number of items per page
+            limit: Number of items
 
         Returns:
             FetchResult containing the listing action data
@@ -482,8 +477,6 @@ def fetch_offers(
             nft_type.value if isinstance(nft_type, BtcNftType) else str(nft_type)
         )
 
-        start = int(cursor) if cursor else 0
-
         filter_dict = {}
         if nft_type_str:
             filter_dict["nftType"] = nft_type_str
@@ -500,9 +493,15 @@ def fetch_offers(
         if collection:
             filter_dict["collectionId"] = collection
 
+        if limit >= 500:
+            logger.warning(
+                f"Unisat API limit is 500. You tried to fetch {limit} items. Truncating to 499."
+            )
+            limit = 499
+
         request_body = {
             "filter": filter_dict,
-            "start": start,
+            "start": 0,
             "limit": limit,
         }