diff --git a/bot/__init__.py b/bot/__init__.py index 7030b2c..8ae96f6 100644 --- a/bot/__init__.py +++ b/bot/__init__.py @@ -1,2 +1,3 @@ from .account import Account from .utils import launch_docker_container +from . import exceptions diff --git a/bot/account.py b/bot/account.py index 8837703..807a811 100644 --- a/bot/account.py +++ b/bot/account.py @@ -1,6 +1,7 @@ from typing import Optional, Union, Generator, Any import requests, datetime, re, logging, os, json, ast, shutil, eth_abi, shutil import pandas as pd +from . import exceptions from PIL import Image from PIL.JpegImagePlugin import JpegImageFile from . import constants @@ -114,7 +115,7 @@ class Account: - `coingecko_api_key` - https://www.coingecko.com/ API key to allow Status Backend to use a wallet """ if not key_uid and not display_name: - raise ValueError("Please provide either a Key Unique Identifier (key_uid) or a Display Name (display_name)...") + raise exceptions.InvalidContactError() available_accounts = self.available_accounts # Login combination: display_name + password @@ -130,7 +131,7 @@ class Account: available_key_uids = [current["key_uid"] for current in available_accounts] if key_uid not in available_key_uids: info = "\n".join([f"{current['key_uid']} - {current['display_name']}" for current in self.available_accounts]) - raise ValueError(f"Given Key Unique Identifier is invalid...\nAvailable Key Unique Identifiers:\n{info}") + raise exceptions.InvalidContactError(f"Given Key Unique Identifier is invalid...\nAvailable Key Unique Identifiers:\n{info}") is_new_account = isinstance(key_uid, type(None)) is_recovery = not isinstance(mnemonic, type(None)) and not key_uid @@ -187,7 +188,7 @@ class Account: response = requests.post(url, json=params) signal_event = self.__signal.get("node.login") if signal_event["is_error"]: - raise Exception(f"There was an error with Status Backend...\n{signal_event['error_message']}") + raise exceptions.BackendError(f"There was an error with Status Backend...\n{signal_event['error_message']}") self.logger.info("Successfully logged in!") event: dict = signal_event["event"]["settings"] @@ -259,7 +260,7 @@ class Account: Can also be used to verify if the user has logged in. """ if not self.__info: - raise Exception("Make sure you are logged in to your Status account with login() first...") + raise exceptions.NotLoggedInError() return self.__info @property @@ -295,7 +296,7 @@ class Account: # Limit based from Status App CHARACTERS = 240 if len(bio) > CHARACTERS: - raise ValueError(f"Bio cannot be longer than {CHARACTERS} characters...") + raise exceptions.InvalidDisplayNameError(f"Bio cannot be longer than {CHARACTERS} characters...") self.__call_rpc("messaging", "setBio", [bio]) # It seems that if a valid bio is given, it will be instantly updated @@ -336,11 +337,11 @@ class Account: return if not os.path.exists(file_path): - raise Exception(f"File path {file_path} does not exist") + raise exceptions.ProfilePictureError(f"File path {file_path} does not exist") suffix = (".jpg", ".png", ".jpeg") if not file_path.endswith(suffix): - raise Exception(f"Image must be one of the following extensions: {suffix}") + raise exceptions.ProfilePictureError(f"Image must be one of the following extensions: {suffix}") file_name = os.path.basename(file_path) @@ -619,7 +620,7 @@ class Account: ccy = key.upper() if ccy not in self.__get_fiat_ccy(): - raise Exception(f"{ccy} is an invalid fiat (ISO 4217) currency code...") + raise exceptions.InvalidCurrencyError(f"{ccy} is an invalid fiat (ISO 4217) currency code...") balance = self.balance tokens = (balance["chain_id"].astype(str) + "-" + balance["address"]).to_list() @@ -746,7 +747,7 @@ class Account: break if not display_name: - raise ValueError(f"Cannot add contact {public_key}...\nPlease make sure you add display_name for contacts that you are sending friend requests to and have never interacted with before!") + raise exceptions.InvalidContactError(f"Cannot add contact {public_key}...\nPlease make sure you add display_name for contacts that you are sending friend requests to and have never interacted with before!") params = [{"id": public_key, "nickname": "", "displayName": display_name, "ensName": ""}] self.__call_rpc("messaging", "addContact", params) @@ -813,7 +814,7 @@ class Account: file_path = result.get("filePath") if not file_path or (isinstance(file_path, str) and len(file_path) == 0): - raise Exception(f"There was an error with creating a backup for {self.info['display_name']}") + raise exceptions.BackupError(f"There was an error with creating a backup for {self.info['display_name']}") file_name = os.path.basename(file_path) sdk_file_path = os.path.join(self.__backup_sdk_folder, file_name) @@ -976,7 +977,7 @@ class Account: ccy = ccy.upper() available_ccy = self.__get_fiat_ccy() if ccy not in available_ccy: - raise Exception(f"Given currency {ccy} is invalid...\nAvailable ISO 4217 currencies: {available_ccy}") + raise exceptions.InvalidCurrencyError(f"Given currency {ccy} is invalid...\nAvailable ISO 4217 currencies: {available_ccy}") tokens = self.__get_valid_tokens(chain_ids, token_addresses) market_info = pd.DataFrame([ @@ -1033,18 +1034,18 @@ class Account: tokens = self.get_tokens()[["chain_id", "address", "symbol", "decimals"]].drop_duplicates().reset_index(drop=True) query = (tokens["address" if is_address else "symbol"] == symbol) & (tokens["chain_id"] == chain_id) if query.sum() == 0: - raise Exception(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} does not exist...") + raise exceptions.InvalidTokenError(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} does not exist...") token_info = tokens.loc[query].to_dict("records")[0] balance = self.balance query = (balance["address"] == token_info["address"]) & (balance["chain_id"] == chain_id) if query.sum() == 0: - raise Exception(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} was not found in your wallet ({self.info['wallet_address']})...") + raise exceptions.InvalidTokenError(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} was not found in your wallet ({self.info['wallet_address']})...") wallet_amount = balance.loc[query].reset_index(drop=True)["amount"].iloc[0] if amount > wallet_amount: - raise Exception(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} has {wallet_amount} but you are trying to send {amount}...") + raise exceptions.InvalidTokenError(f"Given {'address' if is_address else 'symbol'} {symbol} on chain ID {chain_id} has {wallet_amount} but you are trying to send {amount}...") raw_amount = int(amount * (10**token_info["decimals"])) @@ -1083,7 +1084,7 @@ class Account: - Wallet's transactions """ if not self.__is_wallet_set: - raise Exception(f"Cannot use this method without setting an `alchemy_token` and `coingecko_api_key` when calling `login`.") + raise exceptions.WalletNotConfiguredError() if not refresh and isinstance(self.__transactions, pd.DataFrame): return self.__transactions.copy() @@ -1196,7 +1197,7 @@ class Account: file_name = self.info["compressed_key"][-6:] + "_user_data.bkp" file_path = os.path.join(folder, self.info["compressed_key"][-6:] + "_user_data.bkp") if not os.path.exists(file_path): - raise Exception(f"Backup file was not found in {folder}...") + raise exceptions.BackupError(f"Backup file was not found in {folder}...") sdk_file_path = os.path.join(self.__backup_sdk_folder, file_name) if sdk_file_path != file_path: @@ -1235,10 +1236,10 @@ class Account: self.info name = self.__prefix_mapping.get(prefix) if not name: - raise ValueError(f"Name {name} does not exist... Available options: {list(self.__prefix_mapping.keys())}") + raise exceptions.BackendError(f"Name {name} does not exist... Available options: {list(self.__prefix_mapping.keys())}") if name == "wallet" and not self.__is_wallet_set: - raise Exception(f"Cannot use this method without setting an `alchemy_token` and `coingecko_api_key` when calling `login`.") + raise exceptions.WalletNotConfiguredError() data = { 'jsonrpc': '2.0', @@ -1323,15 +1324,15 @@ class Account: - `True` if the name was successfully changed. A """ if name != name.strip(): - raise ValueError("Display name cannot start or end with a space.") + raise exceptions.InvalidDisplayNameError("Display name cannot start or end with a space.") if len(name) < 5: - raise ValueError("Display name must be at least 5 characters long.") + raise exceptions.InvalidDisplayNameError("Display name must be at least 5 characters long.") if len(name) > 24: - raise ValueError("Display name cannot be more than 24 characters long.") + raise exceptions.InvalidDisplayNameError("Display name cannot be more than 24 characters long.") if not re.fullmatch(r"[A-Za-z0-9 _-]+", name): - raise ValueError("Display name can contain only A-Z, 0-9, hyphens (-), underscores (_) and spaces.") + raise exceptions.InvalidDisplayNameError("Display name can contain only A-Z, 0-9, hyphens (-), underscores (_) and spaces.") return True diff --git a/bot/exceptions.py b/bot/exceptions.py new file mode 100644 index 0000000..ca9c0cb --- /dev/null +++ b/bot/exceptions.py @@ -0,0 +1,35 @@ +class BackendError(Exception): + pass + +class NotLoggedInError(Exception): + def __init__(self): + super().__init__("Make sure you are logged in to your Status account with login() first...") + +class WalletNotConfiguredError(Exception): + def __init__(self): + super().__init__("Cannot use this method without setting an `alchemy_token` and `coingecko_api_key` when calling `login`.") + +class InvalidDisplayNameError(ValueError): + pass + +class InvalidContactError(ValueError): + def __init__(self, msg=None): + super().__init__(msg or "Please provide either a Key Unique Identifier (key_uid) or a Display Name (display_name)...") + +class InvalidCurrencyError(Exception): + pass + +class InvalidTokenError(Exception): + pass + +class BackupError(Exception): + pass + +class ProfilePictureError(Exception): + pass + +class DockerError(Exception): + pass + +class SignalError(Exception): + pass diff --git a/bot/signal.py b/bot/signal.py index 46b070b..6881123 100644 --- a/bot/signal.py +++ b/bot/signal.py @@ -1,5 +1,6 @@ from typing import Optional import datetime, websocket, json, copy, queue, threading +from . import exceptions class Signal: """ @@ -89,7 +90,7 @@ class Signal: ws.run_forever() if self.__error_message: - raise Exception(self.__error_message) + raise exceptions.SignalError(self.__error_message) return copy.deepcopy(self.__data) @@ -143,7 +144,7 @@ class Signal: except KeyboardInterrupt: break if self.__error_message: - raise Exception(self.__error_message) + raise exceptions.SignalError(self.__error_message) yield data diff --git a/bot/utils.py b/bot/utils.py index 27b7c9b..7efbd9e 100644 --- a/bot/utils.py +++ b/bot/utils.py @@ -1,6 +1,7 @@ import shutil, os, subprocess, sys, time from pathlib import Path from .logger import Logger +from . import exceptions def launch_docker_container(wait_seconds: int = 5): """ @@ -13,7 +14,7 @@ def launch_docker_container(wait_seconds: int = 5): platform = sys.platform is_windows = platform == "win32" if not shutil.which("docker"): - raise Exception("Please install Docker.") + raise exceptions.DockerError("Please install Docker.") logger.info(f"Running Docker on {platform}") DOCKER_COMPOSE_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "docker-compose.yaml") @@ -26,7 +27,7 @@ def launch_docker_container(wait_seconds: int = 5): cmd = ["docker", "compose", "-f", docker_path, "up", "-d"] if is_windows: if not shutil.which("wsl"): - raise Exception("Please install wsl - https://learn.microsoft.com/en-us/windows/wsl/install.") + raise exceptions.DockerError("Please install wsl - https://learn.microsoft.com/en-us/windows/wsl/install.") cmd.insert(0, "wsl") logger.info(f"Running:\n{' '.join(cmd)}") @@ -38,7 +39,7 @@ def launch_docker_container(wait_seconds: int = 5): ) if result.returncode != 0: - raise Exception(result.stderr.strip()) + raise exceptions.DockerError(result.stderr.strip()) logger.info(f"Sleeping for {wait_seconds}s") time.sleep(wait_seconds)