implement tx-list

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: <contract-address>.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",

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",

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",
 ]

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()