track ethereum token

.gitignore

.env
*.db

track_eth_token.py

from datetime import datetime
import json
import logging
import os
import requests
import sqlite3


def send_telegram_message(message):
    telegram_token = os.environ["TELEGRAM_TOKEN"]
    telegram_channel_id = os.environ["TELEGRAM_CHANNEL_ID"]

    if not telegram_token or not telegram_channel_id:
        logging.error("Telegram token or channel id not set")
        return

    url = f"https://api.telegram.org/bot{telegram_token}/sendMessage"
    data = {
        "chat_id": telegram_channel_id,
        "text": message,
    }
    response = requests.post(url, data=data)
    return response.json()


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

    data = track_eth_wallet(api_key, wallet_address)

    if data["status"] != "1":
        logging.error(f"Error in response {data}")
        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))

            message = f"new token transaction! date: {date} hash: {tx['hash']}"

            logging.info(message)
            send_telegram_message(message)

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