Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
ercxxx_distribution
import json
import os
from collections import defaultdict
from json import JSONDecodeError
from pprint import pprint
from typing import Dict, Union, List
import requests
from bs4 import BeautifulSoup
def get_top_tokens(num_pages: int = 10) -> List[Dict[str, Union[int, str]]]:
"""
Gets the tokens listed on the first `num_pages` pages of "https://etherscan.io/tokens?p=page_number"
:param num_pages: the number of pages of tokens
:return: list of objects in the following format:
tokens = [
{"name": <token name>, "rank": <rank on etherscan>, "address": <token address>}
]
"""
url = "https://etherscan.io/tokens?p={}"
tokens = []
for i in range(num_pages):
print(f"Getting tokens {(i*50)} - {(i*50) + 50}")
response = requests.get(url.format(i+1))
if response.ok:
soup = BeautifulSoup(response.text, 'html.parser')
rows = soup.find('tbody')
if rows:
for j, tr in enumerate(rows.find_all('tr')):
# all token info is in the a tag
token = tr.find('a')
tokens.append({'name': token.text, 'rank': i*50 + j + 1, 'address': token['href'].split('/')[-1]})
else:
print("No data")
print(f"Got {len(tokens)} tokens")
return tokens
def get_abi(address: str) -> str:
"""
Gets the abi for the `address` using etherscan.io api.
Assumes environment variable `API_KEY` exists.
:param address: the contract address for abi
:return: abi object json string or empty string
"""
url = f"https://api.etherscan.io/api?module=contract&action=getabi&address={address}&apikey=" + os.environ["API_KEY"]
print(f'Getting {url}')
r = requests.get(url)
if r.ok:
body = r.json()
if body['message'] == "OK":
return body['result']
else:
print(body['message'])
return ""
else:
print(f'Bad request: {r}')
return ""
def download_abis(download_dir, num_top_tokens=500):
# download the top tokens
top_tokens = get_top_tokens(num_pages=num_top_tokens//50)
# save the top tokens locally
with open(f'top{len(top_tokens)}_tokens.json', 'w', encoding='utf-8') as f:
json.dump(top_tokens, f)
try:
os.mkdir(download_dir)
except FileExistsError:
print("Not creating download folder")
for token_info in top_tokens:
token_address = token_info['address']
file_name = os.path.join(download_dir, f"{token_info['name'].replace(' ', '_')}__{token_info['rank']}")
if os.path.exists(file_name):
print("abi already downloaded")
continue
print(f"Getting {token_info['name']}@{token_address}")
downloaded_abi = get_abi(token_address)
if downloaded_abi:
with open(file_name, 'w', encoding='utf-8') as f:
f.write(downloaded_abi)
def get_distributions(abi_files: List):
func_counter = defaultdict(int)
erc_distributions = defaultdict(list)
for abi_file in abi_files:
with open(os.path.join(abi_folder, abi_file), 'r', encoding='utf-8') as f:
try:
abi = json.load(f)
except JSONDecodeError as e:
print(f'Failed reading abi: {abi_file}', e)
for func in abi:
if 'name' in func and func['type'] != 'event':
name = func['name'].lower()
func_counter[name] += 1
if name == 'transfer':
# erc223 has function transfer(address _to, uint _value, bytes _data) returns (bool success)
if len(func['inputs']) > 2:
erc_distributions['erc223 - transferwithbytes'].append(abi_file)
elif name == 'transferandcall':
# erc677 has function transferAndCall(address receiver, uint amount, bytes data) returns (bool success)
erc_distributions['erc677 - transferandcall'].append(abi_file)
elif name == 'approveandcall':
# non standard but a bunch implement function approveAndCall(address _recipient, uint256 _value, bytes _extraData)
erc_distributions['non-standard - approveandcall'].append(abi_file)
elif name == 'send':
# erc223 has function transfer(address _to, uint _value, bytes _data) returns (bool success)
if len(func['inputs']) > 2:
erc_distributions['erc777 - send'].append(abi_file)
if is_erc20(abi):
erc_distributions['erc20compatible'].append(abi_file)
with open('function_list.json', 'w', encoding='utf-8') as f:
json.dump(func_counter, f)
pprint(erc_distributions)
return erc_distributions
def is_erc20(abi: List) -> bool:
has_approve = False
has_transferfrom = False
for func in abi:
if 'name' in func and func['type'] != 'event':
name = func['name'].lower()
if name == 'transferfrom':
has_transferfrom = True
if name == "approve":
has_approve = True
return has_approve and has_transferfrom
if __name__ == "__main__":
# folder to download the abis for cache
abi_folder = "contract_abis"
# comment line if using cached abis
# download_abis(abi_folder, num_top_tokens=1000)
contract_abis = os.listdir(abi_folder)
distributions = get_distributions(contract_abis)
for erc, tokens in distributions.items():
print(f'{erc}: {len(tokens)}')
approveandcall_in_erc20 = [o for o in distributions["non-standard - approveandcall"] if o in distributions["erc20compatible"]]
print(f'{len(approveandcall_in_erc20)} approveandcall tokens are erc20compatible')
print(f'Scanned {len(contract_abis)} contracts')
@jtakalai
Copy link

jtakalai commented Mar 2, 2021

Still working like a charm, thank you for the script!

For anyone else trying, here's my command-line session for running it:

export API_KEY=[see your api-key at https://etherscan.io/myapikey]
python3 -m venv erc_distribution
cd erc_distribution/
mkdir contract_abis
wget https://gist.githubusercontent.com/MosheStauber/5cf83310f5b33acd8ac352f9d3f37843/raw/e8186c6db52d6656e459bde49da8f62338730c11/ercxxx_distribution.py
python3 -m pip install beautifulsoup4
python3 -m pip install cloudscraper

you also need to change line 8 import requests to:

import cloudscraper
requests = cloudscraper.create_scraper()

This gets around Cloudflare block on requests. Also uncomment line 152, then just

python ercxxx_distribution.py

and cloudscraper goes brrrrrrr

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment