diff --git a/pondereplay/cli.py b/pondereplay/cli.py index 8b22b4c..aab62ad 100644 --- a/pondereplay/cli.py +++ b/pondereplay/cli.py @@ -360,6 +360,137 @@ def replay_history( sys.exit(1) +@cli.command("tx-list") +@click.option( + "--rpc-url", + required=True, + envvar="ETH_RPC_URL", + help="Ethereum RPC URL (or set ETH_RPC_URL env var)", +) +@click.option( + "--contract-address", + required=True, + type=str, + help="Contract address (0x-prefixed)", +) +@click.option( + "--etherscan-api-key", + required=False, + envvar="ETHERSCAN_API_KEY", + help="Etherscan API key (or set ETHERSCAN_API_KEY)", +) +@click.option( + "--etherscan-network", + required=False, + type=click.Choice(["mainnet", "sepolia", "holesky"], case_sensitive=False), + default="mainnet", + show_default=True, + help="Etherscan network to query", +) +@click.option( + "--start-block", + type=int, + default=None, + help="Starting block for history fetch (default: explorer/provider default)", +) +@click.option( + "--end-block", + type=int, + default=None, + help="Ending block for history fetch (default: explorer/provider default)", +) +@click.option( + "--limit", + type=int, + default=None, + help="Maximum number of txs to write to the JSON file (default: all)", +) +@click.option( + "--output", + "output_path", + required=False, + type=str, + help="Output JSON file path (default: .json)", +) +@click.option( + "--verbose", + "-v", + is_flag=True, + help="Enable verbose output", +) +def tx_list( + rpc_url: str, + contract_address: str, + etherscan_api_key: Optional[str], + etherscan_network: str, + start_block: Optional[int], + end_block: Optional[int], + limit: Optional[int], + output_path: Optional[str], + verbose: bool, +): + """ + Fetch and print the transaction hashes involving a contract. + + This does NOT replay transactions; it only retrieves the tx hash list and writes + it to a JSON file. + """ + try: + if not etherscan_api_key: + raise click.UsageError( + "tx-list requires --etherscan-api-key (or ETHERSCAN_API_KEY in .env)" + ) + + if verbose: + click.echo( + f"🔍 Fetching tx history from Etherscan ({etherscan_network})...", + err=True, + ) + + # Fetch up to `limit` transactions from Etherscan. If limit is None, + # this will return as many as the API allows (up to its internal cap). + tx_hashes = get_contract_history( + api_key=etherscan_api_key, + contract_address=contract_address, + network=etherscan_network, + start_block=start_block, + end_block=end_block, + limit=limit, + include_internal=True, + ) + + filename = output_path or f"{contract_address}.json" + payload = { + "contract_address": contract_address, + "count": len(tx_hashes), + "tx_hashes": tx_hashes, + "source": "etherscan", + "network": etherscan_network, + } + + with open(filename, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2) + + click.echo(f"Wrote {len(tx_hashes)} transactions to {filename}") + + sys.exit(0) + + except (EtherscanError, click.ClickException) as e: + click.echo(f"❌ Error: {str(e)}", err=True) + if verbose: + import traceback + + traceback.print_exc() + sys.exit(1) + except Exception as e: + click.echo(f"❌ Error: {str(e)}", err=True) + if verbose: + import traceback + + traceback.print_exc() + sys.exit(1) + + @cli.command() @click.option( "--rpc-url", diff --git a/pondereplay/etherscan.py b/pondereplay/etherscan.py index 187cefb..7437640 100644 --- a/pondereplay/etherscan.py +++ b/pondereplay/etherscan.py @@ -12,10 +12,15 @@ import requests -_NETWORK_TO_BASE_URL: Dict[str, str] = { - "mainnet": "https://api.etherscan.io", - "sepolia": "https://api-sepolia.etherscan.io", - "holesky": "https://api-holesky.etherscan.io", +# V2 base URL is shared across all chains; the target network is selected via +# the `chainid` query parameter. +_ETHERSCAN_V2_BASE_URL = "https://api.etherscan.io/v2/api" + +# Supported network names -> chain IDs for the V2 API. +_NETWORK_TO_CHAIN_ID: Dict[str, int] = { + "mainnet": 1, + "sepolia": 11155111, + "holesky": 17000, } @@ -44,10 +49,13 @@ def _dedupe_preserve_order(items: Sequence[_TxRow]) -> List[str]: def _etherscan_api_get( - base_url: str, params: Dict[str, str], timeout_s: int = 30 + chain_id: int, params: Dict[str, str], timeout_s: int = 30 ) -> dict: try: - resp = requests.get(f"{base_url}/api", params=params, timeout=timeout_s) + all_params = {**params, "chainid": str(chain_id)} + resp = requests.get( + _ETHERSCAN_V2_BASE_URL, params=all_params, timeout=timeout_s + ) resp.raise_for_status() data = resp.json() except Exception as e: # pragma: no cover (network stack specifics) @@ -73,7 +81,7 @@ def _etherscan_api_get( def _fetch_account_txs( *, - base_url: str, + chain_id: int, api_key: str, address: str, action: str, @@ -84,12 +92,18 @@ def _fetch_account_txs( rows: List[_TxRow] = [] page = 1 - offset = 10_000 # Etherscan maximum page size + # Etherscan enforces page * offset <= 10_000; use a conservative page size. + offset = 1_000 + max_window = 10_000 + max_pages = max_window // offset while True: remaining = None if limit is None else max(0, limit - len(rows)) if remaining == 0: break + if page > max_pages: + # We've reached the maximum result window that Etherscan allows. + break params = { "module": "account", @@ -105,7 +119,7 @@ def _fetch_account_txs( if end_block is not None: params["endblock"] = str(end_block) - data = _etherscan_api_get(base_url, params) + data = _etherscan_api_get(chain_id, params) result = data.get("result", []) if not result: break @@ -157,17 +171,21 @@ def get_contract_history( - (Optional) Internal transactions: account/txlistinternal (deduped by hash) """ network = str(network).strip().lower() - if network not in _NETWORK_TO_BASE_URL: + if network not in _NETWORK_TO_CHAIN_ID: raise ValueError( f"Unsupported etherscan network '{network}'. " - f"Supported: {', '.join(sorted(_NETWORK_TO_BASE_URL.keys()))}" + f"Supported: {', '.join(sorted(_NETWORK_TO_CHAIN_ID.keys()))}" ) - base_url = _NETWORK_TO_BASE_URL[network] + chain_id = _NETWORK_TO_CHAIN_ID[network] address = contract_address.strip() + # Inject chain ID into all subsequent calls + # (mutating global params dicts would be error-prone, so we pass it via a + # closure parameter). + txs = _fetch_account_txs( - base_url=base_url, + chain_id=chain_id, api_key=api_key, address=address, action="txlist", @@ -179,7 +197,7 @@ def get_contract_history( internal: List[_TxRow] = [] if include_internal: internal = _fetch_account_txs( - base_url=base_url, + chain_id=chain_id, api_key=api_key, address=address, action="txlistinternal", diff --git a/pyproject.toml b/pyproject.toml index d541a1f..a7e6c87 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,5 @@ dev = [ "pytest-cov>=4.0.0", "pytest-mock>=3.10.0", "black>=22.0.0", - "flake8>=4.0.0", "mypy>=0.900", ] diff --git a/tests/test_cli.py b/tests/test_cli.py index c460daf..eb39d74 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -392,3 +392,54 @@ def test_replay_history_requires_exactly_one_source(self, tmp_path): ) assert res.exit_code != 0 assert "exactly one history source" in res.output.lower() + + +class TestCLITxList: + @patch("pondereplay.cli.get_contract_history") + def test_tx_list_etherscan_json(self, mock_get_history): + runner = CliRunner() + mock_get_history.return_value = ["0x" + "1" * 64, "0x" + "2" * 64] + + out_file = "txs.json" + + res = runner.invoke( + cli, + [ + "tx-list", + "--rpc-url", + "http://localhost:8545", + "--contract-address", + "0xabcd", + "--etherscan-api-key", + "k", + "--etherscan-network", + "mainnet", + "--output", + out_file, + ], + ) + + assert res.exit_code == 0 + path = Path(out_file) + assert path.exists() + data = json.loads(path.read_text()) + assert data["count"] == 2 + assert data["tx_hashes"] == ["0x" + "1" * 64, "0x" + "2" * 64] + + def test_tx_list_etherscan_requires_key(self): + runner = CliRunner() + + res = runner.invoke( + cli, + [ + "tx-list", + "--rpc-url", + "http://localhost:8545", + "--contract-address", + "0xabcd", + ], + env={"ETHERSCAN_API_KEY": ""}, + ) + + assert res.exit_code != 0 + assert "tx-list requires --etherscan-api-key" in res.output.lower()