From 1dece54e5713ce356735b5b03da761bddae44c4e Mon Sep 17 00:00:00 2001 From: chao-peng-story Date: Mon, 2 Mar 2026 10:57:30 +0800 Subject: [PATCH 1/2] feat(Royalty): add batch_claim_all_revenue functionality Implemented batch_claim_all_revenue method to claim revenue from multiple ancestor IPs in a single operation, with support for both sequential and multicall modes. Changes: - Added multicall and build_multicall_transaction methods to RoyaltyWorkflows_client - Implemented batch_claim_all_revenue in Royalty resource with: - Support for single or multiple ancestor IPs - Optional multicall mode for batching multiple claims - Token aggregation by claimer and token address - Auto-transfer and auto-unwrap functionality - Added _encode_transaction_data() for multicall encoding to avoid premature gas estimation - Added 4 unit tests covering various batch claim scenarios - Added 2 integration tests for single and multiple ancestor scenarios --- .../RoyaltyWorkflows_client.py | 6 + .../resources/Royalty.py | 161 ++++++++++ .../utils/transaction_utils.py | 4 + tests/integration/test_integration_royalty.py | 277 ++++++++++++++++++ .../resources/test_batch_claim_all_revenue.py | 199 +++++++++++++ 5 files changed, 647 insertions(+) create mode 100644 tests/unit/resources/test_batch_claim_all_revenue.py diff --git a/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py b/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py index 1a8270b5..d6c5ae89 100644 --- a/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py +++ b/src/story_protocol_python_sdk/abi/RoyaltyWorkflows/RoyaltyWorkflows_client.py @@ -55,3 +55,9 @@ def build_claimAllRevenue_transaction( return self.contract.functions.claimAllRevenue( ancestorIpId, claimer, childIpIds, royaltyPolicies, currencyTokens ).build_transaction(tx_params) + + def multicall(self, data): + return self.contract.functions.multicall(data).transact() + + def build_multicall_transaction(self, data, tx_params): + return self.contract.functions.multicall(data).build_transaction(tx_params) diff --git a/src/story_protocol_python_sdk/resources/Royalty.py b/src/story_protocol_python_sdk/resources/Royalty.py index 777febbf..735e775c 100644 --- a/src/story_protocol_python_sdk/resources/Royalty.py +++ b/src/story_protocol_python_sdk/resources/Royalty.py @@ -12,6 +12,7 @@ IpRoyaltyVaultImplClient, ) from story_protocol_python_sdk.abi.MockERC20.MockERC20_client import MockERC20Client +from story_protocol_python_sdk.abi.Multicall3.Multicall3_client import Multicall3Client from story_protocol_python_sdk.abi.RoyaltyModule.RoyaltyModule_client import ( RoyaltyModuleClient, ) @@ -56,6 +57,7 @@ def __init__(self, web3: Web3, account, chain_id: int): self.mock_erc20_client = MockERC20Client(web3) self.royalty_policy_lrp_client = RoyaltyPolicyLRPClient(web3) self.wrapped_ip_client = WrappedIPClient(web3) + self.multicall3_client = Multicall3Client(web3) def get_royalty_vault_address(self, ip_id: str) -> str: """ @@ -222,6 +224,165 @@ def claim_all_revenue( except Exception as e: raise ValueError(f"Failed to claim all revenue: {str(e)}") + def batch_claim_all_revenue( + self, + ancestor_ips: list[dict], + claim_options: dict | None = None, + options: dict | None = None, + tx_options: dict | None = None, + ) -> dict: + """ + Batch claims all revenue from the child IPs of multiple ancestor IPs. + If multicall is disabled, it will call claim_all_revenue for each ancestor IP. + Then transfer all claimed tokens to the wallet if the wallet owns the IP or is the claimer. + If claimed token is WIP, it will also be converted back to native tokens. + + Even if there are no child IPs, you must still populate `currency_tokens` in each ancestor IP + with the token addresses you wish to claim. This is required for the claim operation to know which + token balances to process. + + :param ancestor_ips list[dict]: List of ancestor IP configurations, each containing: + :param ip_id str: The IP ID of the ancestor. + :param claimer str: The address of the claimer. + :param child_ip_ids list: List of child IP IDs. + :param royalty_policies list: List of royalty policy addresses. + :param currency_tokens list: List of currency token addresses. + :param claim_options dict: [Optional] Options for auto_transfer_all_claimed_tokens_from_ip and auto_unwrap_ip_tokens. Default values are True. + :param options dict: [Optional] Options for use_multicall_when_possible. Default is True. + :param tx_options dict: [Optional] Transaction options. + :return dict: Dictionary with transaction hashes, receipts, and claimed tokens. + :return tx_hashes list[str]: List of transaction hashes. + :return receipts list[dict]: List of transaction receipts. + :return claimed_tokens list[dict]: Aggregated list of claimed tokens. + """ + try: + tx_hashes = [] + receipts = [] + claimed_tokens = [] + + use_multicall = options.get("use_multicall_when_possible", True) if options else True + + # If only 1 ancestor IP or multicall is disabled, call claim_all_revenue for each + if len(ancestor_ips) == 1 or not use_multicall: + for ancestor_ip in ancestor_ips: + result = self.claim_all_revenue( + ancestor_ip_id=ancestor_ip["ip_id"], + claimer=ancestor_ip["claimer"], + child_ip_ids=ancestor_ip["child_ip_ids"], + royalty_policies=ancestor_ip["royalty_policies"], + currency_tokens=ancestor_ip["currency_tokens"], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + tx_options=tx_options, + ) + tx_hashes.extend(result["tx_hashes"]) + receipts.append(result["receipt"]) + if result.get("claimed_tokens"): + claimed_tokens.extend(result["claimed_tokens"]) + else: + # Batch claimAllRevenue calls into a single multicall + encoded_txs = [] + for ancestor_ip in ancestor_ips: + encoded_data = self.royalty_workflows_client.contract.functions.claimAllRevenue( + validate_address(ancestor_ip["ip_id"]), + validate_address(ancestor_ip["claimer"]), + validate_addresses(ancestor_ip["child_ip_ids"]), + validate_addresses(ancestor_ip["royalty_policies"]), + validate_addresses(ancestor_ip["currency_tokens"]), + )._encode_transaction_data() + encoded_txs.append(encoded_data) + + response = build_and_send_transaction( + self.web3, + self.account, + self.royalty_workflows_client.build_multicall_transaction, + encoded_txs, + tx_options=tx_options, + ) + tx_hashes.append(response["tx_hash"]) + receipts.append(response["tx_receipt"]) + + # Parse claimed tokens from the receipt + claimed_token_logs = self._parse_tx_revenue_token_claimed_event( + response["tx_receipt"] + ) + claimed_tokens.extend(claimed_token_logs) + + # Aggregate claimed tokens by claimer and token address + aggregated_claimed_tokens = {} + for token in claimed_tokens: + key = f"{token['claimer']}_{token['token']}" + if key not in aggregated_claimed_tokens: + aggregated_claimed_tokens[key] = dict(token) + else: + aggregated_claimed_tokens[key]["amount"] += token["amount"] + + aggregated_claimed_tokens = list(aggregated_claimed_tokens.values()) + + # Get unique claimers + claimers = list(set(ancestor_ip["claimer"] for ancestor_ip in ancestor_ips)) + + auto_transfer = ( + claim_options.get("auto_transfer_all_claimed_tokens_from_ip", True) + if claim_options + else True + ) + auto_unwrap = ( + claim_options.get("auto_unwrap_ip_tokens", True) + if claim_options + else True + ) + + wip_claimable_amounts = 0 + + for claimer in claimers: + owns_claimer, is_claimer_ip, ip_account = self._get_claimer_info(claimer) + + # If ownsClaimer is false, skip + if not owns_claimer: + continue + + filter_claimed_tokens = [ + token for token in aggregated_claimed_tokens if token["claimer"] == claimer + ] + + # Transfer claimed tokens from IP to wallet if wallet owns IP + if auto_transfer and is_claimer_ip and owns_claimer: + hashes = self._transfer_claimed_tokens_from_ip_to_wallet( + ip_account, filter_claimed_tokens + ) + tx_hashes.extend(hashes) + + # Sum up the amount of WIP tokens claimed + for token in filter_claimed_tokens: + if token["token"] == WIP_TOKEN_ADDRESS: + wip_claimable_amounts += token["amount"] + + # Unwrap WIP tokens if needed + if wip_claimable_amounts > 0 and auto_unwrap: + hashes = self._unwrap_claimed_tokens_from_ip_to_wallet( + [ + { + "token": WIP_TOKEN_ADDRESS, + "amount": wip_claimable_amounts, + "claimer": self.account.address, + } + ] + ) + tx_hashes.extend(hashes) + + return { + "receipts": receipts, + "claimed_tokens": aggregated_claimed_tokens, + "tx_hashes": tx_hashes, + } + + except Exception as e: + error_msg = str(e).replace("Failed to claim all revenue: ", "").strip() + raise ValueError(f"Failed to batch claim all revenue: {error_msg}") + def transfer_to_vault( self, ip_id: str, diff --git a/src/story_protocol_python_sdk/utils/transaction_utils.py b/src/story_protocol_python_sdk/utils/transaction_utils.py index dfbc74ac..f44ae05e 100644 --- a/src/story_protocol_python_sdk/utils/transaction_utils.py +++ b/src/story_protocol_python_sdk/utils/transaction_utils.py @@ -55,6 +55,10 @@ def _get_transaction_options( if "maxFeePerGas" in tx_options: opts["maxFeePerGas"] = tx_options["maxFeePerGas"] + # Gas limit: use explicit gas if provided to avoid estimation + if "gas" in tx_options: + opts["gas"] = tx_options["gas"] + return opts diff --git a/tests/integration/test_integration_royalty.py b/tests/integration/test_integration_royalty.py index 6c489115..dafbc102 100644 --- a/tests/integration/test_integration_royalty.py +++ b/tests/integration/test_integration_royalty.py @@ -151,6 +151,283 @@ def test_pay_royalty_invalid_amount( amount=-1, ) + def test_batch_claim_all_revenue_single_ancestor(self, story_client: StoryClient): + """Test batch claiming revenue using the same pattern as test_claim_all_revenue + + This test verifies that batch_claim_all_revenue works correctly by: + 1. Creating a derivative chain A->B->C + 2. Using batch_claim_all_revenue to claim revenue for A + 3. Verifying the claimed amount matches expectations + """ + # Create NFT collection + collection_response = story_client.NFTClient.create_nft_collection( + name="batch-claim-test", + symbol="BCT", + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee_recipient=ZERO_ADDRESS, + ) + spg_nft_contract = collection_response["nft_contract"] + + def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): + """Helper to create derivative with WIP tokens""" + minting_fee = story_client.License.predict_minting_license_fee( + licensor_ip_id=parent_ip_id, + license_terms_id=license_terms_id, + amount=1, + ) + amount = minting_fee["amount"] + + story_client.WIP.deposit(amount=amount) + story_client.WIP.approve(spender=spg_nft_contract, amount=amount) + derivative_workflows_address = DerivativeWorkflowsClient( + story_client.web3 + ).contract.address + story_client.WIP.approve( + spender=derivative_workflows_address, amount=amount + ) + + response = story_client.IPAsset.mint_and_register_ip_and_make_derivative( + spg_nft_contract=spg_nft_contract, + deriv_data=DerivativeDataInput( + parent_ip_ids=[parent_ip_id], + license_terms_ids=[license_terms_id], + ), + ) + return response["ip_id"] + + # Define license terms: 100 WIP minting fee + 10% royalty share + license_terms_template = [ + { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + ] + + # Register IP A with PIL terms + ip_a_response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=spg_nft_contract, + terms=license_terms_template, + ) + ip_a = ip_a_response["ip_id"] + license_terms_id = ip_a_response["license_terms_ids"][0] + + # Build derivative chain: A -> B -> C -> D (same as test_claim_all_revenue) + ip_b = wrapper_derivative_with_wip(ip_a, license_terms_id) # B pays 100 WIP + ip_c = wrapper_derivative_with_wip(ip_b, license_terms_id) # C pays 100 WIP (10 to A, 90 to B) + wrapper_derivative_with_wip(ip_c, license_terms_id) # D pays 100 WIP (10 to A, 10 to B, 80 to C) + + # Batch claim revenue for IP A (should get 120 WIP: 100 from B + 10 from C + 10 from D) + # Note: Only pass [ip_b, ip_c] as child_ip_ids, not ip_d, matching test_claim_all_revenue + response = story_client.Royalty.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ip_a, + "claimer": ip_a, + "child_ip_ids": [ip_b, ip_c], + "royalty_policies": [ROYALTY_POLICY, ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS, WIP_TOKEN_ADDRESS], + }, + ], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + ) + + # Verify response + assert response is not None + assert "tx_hashes" in response + assert len(response["tx_hashes"]) >= 1 + assert "receipts" in response + assert len(response["receipts"]) >= 1 + assert "claimed_tokens" in response + assert len(response["claimed_tokens"]) >= 1 + + # Verify IP A received 120 WIP tokens (100 from B + 10 from C + 10 from D) + assert response["claimed_tokens"][0]["amount"] == 120 + + def test_batch_claim_all_revenue_multiple_ancestors(self, story_client: StoryClient): + """Test batch claiming revenue from multiple ancestor IPs + + This test creates two independent derivative chains and claims revenue for both ancestors: + - Chain 1: A1 -> B1 -> C1 -> D1 (A1 gets 120 WIP) + - Chain 2: A2 -> B2 -> C2 (A2 gets 110 WIP) + """ + # Create NFT collection + collection_response = story_client.NFTClient.create_nft_collection( + name="multi-ancestor-test", + symbol="MAT", + is_public_minting=True, + mint_open=True, + contract_uri="test-uri", + mint_fee_recipient=ZERO_ADDRESS, + ) + spg_nft_contract = collection_response["nft_contract"] + + def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): + """Helper to create derivative with WIP tokens""" + minting_fee = story_client.License.predict_minting_license_fee( + licensor_ip_id=parent_ip_id, + license_terms_id=license_terms_id, + amount=1, + ) + amount = minting_fee["amount"] + + story_client.WIP.deposit(amount=amount) + story_client.WIP.approve(spender=spg_nft_contract, amount=amount) + derivative_workflows_address = DerivativeWorkflowsClient( + story_client.web3 + ).contract.address + story_client.WIP.approve( + spender=derivative_workflows_address, amount=amount + ) + + response = story_client.IPAsset.mint_and_register_ip_and_make_derivative( + spg_nft_contract=spg_nft_contract, + deriv_data=DerivativeDataInput( + parent_ip_ids=[parent_ip_id], + license_terms_ids=[license_terms_id], + ), + ) + return response["ip_id"] + + # Define license terms: 100 WIP minting fee + 10% royalty share + license_terms_template = [ + { + "terms": { + "transferable": True, + "royalty_policy": ROYALTY_POLICY, + "default_minting_fee": 100, + "expiration": 0, + "commercial_use": True, + "commercial_attribution": False, + "commercializer_checker": ZERO_ADDRESS, + "commercializer_checker_data": ZERO_ADDRESS, + "commercial_rev_share": 10, + "commercial_rev_ceiling": 0, + "derivatives_allowed": True, + "derivatives_attribution": True, + "derivatives_approval": False, + "derivatives_reciprocal": True, + "derivative_rev_ceiling": 0, + "currency": WIP_TOKEN_ADDRESS, + "uri": "", + }, + "licensing_config": { + "is_set": True, + "minting_fee": 100, + "hook_data": ZERO_ADDRESS, + "licensing_hook": ZERO_ADDRESS, + "commercial_rev_share": 0, + "disabled": False, + "expect_minimum_group_reward_share": 0, + "expect_group_reward_pool": ZERO_ADDRESS, + }, + } + ] + + # Register IP A1 with PIL terms + ip_a1_response = story_client.IPAsset.mint_and_register_ip_asset_with_pil_terms( + spg_nft_contract=spg_nft_contract, + terms=license_terms_template, + ) + ip_a1 = ip_a1_response["ip_id"] + license_terms_id = ip_a1_response["license_terms_ids"][0] + + # Build derivative chain 1: A1 -> B1 -> C1 -> D1 + ip_b1 = wrapper_derivative_with_wip(ip_a1, license_terms_id) + ip_c1 = wrapper_derivative_with_wip(ip_b1, license_terms_id) + wrapper_derivative_with_wip(ip_c1, license_terms_id) # D1 + + # Register IP A2 and attach the same license terms (to avoid duplicate license terms error) + ip_a2_response = story_client.IPAsset.mint_and_register_ip( + spg_nft_contract=spg_nft_contract, + ) + ip_a2 = ip_a2_response["ip_id"] + + # Attach the same license terms to IP A2 + story_client.License.attach_license_terms( + ip_id=ip_a2, + license_template=PIL_LICENSE_TEMPLATE, + license_terms_id=license_terms_id, + ) + + # Build derivative chain 2: A2 -> B2 -> C2 + ip_b2 = wrapper_derivative_with_wip(ip_a2, license_terms_id) + ip_c2 = wrapper_derivative_with_wip(ip_b2, license_terms_id) + + # Batch claim revenue for both ancestors (disable multicall to avoid potential issues) + response = story_client.Royalty.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ip_a1, + "claimer": ip_a1, + "child_ip_ids": [ip_b1, ip_c1], + "royalty_policies": [ROYALTY_POLICY, ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS, WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ip_a2, + "claimer": ip_a2, + "child_ip_ids": [ip_b2], + "royalty_policies": [ROYALTY_POLICY], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + claim_options={ + "auto_transfer_all_claimed_tokens_from_ip": False, + "auto_unwrap_ip_tokens": False, + }, + options={ + "use_multicall_when_possible": False, # Disable multicall for stability + }, + ) + + # Verify response + assert response is not None + assert "tx_hashes" in response + assert len(response["tx_hashes"]) >= 2 # Should have 2 separate txs + assert "receipts" in response + assert len(response["receipts"]) >= 2 + assert "claimed_tokens" in response + assert len(response["claimed_tokens"]) == 2 # Two ancestors claimed + + # Verify both ancestors received their expected amounts + # A1 should get 120 WIP (100 from B1 + 10 from C1 + 10 from D1) + # A2 should get 110 WIP (100 from B2 + 10 from C2) + claimed_amounts = {token["claimer"]: token["amount"] for token in response["claimed_tokens"]} + assert claimed_amounts[ip_a1] == 120 + assert claimed_amounts[ip_a2] == 110 + class TestClaimAllRevenue: def test_claim_all_revenue(self, story_client: StoryClient): diff --git a/tests/unit/resources/test_batch_claim_all_revenue.py b/tests/unit/resources/test_batch_claim_all_revenue.py new file mode 100644 index 00000000..698e7778 --- /dev/null +++ b/tests/unit/resources/test_batch_claim_all_revenue.py @@ -0,0 +1,199 @@ +from unittest.mock import patch + +import pytest + +from story_protocol_python_sdk.resources.Royalty import Royalty +from story_protocol_python_sdk.utils.constants import WIP_TOKEN_ADDRESS +from tests.unit.fixtures.data import ACCOUNT_ADDRESS, ADDRESS, TX_HASH + + +@pytest.fixture(scope="class") +def royalty_client(mock_web3, mock_account): + return Royalty(mock_web3, mock_account, 1) + + +class TestBatchClaimAllRevenue: + def test_batch_claim_all_revenue_single_ancestor(self, royalty_client): + """Test batch claim with single ancestor IP (should call claim_all_revenue)""" + with patch.object( + royalty_client, + "claim_all_revenue", + return_value={ + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + } + ], + ) + assert len(result["tx_hashes"]) >= 1 + assert len(result["receipts"]) == 1 + assert len(result["claimed_tokens"]) == 1 + + def test_batch_claim_all_revenue_multiple_ancestors_with_multicall( + self, royalty_client + ): + """Test batch claim with multiple ancestors using multicall""" + with patch.object( + royalty_client.royalty_workflows_client.contract.functions, + "claimAllRevenue", + return_value=type( + "MockFunction", + (), + { + "build_transaction": lambda self, opts: {"data": "0x1234"}, + "_encode_transaction_data": lambda self: "0x1234", + }, + )(), + ): + with patch.object( + royalty_client, + "_parse_tx_revenue_token_claimed_event", + return_value=[ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + }, + { + "claimer": ACCOUNT_ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 2000, + }, + ], + ): + with patch.object( + royalty_client, "_get_claimer_info", return_value=(False, False, None) + ): + with patch( + "story_protocol_python_sdk.resources.Royalty.build_and_send_transaction", + return_value={"tx_hash": TX_HASH.hex(), "tx_receipt": {"logs": []}}, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ACCOUNT_ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + ) + assert len(result["tx_hashes"]) == 1 + assert len(result["receipts"]) == 1 + assert len(result["claimed_tokens"]) == 2 + + def test_batch_claim_all_revenue_without_multicall(self, royalty_client): + """Test batch claim with multicall disabled""" + with patch.object( + royalty_client, + "claim_all_revenue", + return_value={ + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ACCOUNT_ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + options={"use_multicall_when_possible": False}, + ) + assert len(result["tx_hashes"]) >= 2 + assert len(result["receipts"]) == 2 + + def test_batch_claim_all_revenue_aggregates_tokens(self, royalty_client): + """Test that claimed tokens are properly aggregated""" + with patch.object( + royalty_client, + "claim_all_revenue", + side_effect=[ + { + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 1000, + } + ], + }, + { + "tx_hashes": [TX_HASH.hex()], + "receipt": {"logs": []}, + "claimed_tokens": [ + { + "claimer": ADDRESS, + "token": WIP_TOKEN_ADDRESS, + "amount": 500, + } + ], + }, + ], + ): + result = royalty_client.batch_claim_all_revenue( + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + options={"use_multicall_when_possible": False}, + ) + # Should aggregate tokens for same claimer and token + assert len(result["claimed_tokens"]) == 1 + assert result["claimed_tokens"][0]["amount"] == 1500 From 90070cc28c0179ad8fd4c98b4f6a81f7ce10a28f Mon Sep 17 00:00:00 2001 From: chao-peng-story Date: Tue, 3 Mar 2026 15:50:58 +0800 Subject: [PATCH 2/2] chore: bump version to 0.3.18 --- setup.py | 2 +- src/story_protocol_python_sdk/__init__.py | 2 +- .../GroupingModule/GroupingModule_client.py | 6 +-- .../abi/LicenseToken/LicenseToken_client.py | 4 +- .../resources/IPAsset.py | 6 ++- .../resources/Royalty.py | 12 +++-- .../utils/transaction_utils.py | 9 +--- tests/integration/test_integration_group.py | 54 ++++++++++--------- tests/integration/test_integration_royalty.py | 28 ++++++---- .../resources/test_batch_claim_all_revenue.py | 43 ++++++++------- 10 files changed, 91 insertions(+), 75 deletions(-) diff --git a/setup.py b/setup.py index 8c380f98..852fbe4f 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="story_protocol_python_sdk", - version="0.3.17", + version="0.3.18", packages=find_packages(where="src", exclude=["tests"]), package_dir={"": "src"}, install_requires=["web3>=7.0.0", "pytest", "python-dotenv", "base58"], diff --git a/src/story_protocol_python_sdk/__init__.py b/src/story_protocol_python_sdk/__init__.py index 95a383af..9270cc77 100644 --- a/src/story_protocol_python_sdk/__init__.py +++ b/src/story_protocol_python_sdk/__init__.py @@ -1,4 +1,4 @@ -__version__ = "0.3.17" +__version__ = "0.3.18" from .resources.Dispute import Dispute from .resources.IPAccount import IPAccount diff --git a/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py b/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py index 4149bbbd..d5826c2f 100644 --- a/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py +++ b/src/story_protocol_python_sdk/abi/GroupingModule/GroupingModule_client.py @@ -47,9 +47,9 @@ def removeIp(self, groupIpId, ipIds): return self.contract.functions.removeIp(groupIpId, ipIds).transact() def build_removeIp_transaction(self, groupIpId, ipIds, tx_params): - return self.contract.functions.removeIp( - groupIpId, ipIds - ).build_transaction(tx_params) + return self.contract.functions.removeIp(groupIpId, ipIds).build_transaction( + tx_params + ) def claimReward(self, groupId, token, ipIds): return self.contract.functions.claimReward(groupId, token, ipIds).transact() diff --git a/src/story_protocol_python_sdk/abi/LicenseToken/LicenseToken_client.py b/src/story_protocol_python_sdk/abi/LicenseToken/LicenseToken_client.py index 129bb81b..d768a4c2 100644 --- a/src/story_protocol_python_sdk/abi/LicenseToken/LicenseToken_client.py +++ b/src/story_protocol_python_sdk/abi/LicenseToken/LicenseToken_client.py @@ -32,9 +32,7 @@ def __init__(self, web3: Web3): self.contract = self.web3.eth.contract(address=contract_address, abi=abi) def getTotalTokensByLicensor(self, licensorIpId): - return self.contract.functions.getTotalTokensByLicensor( - licensorIpId - ).call() + return self.contract.functions.getTotalTokensByLicensor(licensorIpId).call() def ownerOf(self, tokenId): return self.contract.functions.ownerOf(tokenId).call() diff --git a/src/story_protocol_python_sdk/resources/IPAsset.py b/src/story_protocol_python_sdk/resources/IPAsset.py index 5066ad28..5a0a3c4f 100644 --- a/src/story_protocol_python_sdk/resources/IPAsset.py +++ b/src/story_protocol_python_sdk/resources/IPAsset.py @@ -2144,7 +2144,7 @@ def is_registered(self, ip_id: str) -> bool: """ if not ip_id: raise ValueError("is_registered: ip_id is required") - + if not self.web3.is_address(ip_id): raise ValueError(f"is_registered: invalid IP ID address format: {ip_id}") @@ -2170,7 +2170,9 @@ def _parse_tx_ip_registered_event(self, tx_receipt: dict) -> list[RegisteredIP]: ) registered_ips.append( RegisteredIP( - ip_id=self.web3.to_checksum_address(event_result["args"]["ipId"]), + ip_id=self.web3.to_checksum_address( + event_result["args"]["ipId"] + ), token_id=event_result["args"]["tokenId"], ) ) diff --git a/src/story_protocol_python_sdk/resources/Royalty.py b/src/story_protocol_python_sdk/resources/Royalty.py index 735e775c..3d03e442 100644 --- a/src/story_protocol_python_sdk/resources/Royalty.py +++ b/src/story_protocol_python_sdk/resources/Royalty.py @@ -260,7 +260,9 @@ def batch_claim_all_revenue( receipts = [] claimed_tokens = [] - use_multicall = options.get("use_multicall_when_possible", True) if options else True + use_multicall = ( + options.get("use_multicall_when_possible", True) if options else True + ) # If only 1 ancestor IP or multicall is disabled, call claim_all_revenue for each if len(ancestor_ips) == 1 or not use_multicall: @@ -338,14 +340,18 @@ def batch_claim_all_revenue( wip_claimable_amounts = 0 for claimer in claimers: - owns_claimer, is_claimer_ip, ip_account = self._get_claimer_info(claimer) + owns_claimer, is_claimer_ip, ip_account = self._get_claimer_info( + claimer + ) # If ownsClaimer is false, skip if not owns_claimer: continue filter_claimed_tokens = [ - token for token in aggregated_claimed_tokens if token["claimer"] == claimer + token + for token in aggregated_claimed_tokens + if token["claimer"] == claimer ] # Transfer claimed tokens from IP to wallet if wallet owns IP diff --git a/src/story_protocol_python_sdk/utils/transaction_utils.py b/src/story_protocol_python_sdk/utils/transaction_utils.py index f44ae05e..ae6796cf 100644 --- a/src/story_protocol_python_sdk/utils/transaction_utils.py +++ b/src/story_protocol_python_sdk/utils/transaction_utils.py @@ -44,9 +44,7 @@ def _get_transaction_options( # Gas: bump for replacement, or use tx_options if bump_gas: try: - opts["gasPrice"] = int( - web3.eth.gas_price * REPLACEMENT_GAS_BUMP_RATIO - ) + opts["gasPrice"] = int(web3.eth.gas_price * REPLACEMENT_GAS_BUMP_RATIO) except Exception: opts["gasPrice"] = web3.to_wei(2, "gwei") else: @@ -65,10 +63,7 @@ def _get_transaction_options( def _is_retryable_send_error(exc: Exception) -> bool: """True if we should retry send (same nonce, higher gas).""" msg = str(exc).lower() - return ( - "replacement transaction underpriced" in msg - or "nonce too low" in msg - ) + return "replacement transaction underpriced" in msg or "nonce too low" in msg def _send_one( diff --git a/tests/integration/test_integration_group.py b/tests/integration/test_integration_group.py index a0b1ab44..c9ed970e 100644 --- a/tests/integration/test_integration_group.py +++ b/tests/integration/test_integration_group.py @@ -376,9 +376,7 @@ def _normalize_address(web3, addr: str) -> str: class TestAddIpsToGroupAndRemoveIpsFromGroup: """Integration tests for add_ips_to_group and remove_ips_from_group with strict on-chain verification.""" - def test_add_ips_to_group( - self, story_client: StoryClient, nft_collection: Address - ): + def test_add_ips_to_group(self, story_client: StoryClient, nft_collection: Address): """Test adding IPs to an existing group; verify chain state via AddedIpToGroup event and get_claimable_reward.""" result1 = GroupTestHelper.mint_and_register_ip_asset_with_pil_terms( story_client, nft_collection @@ -403,17 +401,19 @@ def test_add_ips_to_group( assert isinstance(result["tx_hash"], str) assert len(result["tx_hash"]) > 0 # Strict: verify on-chain AddedIpToGroup event - assert "tx_receipt" in result, "add_ips_to_group must return tx_receipt for verification" + assert ( + "tx_receipt" in result + ), "add_ips_to_group must return tx_receipt for verification" added_events = story_client.Group.get_added_ip_to_group_events( result["tx_receipt"] ) assert len(added_events) == 1 - assert _normalize_address(story_client.web3, added_events[0]["groupId"]) == _normalize_address( - story_client.web3, group_ip_id - ) - assert set(_normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"]) == { - _normalize_address(story_client.web3, ip_id2) - } + assert _normalize_address( + story_client.web3, added_events[0]["groupId"] + ) == _normalize_address(story_client.web3, group_ip_id) + assert set( + _normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"] + ) == {_normalize_address(story_client.web3, ip_id2)} # Verify new member is in group: get_claimable_reward for [ip_id1, ip_id2] should succeed claimable = story_client.Group.get_claimable_reward( group_ip_id=group_ip_id, @@ -454,12 +454,12 @@ def test_add_ips_to_group_with_max_reward_share( result["tx_receipt"] ) assert len(added_events) == 1 - assert _normalize_address(story_client.web3, added_events[0]["groupId"]) == _normalize_address( - story_client.web3, group_ip_id - ) - assert set(_normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"]) == { - _normalize_address(story_client.web3, ip_id2) - } + assert _normalize_address( + story_client.web3, added_events[0]["groupId"] + ) == _normalize_address(story_client.web3, group_ip_id) + assert set( + _normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"] + ) == {_normalize_address(story_client.web3, ip_id2)} def test_remove_ips_from_group( self, story_client: StoryClient, nft_collection: Address @@ -492,12 +492,12 @@ def test_remove_ips_from_group( result["tx_receipt"] ) assert len(removed_events) == 1 - assert _normalize_address(story_client.web3, removed_events[0]["groupId"]) == _normalize_address( - story_client.web3, group_ip_id - ) - assert set(_normalize_address(story_client.web3, a) for a in removed_events[0]["ipIds"]) == { - _normalize_address(story_client.web3, ip_id2) - } + assert _normalize_address( + story_client.web3, removed_events[0]["groupId"] + ) == _normalize_address(story_client.web3, group_ip_id) + assert set( + _normalize_address(story_client.web3, a) for a in removed_events[0]["ipIds"] + ) == {_normalize_address(story_client.web3, ip_id2)} # After remove, only ip_id1 remains; get_claimable_reward for [ip_id1] must succeed claimable = story_client.Group.get_claimable_reward( group_ip_id=group_ip_id, @@ -540,7 +540,9 @@ def test_add_then_remove_ips_from_group( add_result["tx_receipt"] ) assert len(added_events) == 1 - assert set(_normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"]) == { + assert set( + _normalize_address(story_client.web3, a) for a in added_events[0]["ipIds"] + ) == { _normalize_address(story_client.web3, ip_id2), _normalize_address(story_client.web3, ip_id3), } @@ -556,9 +558,9 @@ def test_add_then_remove_ips_from_group( remove_result["tx_receipt"] ) assert len(removed_events) == 1 - assert set(_normalize_address(story_client.web3, a) for a in removed_events[0]["ipIds"]) == { - _normalize_address(story_client.web3, ip_id2) - } + assert set( + _normalize_address(story_client.web3, a) for a in removed_events[0]["ipIds"] + ) == {_normalize_address(story_client.web3, ip_id2)} # Final state: only ip_id1 and ip_id3 are members claimable = story_client.Group.get_claimable_reward( diff --git a/tests/integration/test_integration_royalty.py b/tests/integration/test_integration_royalty.py index dafbc102..0fb3f797 100644 --- a/tests/integration/test_integration_royalty.py +++ b/tests/integration/test_integration_royalty.py @@ -153,7 +153,7 @@ def test_pay_royalty_invalid_amount( def test_batch_claim_all_revenue_single_ancestor(self, story_client: StoryClient): """Test batch claiming revenue using the same pattern as test_claim_all_revenue - + This test verifies that batch_claim_all_revenue works correctly by: 1. Creating a derivative chain A->B->C 2. Using batch_claim_all_revenue to claim revenue for A @@ -242,8 +242,12 @@ def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): # Build derivative chain: A -> B -> C -> D (same as test_claim_all_revenue) ip_b = wrapper_derivative_with_wip(ip_a, license_terms_id) # B pays 100 WIP - ip_c = wrapper_derivative_with_wip(ip_b, license_terms_id) # C pays 100 WIP (10 to A, 90 to B) - wrapper_derivative_with_wip(ip_c, license_terms_id) # D pays 100 WIP (10 to A, 10 to B, 80 to C) + ip_c = wrapper_derivative_with_wip( + ip_b, license_terms_id + ) # C pays 100 WIP (10 to A, 90 to B) + wrapper_derivative_with_wip( + ip_c, license_terms_id + ) # D pays 100 WIP (10 to A, 10 to B, 80 to C) # Batch claim revenue for IP A (should get 120 WIP: 100 from B + 10 from C + 10 from D) # Note: Only pass [ip_b, ip_c] as child_ip_ids, not ip_d, matching test_claim_all_revenue @@ -271,13 +275,15 @@ def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): assert len(response["receipts"]) >= 1 assert "claimed_tokens" in response assert len(response["claimed_tokens"]) >= 1 - + # Verify IP A received 120 WIP tokens (100 from B + 10 from C + 10 from D) assert response["claimed_tokens"][0]["amount"] == 120 - def test_batch_claim_all_revenue_multiple_ancestors(self, story_client: StoryClient): + def test_batch_claim_all_revenue_multiple_ancestors( + self, story_client: StoryClient + ): """Test batch claiming revenue from multiple ancestor IPs - + This test creates two independent derivative chains and claims revenue for both ancestors: - Chain 1: A1 -> B1 -> C1 -> D1 (A1 gets 120 WIP) - Chain 2: A2 -> B2 -> C2 (A2 gets 110 WIP) @@ -373,7 +379,7 @@ def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): spg_nft_contract=spg_nft_contract, ) ip_a2 = ip_a2_response["ip_id"] - + # Attach the same license terms to IP A2 story_client.License.attach_license_terms( ip_id=ip_a2, @@ -383,7 +389,7 @@ def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): # Build derivative chain 2: A2 -> B2 -> C2 ip_b2 = wrapper_derivative_with_wip(ip_a2, license_terms_id) - ip_c2 = wrapper_derivative_with_wip(ip_b2, license_terms_id) + wrapper_derivative_with_wip(ip_b2, license_terms_id) # Batch claim revenue for both ancestors (disable multicall to avoid potential issues) response = story_client.Royalty.batch_claim_all_revenue( @@ -420,11 +426,13 @@ def wrapper_derivative_with_wip(parent_ip_id, license_terms_id): assert len(response["receipts"]) >= 2 assert "claimed_tokens" in response assert len(response["claimed_tokens"]) == 2 # Two ancestors claimed - + # Verify both ancestors received their expected amounts # A1 should get 120 WIP (100 from B1 + 10 from C1 + 10 from D1) # A2 should get 110 WIP (100 from B2 + 10 from C2) - claimed_amounts = {token["claimer"]: token["amount"] for token in response["claimed_tokens"]} + claimed_amounts = { + token["claimer"]: token["amount"] for token in response["claimed_tokens"] + } assert claimed_amounts[ip_a1] == 120 assert claimed_amounts[ip_a2] == 110 diff --git a/tests/unit/resources/test_batch_claim_all_revenue.py b/tests/unit/resources/test_batch_claim_all_revenue.py index 698e7778..de506f1f 100644 --- a/tests/unit/resources/test_batch_claim_all_revenue.py +++ b/tests/unit/resources/test_batch_claim_all_revenue.py @@ -78,30 +78,35 @@ def test_batch_claim_all_revenue_multiple_ancestors_with_multicall( ], ): with patch.object( - royalty_client, "_get_claimer_info", return_value=(False, False, None) + royalty_client, + "_get_claimer_info", + return_value=(False, False, None), ): with patch( "story_protocol_python_sdk.resources.Royalty.build_and_send_transaction", - return_value={"tx_hash": TX_HASH.hex(), "tx_receipt": {"logs": []}}, + return_value={ + "tx_hash": TX_HASH.hex(), + "tx_receipt": {"logs": []}, + }, ): result = royalty_client.batch_claim_all_revenue( - ancestor_ips=[ - { - "ip_id": ADDRESS, - "claimer": ADDRESS, - "child_ip_ids": [], - "royalty_policies": [], - "currency_tokens": [WIP_TOKEN_ADDRESS], - }, - { - "ip_id": ACCOUNT_ADDRESS, - "claimer": ACCOUNT_ADDRESS, - "child_ip_ids": [], - "royalty_policies": [], - "currency_tokens": [WIP_TOKEN_ADDRESS], - }, - ], - ) + ancestor_ips=[ + { + "ip_id": ADDRESS, + "claimer": ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + { + "ip_id": ACCOUNT_ADDRESS, + "claimer": ACCOUNT_ADDRESS, + "child_ip_ids": [], + "royalty_policies": [], + "currency_tokens": [WIP_TOKEN_ADDRESS], + }, + ], + ) assert len(result["tx_hashes"]) == 1 assert len(result["receipts"]) == 1 assert len(result["claimed_tokens"]) == 2