[wall3t-finance]-telegrambot

access-token.env

@@ -0,0 +1,2 @@
+TELEGRAM-BOT-ACCESS-TOKEN=6789798525:AAFcFx4PeWPg2mp1YPTzaAXTlea_ZspepLY
+ETHERSCAN-API-KEY=41TK86IIW22BFZNIHY3TDYJRSXRC56YPFU

readme.md

@@ -0,0 +1,25 @@
+# 🚀 Wall3t.finance: Managing Your Crypto Legacy Securely
+
+## Idea
+1. **Problem Statement**:
+   Managing crypto assets' legacy can be complex and risky due to crypto wallets' limitations and concerns. Users often find themselves in a dead end at the time of securing their beneficiaries in the eventuality of their physical disappearance (AKA death). Wall3t.finance aims to address these challenges by providing an effective, secure, and user-friendly solution.
+
+2. **Solution and Technology**:
+   Wall3t.finance utilizes smart contracts and account abstraction technology to secure and help manage users' crypto assets succession process. It is built on the XDC Network, offering top security and transparency.
+
+## Introducing Wall3tNotificatorBot 🤖
+This Telegram bot helps you keep track of and notifies you of the movements of your Wall3t! 🔒💰
+
+### Features:
+- 📈 Real-time Tracking: Stay up-to-date with the movements of your crypto assets.
+- 🔐 Secure and Transparent: Wall3t.finance leverages smart contracts and the XDC Network for top-notch security and transparency.
+- 💼 Manage Your Legacy: Ensure your beneficiaries have a clear path to your crypto assets in case of unforeseen circumstances.
+- 📣 User-Friendly: Wall3tBot is designed with the user in mind, making it easy for anyone to use.
+
+## Getting Started
+To start using Wall3tBot, simply [add the bot to your Telegram](t.me/Wall3tNotificatorBot) and follow the setup instructions.
+
+## Feedback and Contribution
+We welcome your feedback and contributions to make Wall3t.finance even better! Feel free to open issues or pull requests on our [GitHub repository](https://github.com/wall3t-finance/Wall3tNotificatorBot.git).
+
+👨‍💻 Happy tracking and securing your crypto legacy with Wall3t.finance!

requirements.txt

@@ -0,0 +1,46 @@
+aiohttp==3.8.6
+aiosignal==1.3.1
+anyio==4.0.0
+async-timeout==4.0.3
+attrs==23.1.0
+bitarray==2.8.2
+certifi==2023.7.22
+charset-normalizer==3.3.1
+cytoolz==0.12.2
+eth-abi==4.2.1
+eth-account==0.9.0
+eth-hash==0.5.2
+eth-keyfile==0.6.1
+eth-keys==0.4.0
+eth-rlp==0.3.0
+eth-typing==3.5.1
+eth-utils==2.3.0
+exceptiongroup==1.1.3
+frozenlist==1.4.0
+h11==0.14.0
+hexbytes==0.3.1
+httpcore==0.18.0
+httpx==0.25.0
+idna==3.4
+jsonschema==4.19.1
+jsonschema-specifications==2023.7.1
+lru-dict==1.2.0
+multidict==6.0.4
+parsimonious==0.9.0
+protobuf==4.24.4
+pycryptodome==3.19.0
+pyTelegramBotAPI==4.14.0
+python-dotenv==1.0.0
+pyunormalize==15.0.0
+referencing==0.30.2
+regex==2023.10.3
+requests==2.31.0
+rlp==3.0.0
+rpds-py==0.10.6
+sniffio==1.3.0
+toolz==0.12.0
+typing_extensions==4.8.0
+urllib3==2.0.7
+web3==6.11.1
+websockets==12.0
+yarl==1.9.2

telegram-bot.py

@@ -0,0 +1,167 @@
+import telebot, os, threading, time
+from telebot import types
+from dotenv import load_dotenv
+import utils
+
+load_dotenv("access-token.env")
+BOT_TOKEN = os.getenv("TELEGRAM-BOT-ACCESS-TOKEN")
+bot = telebot.TeleBot(BOT_TOKEN)
+global is_running
+is_running = False
+
+# Dictionary to store user data, including their contract address
+user_data = {}
+
+@bot.message_handler(commands=['start', 'hello'])
+def send_welcome(message):
+    bot.reply_to(message, "Howdy, how are you doing?")
+    bot.send_message(message.chat.id, "Please enter a public contract address:")
+
+@bot.message_handler(commands=['stop'])
+def stop_notifications(message):
+    user_id = message.from_user.id
+
+    if user_id not in user_data or not user_data[user_id]["contract_addresses"]:
+        bot.send_message(message.chat.id, "You are not tracking any contract addresses.")
+        return
+
+    markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
+    keyboard_items = [types.KeyboardButton(address) for address in user_data[user_id]["contract_addresses"]]
+    keyboard_items.append(types.KeyboardButton("Delete All"))
+    markup.add(*keyboard_items)
+
+    bot.send_message(message.chat.id, "Select the contract address you want to stop tracking:", reply_markup=markup)
+    bot.register_next_step_handler(message, delete_contract_address)
+
+def delete_contract_address(message):
+    user_id = message.from_user.id
+    selected_option = message.text
+
+    if selected_option == "Delete All":
+        user_data[user_id]["contract_addresses"] = []
+        user_data[user_id]["last_hashes"] = {}
+        bot.send_message(message.chat.id, "All contract addresses have been deleted.")
+    elif selected_option in user_data[user_id]["contract_addresses"]:
+        user_data[user_id]["contract_addresses"].remove(selected_option)
+        user_data[user_id]["last_hashes"].pop(selected_option, None)
+        bot.send_message(message.chat.id, f"Contract address {selected_option} has been deleted.")
+    else:
+        bot.send_message(message.chat.id, "Invalid choice. Please select a valid contract address or 'Delete All'.")
+
+    bot.send_message(message.chat.id, "Keyboard removed.", reply_markup=types.ReplyKeyboardRemove())
+
+@bot.message_handler(func=lambda message: True)
+def save_contract_address(message):
+    user_id = message.from_user.id
+    contract_address = message.text
+    # TODO: Validate contract address
+    if user_id not in user_data:
+        print("New user")
+        user_data[user_id] = {"contract_addresses": [contract_address], "message_chat_id": message.chat.id, "last_hashes": {}}
+    else:
+        print("Existing user")
+        user_data[user_id]["contract_addresses"].append(contract_address)
+
+    bot.send_message(message.chat.id, f"Contract address {user_data[user_id]['contract_addresses']} saved")
+    print(user_data)
+    if not is_running:
+        thread = threading.Thread(target=handle_user_choice)
+        thread.daemon = True
+        thread.start()
+
+def handle_user_choice():
+    is_running = True
+
+    while len(user_data) > 0:
+        print("\nChecking for new transactions...")
+        for user_id, user_info in user_data.items():
+            # try:
+                print(f"Checking for user with id {user_id}...")
+                for contract_address in user_info["contract_addresses"]:
+                    last_hash = user_info["last_hashes"].get(contract_address, {})
+                    print(f"last_hash: {last_hash}")
+                    message_chat_id = user_info["message_chat_id"]
+                    response = utils.get_last_movement(contract_address, 5)
+                    if response == "Error":
+                        print("Error fetching information about the contract. Please try again.")
+                        bot.send_message(message_chat_id, "Error fetching information about the contract. Please try again.")
+                    elif last_hash == {}:
+                        print("First time checking for transactions, saving last hash, sending last transaction")
+                        user_info["last_hashes"][contract_address] = response[0]["hash"]
+                        print(user_info["last_hashes"])
+                        json_data = response[0]
+                        formatted_json = f"\t🔙LAST TRANSACTION:\n" \
+                                            f"🆔 *ID*: {json_data['_id']}\n" \
+                                            f"📈 *Type*: {json_data['type']}\n" \
+                                            f"⛽ *Base Gas Price*: {json_data['baseGasPrice']} Wei\n" \
+                                            f"✅ *Status*: {'✅' if json_data['status'] else '❌'}\n" \
+                                            f"🔄 *Transaction Index*: {json_data['i_tx']}\n" \
+                                            f"🔗 *Block Hash*: {json_data['blockHash']}\n" \
+                                            f"🧱 *Block Number*: {json_data['blockNumber']}\n" \
+                                            f"👤 *From*: {json_data['from']}\n" \
+                                            f"⛽ *Gas*: {json_data['gas']} Wei\n" \
+                                            f"💹 *Gas Price*: {json_data['gasPrice']} Wei\n" \
+                                            f"📄 *Hash*: {json_data['hash']}\n" \
+                                            f"🔀 *Input*: {json_data['input'][:10]}...\n" \
+                                            f"🔢 *Nonce*: {json_data['nonce']}\n" \
+                                            f"👛 *To*: {json_data['to']}\n" \
+                                            f"💰 *Value*: {json_data['value']} Wei\n" \
+                                            f"🕒 *Created At*: {json_data['createdAt']}\n" \
+                                            f"🕒 *Updated At*: {json_data['updatedAt']}\n" \
+                                            f"📃 *Contract Address*: {json_data['contractAddress']}\n" \
+                                            f"🏆 *Cumulative Gas Used*: {json_data['cumulativeGasUsed']} Wei\n" \
+                                            f"⛽ *Gas Used*: {json_data['gasUsed']} Wei\n" \
+                                            f"📅 *Timestamp*: {json_data['timestamp']}\n"
+                        bot.send_message(message_chat_id, formatted_json, parse_mode="Markdown")
+                    elif response[0]["hash"] != last_hash:
+                        print("New transaction found!")
+                        bot.send_message(message_chat_id, f"🔔 New transaction found for contract address {contract_address}!")
+                        n = 0
+                        while n < len(response) and response[n]["hash"] != last_hash:
+                            json_data = response[n]
+                            formatted_json = f"\t🆕*NEW TRANSACTION*🆕\n" \
+                                            f"🆔 *ID*: {json_data['_id']}\n" \
+                                            f"📈 *Type*: {json_data['type']}\n" \
+                                            f"⛽ *Base Gas Price*: {json_data['baseGasPrice']} Wei\n" \
+                                            f"✅ *Status*: {'✅' if json_data['status'] else '❌'}\n" \
+                                            f"🔄 *Transaction Index*: {json_data['i_tx']}\n" \
+                                            f"🔗 *Block Hash*: {json_data['blockHash']}\n" \
+                                            f"🧱 *Block Number*: {json_data['blockNumber']}\n" \
+                                            f"👤 *From*: {json_data['from']}\n" \
+                                            f"⛽ *Gas*: {json_data['gas']} Wei\n" \
+                                            f"💹 *Gas Price*: {json_data['gasPrice']} Wei\n" \
+                                            f"📄 *Hash*: {json_data['hash']}\n" \
+                                            f"🔀 *Input*: {json_data['input'][:10]}...\n" \
+                                            f"🔢 *Nonce*: {json_data['nonce']}\n" \
+                                            f"👛 *To*: {json_data['to']}\n" \
+                                            f"💰 *Value*: {json_data['value']} Wei\n" \
+                                            f"🕒 *Created At*: {json_data['createdAt']}\n" \
+                                            f"🕒 *Updated At*: {json_data['updatedAt']}\n" \
+                                            f"📃 *Contract Address*: {json_data['contractAddress']}\n" \
+                                            f"🏆 *Cumulative Gas Used*: {json_data['cumulativeGasUsed']} Wei\n" \
+                                            f"⛽ *Gas Used*: {json_data['gasUsed']} Wei\n" \
+                                            f"📅 *Timestamp*: {json_data['timestamp']}\n"
+                            print("🆕*NEW TRANSACTION*🆕")
+                            bot.send_message(message_chat_id, formatted_json, parse_mode="Markdown")
+                            n += 1
+                        user_info["last_hashes"][contract_address] = response[0]["hash"]
+                    else:
+                        print(f"No new transactions found for {contract_address}")
+            # except Exception as e:
+            #     print(f"Error:\n{e}")
+            #     print(f"Error trace: {traceback.format_exc()}")
+
+        print("Sleeping for 5 seconds")
+        time.sleep(5)
+
+    print("No users left")
+    is_running = False
+
+
+@bot.message_handler(func=lambda msg: True)
+def echo_all(message):
+    bot.reply_to(message, "This is not a valid command. Please try again.")
+
+
+if __name__ == "__main__":
+    bot.infinity_polling()

utils.py

@@ -0,0 +1,62 @@
+import requests, os
+from web3 import Web3
+from dotenv import load_dotenv
+
+def get_daily_horoscope(sign: str, day: str) -> dict:
+    """Get daily horoscope for a zodiac sign.
+
+    Keyword arguments:
+    sign:str - Zodiac sign
+    day:str - Date in format (YYYY-MM-DD) OR TODAY OR TOMORROW OR YESTERDAY
+    Return:dict - JSON data
+    """
+    url = "https://horoscope-app-api.vercel.app/api/v1/get-horoscope/daily"
+    params = {"sign": sign, "day": day}
+    response = requests.get(url, params)
+
+    return response.json()
+
+def get_last_movement(contract_address: str, num_transactions = 10) -> dict:
+    """Get last movement of a contract.
+
+    Keyword arguments:
+    contract_address:str - Contract address
+    num_transactions:int - Number of transactions to fetch (default: 10)
+    Return:dict - JSON data
+    """
+    headers = {
+    'authority': 'apothem.xinfinscan.com',
+    'accept': 'application/json, text/plain, */*',
+    'access-control-allow-origin': '*',
+    }
+
+    params = {
+        'page': '1',
+        'limit': '20',
+        'tx_type': 'all',
+    }
+
+    url = f"https://apothem.xinfinscan.com/api/txs/listByAccount/{contract_address}"
+
+    response = requests.get(
+        url=url,
+        params=params,
+        headers=headers,
+    )
+
+    if response.status_code == 200:
+        data = response.json()
+        if len(data["items"]) > 0:
+            return data["items"]
+        else:
+            print("No transaction data found.")
+            return data
+    else:
+        print("Failed to fetch data. Check your API key and request parameters.")
+
+    return "Error"
+
+
+if __name__ == "__main__":
+    contract = input("Enter contract address: ")
+    print(get_last_movement(contract))