Last active
May 18, 2024 19:43
-
-
Save juanpabloaj/c67eea01543b422af22f5fc2833c3a3a to your computer and use it in GitHub Desktop.
track ethereum token
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from datetime import datetime | |
| import json | |
| import logging | |
| import os | |
| import requests | |
| import sqlite3 | |
| def init_db(db_path): | |
| conn = sqlite3.connect(db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS transactions ( | |
| hash TEXT PRIMARY KEY, | |
| from_address TEXT, | |
| to_address TEXT, | |
| contract_address TEXT, | |
| timestamp TEXT | |
| ) | |
| """ | |
| ) | |
| conn.commit() | |
| conn.close() | |
| def save_transaction(db_path, tx): | |
| conn = sqlite3.connect(db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| """ | |
| INSERT OR IGNORE INTO transactions (hash, from_address, to_address, contract_address, timestamp) | |
| VALUES (?, ?, ?, ?, ?) | |
| """, | |
| ( | |
| tx["hash"], | |
| tx["from"], | |
| tx["to"], | |
| tx["contractAddress"], | |
| tx["timeStamp"], | |
| ), | |
| ) | |
| conn.commit() | |
| conn.close() | |
| def get_latest_transaction(db_path): | |
| conn = sqlite3.connect(db_path) | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT MAX(timestamp) FROM transactions") | |
| result = cursor.fetchone() | |
| conn.close() | |
| return result[0] if result[0] else None | |
| def track_eth_wallet(api_key, token_address): | |
| url = ( | |
| "https://api.etherscan.io/api?module=account&action=tokennfttx" | |
| f"&contractaddress={token_address}" | |
| "&address=0x0000000000000000000000000000000000000000&page=1" | |
| "&offset=10&startblock=0&endblock=27025780&sort=desc" | |
| f"&apikey={api_key}" | |
| ) | |
| response = requests.get(url) | |
| data = response.json() | |
| return data | |
| def main(): | |
| api_key = os.environ.get("ETHERSCAN_API_KEY") | |
| wallet_address = os.environ.get("TRACKED_WALLET_ADDRESS") | |
| logging.debug(f"API KEY: {api_key:.5}...") | |
| logging.debug(f"Tracking wallet: {wallet_address}") | |
| data = track_eth_wallet(api_key, wallet_address) | |
| if data["status"] != "1": | |
| logging.error("Error in response") | |
| return | |
| result = data["result"] | |
| if not result: | |
| logging.error("No transactions found") | |
| return | |
| script_dir = os.path.dirname(__file__) | |
| db_path = os.path.join(script_dir, "eth_transactions.db") | |
| init_db(db_path) | |
| latest_tx_timestamp = get_latest_transaction(db_path) | |
| for tx in result: | |
| tx_timestamp = tx["timeStamp"] | |
| if not latest_tx_timestamp or tx_timestamp > latest_tx_timestamp: | |
| date = datetime.fromtimestamp(int(tx_timestamp)) | |
| logging.info(f"new transaction! date:{date} hash:{tx['hash']}") | |
| save_transaction(db_path, tx) | |
| if __name__ == "__main__": | |
| LOGLEVEL = os.environ.get("LOGLEVEL", "INFO").upper() | |
| logging.basicConfig(format="%(asctime)s %(message)s", level=LOGLEVEL) | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment