diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f6148f7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,130 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyderworkspace + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# VS Code +.vscode/ + +# Ignore test.py +test.py \ No newline at end of file diff --git a/bot.py b/bot.py index 443a0f8..5d4083c 100644 --- a/bot.py +++ b/bot.py @@ -1,53 +1,76 @@ import os -from typing import Final +import logging +from typing import Final, List, Dict, Any import requests -from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup -from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, ContextTypes, ConversationHandler, filters +import telebot +from telebot.types import InlineKeyboardButton, InlineKeyboardMarkup, CallbackQuery, Message # Constants -BOT_USERNAME: Final = 'xyz' -BOT_TOKEN: Final = "your token" +BOT_TOKEN: Final = os.getenv("BOT_TOKEN") COINGECKO_API_URL: Final = "https://api.coingecko.com/api/v3" +SUPPORTED_CURRENCIES: Final = ['usd', 'eur', 'gbp', 'jpy', 'aud', 'cad', 'chf', 'cny', 'inr'] -# Conversation states -MAIN_MENU, CHOOSING_CRYPTO, CHOOSING_CURRENCY, TYPING_SEARCH = range(4) +# Logging configuration +logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO) +logger = logging.getLogger(__name__) -# Supported currencies -SUPPORTED_CURRENCIES = ['usd', 'eur', 'gbp', 'jpy', 'aud', 'cad', 'chf', 'cny', 'inr'] +# Initialize bot +bot = telebot.TeleBot(BOT_TOKEN) + +# User context to store state +user_context: Dict[int, Dict[str, str]] = {} # API Functions -def get_top_cryptos(limit=100): - response = requests.get(f"{COINGECKO_API_URL}/coins/markets", params={ - 'vs_currency': 'usd', - 'order': 'market_cap_desc', - 'per_page': limit, - 'page': 1, - 'sparkline': False - }) - if response.status_code == 200: +def get_top_cryptos(limit: int = 100) -> List[Dict[str, Any]]: + try: + response = requests.get(f"{COINGECKO_API_URL}/coins/markets", params={ + 'vs_currency': 'usd', + 'order': 'market_cap_desc', + 'per_page': limit, + 'page': 1, + 'sparkline': False + }) + response.raise_for_status() return response.json() - return [] - -def get_trending_cryptos(): - response = requests.get(f"{COINGECKO_API_URL}/search/trending") - if response.status_code == 200: - return response.json().get('coins', []) - return [] - -def get_crypto_details(crypto_id: str, currency: str = 'usd'): - params = {'ids': crypto_id, 'vs_currencies': currency, 'include_24hr_change': 'true', 'include_market_cap': 'true'} - response = requests.get(f"{COINGECKO_API_URL}/simple/price", params=params) - if response.status_code == 200: + except requests.RequestException as e: + logger.error(f"Error fetching top cryptos: {e}") + return [] + +def get_trending_cryptos() -> List[Dict[str, Any]]: + try: + response = requests.get(f"{COINGECKO_API_URL}/search/trending") + response.raise_for_status() + coins = response.json().get('coins', []) + return [{'id': coin['item']['id'], 'name': coin['item']['name'], + 'symbol': coin['item']['symbol'], 'image': coin['item']['thumb']} + for coin in coins] + except requests.RequestException as e: + logger.error(f"Error fetching trending cryptos: {e}") + return [] + +def get_crypto_details(crypto_id: str, currency: str) -> Dict[str, Any]: + try: + params = { + 'ids': crypto_id, + 'vs_currencies': currency, + 'include_24hr_change': 'true', + 'include_market_cap': 'true' + } + response = requests.get(f"{COINGECKO_API_URL}/simple/price", params=params) + response.raise_for_status() data = response.json() - return data.get(crypto_id) - return None + return data.get(crypto_id, {}) + except requests.RequestException as e: + logger.error(f"Error fetching crypto details for {crypto_id}: {e}") + return {} # Command Handlers -async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: - await show_main_menu(update, context) - return MAIN_MENU +@bot.message_handler(commands=['start']) +def start(message: Message) -> None: + show_main_menu(message) -async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +@bot.message_handler(commands=['help']) +def help_command(message: Message) -> None: help_text = ( "Welcome to the Crypto Price Bot!\n\n" "Commands:\n" @@ -55,142 +78,141 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No "/help - Show this help message\n\n" "You can check prices of top cryptocurrencies, view trending coins, or search for a specific cryptocurrency." ) - await update.message.reply_text(help_text) + bot.send_message(message.chat.id, help_text) # Menu Functions -async def show_main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - keyboard = [ - [InlineKeyboardButton("Top 100 Cryptocurrencies", callback_data='top100')], - [InlineKeyboardButton("Trending Cryptocurrencies", callback_data='trending')], - [InlineKeyboardButton("Search Cryptocurrency", callback_data='search')] - ] - reply_markup = InlineKeyboardMarkup(keyboard) +def show_main_menu(message: Message) -> None: + keyboard = InlineKeyboardMarkup() + keyboard.row_width = 1 + keyboard.add( + InlineKeyboardButton("Top 100 Cryptocurrencies", callback_data='top100'), + InlineKeyboardButton("Trending Cryptocurrencies", callback_data='trending'), + InlineKeyboardButton("Search Cryptocurrency", callback_data='search') + ) text = "Welcome to the Crypto Price Bot! What would you like to do?" - - if update.callback_query: - await update.callback_query.edit_message_text(text, reply_markup=reply_markup) - else: - await update.message.reply_text(text, reply_markup=reply_markup) + bot.send_message(message.chat.id, text, reply_markup=keyboard) -async def show_crypto_list(update: Update, context: ContextTypes.DEFAULT_TYPE, cryptos, title) -> None: - keyboard = [] +def show_crypto_list(call_or_message, cryptos: List[Dict[str, Any]], title: str) -> None: + keyboard = InlineKeyboardMarkup() for i in range(0, len(cryptos), 2): row = [] for crypto in cryptos[i:i+2]: name = crypto.get('name', 'Unknown') symbol = crypto.get('symbol', 'Unknown') crypto_id = crypto.get('id', 'unknown') - row.append(InlineKeyboardButton(f"{name} ({symbol.upper()})", callback_data=f"crypto:{crypto_id}")) - keyboard.append(row) - - keyboard.append([InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')]) - reply_markup = InlineKeyboardMarkup(keyboard) + image_url = crypto.get('image', '') + + button_text = f"{name} ({symbol.upper()})" + row.append(InlineKeyboardButton(button_text, callback_data=f"crypto:{crypto_id}")) + keyboard.add(*row) - if update.callback_query: - await update.callback_query.edit_message_text(title, reply_markup=reply_markup) + keyboard.add(InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')) + + if isinstance(call_or_message, CallbackQuery): + bot.edit_message_text(title, call_or_message.message.chat.id, call_or_message.message.message_id, reply_markup=keyboard) else: - await update.message.reply_text(title, reply_markup=reply_markup) + bot.send_message(call_or_message.chat.id, title, reply_markup=keyboard) -async def show_currency_options(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - keyboard = [ - [InlineKeyboardButton(currency.upper(), callback_data=f"currency:{currency}")] - for currency in SUPPORTED_CURRENCIES - ] - keyboard.append([InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')]) - reply_markup = InlineKeyboardMarkup(keyboard) - await update.callback_query.edit_message_text('Choose a currency:', reply_markup=reply_markup) +def show_currency_options(call: CallbackQuery) -> None: + keyboard = InlineKeyboardMarkup() + for currency in SUPPORTED_CURRENCIES: + keyboard.add(InlineKeyboardButton(currency.upper(), callback_data=f"currency:{currency}")) + keyboard.add(InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')) + bot.edit_message_text('Choose a currency:', call.message.chat.id, call.message.message_id, reply_markup=keyboard) # Callback Query Handler -async def button_click(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: - query = update.callback_query - await query.answer() - - if query.data == 'main_menu': - await show_main_menu(update, context) - return MAIN_MENU - elif query.data == 'top100': - await query.edit_message_text("Fetching top cryptocurrencies, please wait...") +@bot.callback_query_handler(func=lambda call: True) +def button_click(call: CallbackQuery) -> None: + if call.data == 'main_menu': + show_main_menu(call.message) + elif call.data == 'top100': + bot.edit_message_text("Fetching top cryptocurrencies, please wait...", call.message.chat.id, call.message.message_id) cryptos = get_top_cryptos() - await show_crypto_list(update, context, cryptos, "Top 100 Cryptocurrencies:") - return CHOOSING_CRYPTO - elif query.data == 'trending': - await query.edit_message_text("Fetching trending cryptocurrencies, please wait...") + show_crypto_list(call, cryptos, "Top 100 Cryptocurrencies:") + elif call.data == 'trending': + bot.edit_message_text("Fetching trending cryptocurrencies, please wait...", call.message.chat.id, call.message.message_id) cryptos = get_trending_cryptos() - await show_crypto_list(update, context, cryptos, "Trending Cryptocurrencies:") - return CHOOSING_CRYPTO - elif query.data == 'search': - await query.edit_message_text("Please enter the name of the cryptocurrency you want to check:") - return TYPING_SEARCH - elif query.data.startswith('crypto:'): - context.user_data['crypto'] = query.data.split(':')[1] - await show_currency_options(update, context) - return CHOOSING_CURRENCY - elif query.data.startswith('currency:'): - currency = query.data.split(':')[1] - crypto_id = context.user_data.get('crypto', 'bitcoin') - await show_crypto_details(update, context, crypto_id, currency) - return MAIN_MENU - -async def show_crypto_details(update: Update, context: ContextTypes.DEFAULT_TYPE, crypto_id: str, currency: str) -> None: - details = get_crypto_details(crypto_id, currency) - if details: - price = details.get(currency, 'N/A') - change_24h = details.get(f'{currency}_24h_change', 'N/A') - market_cap = details.get(f'{currency}_market_cap', 'N/A') - - change_symbol = '🔺' if change_24h > 0 else '🔻' if change_24h < 0 else '➖' - message = ( - f"💰 {crypto_id.capitalize()} ({currency.upper()})\n" - f"Price: {price:,.2f} {currency.upper()}\n" - f"24h Change: {change_symbol} {abs(change_24h):.2f}%\n" - f"Market Cap: {market_cap:,.0f} {currency.upper()}" - ) - else: - message = f"Sorry, I couldn't find the details for {crypto_id}." - - keyboard = [[InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')]] - reply_markup = InlineKeyboardMarkup(keyboard) - await update.callback_query.edit_message_text(message, reply_markup=reply_markup) + show_crypto_list(call, cryptos, "Trending Cryptocurrencies:") + elif call.data == 'search': + bot.edit_message_text("Please enter the name of the cryptocurrency you want to check:", call.message.chat.id, call.message.message_id) + bot.register_next_step_handler(call.message, handle_message) + elif call.data.startswith('crypto:'): + handle_crypto_selection(call) + elif call.data.startswith('currency:'): + handle_currency_selection(call) + +def handle_crypto_selection(call: CallbackQuery) -> None: + crypto_id = call.data.split(':')[1] + user_context[call.from_user.id] = {'crypto_id': crypto_id} + bot.answer_callback_query(call.id) + show_currency_options(call) + +def handle_currency_selection(call: CallbackQuery) -> None: + currency = call.data.split(':')[1] + crypto_id = user_context[call.from_user.id].get('crypto_id') + show_crypto_details(call.message, crypto_id, currency) + +def show_crypto_details(message, crypto_id: str, currency: str) -> None: + try: + details = get_crypto_details(crypto_id, currency) + if isinstance(details, dict): + price = details.get(currency, 'N/A') + change_24h = details.get(f'{currency}_24h_change', 'N/A') + market_cap = details.get(f'{currency}_market_cap', 'N/A') + + if isinstance(change_24h, (int, float)): + change_symbol = '+' if change_24h > 0 else ('-' if change_24h < 0 else '') + change_24h = f"{abs(change_24h):.2f}%" + else: + change_symbol = '' + change_24h = 'N/A' + + price = price if isinstance(price, (int, float)) else 'N/A' + market_cap = market_cap if isinstance(market_cap, (int, float)) else 'N/A' + + text = ( + f"{crypto_id.capitalize()} ({currency.upper()})\n" + f"Price: {price} {currency.upper()}\n" + f"24h Change: {change_symbol} {change_24h}\n" + f"Market Cap: {market_cap} {currency.upper()}" + ) + else: + text = f"Sorry, I couldn't find the details for {crypto_id}. Please ensure the cryptocurrency ID is correct." + + except Exception as e: + logger.error(f"Error retrieving crypto details: {e}") + text = "An error occurred while fetching cryptocurrency data. Please try again later." + + keyboard = InlineKeyboardMarkup() + keyboard.add(InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')) + bot.send_message(message.chat.id, text, reply_markup=keyboard) # Message Handler -async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: - user_input = update.message.text.lower() - search_results = requests.get(f"{COINGECKO_API_URL}/search", params={'query': user_input}).json() - coins = search_results.get('coins', []) - - if coins: - await show_crypto_list(update, context, coins[:10], "Search Results:") - return CHOOSING_CRYPTO - else: - await update.message.reply_text("Sorry, I couldn't find any cryptocurrency matching your search.") - await show_main_menu(update, context) - return MAIN_MENU - -# Error Handler -async def error(update: Update, context: ContextTypes.DEFAULT_TYPE): - print(f"Update {update} caused error {context.error}") +@bot.message_handler(func=lambda message: True) +def handle_message(message: Message) -> None: + user_input = message.text.lower() + try: + # Fetch search results from CoinGecko + search_results = requests.get(f"{COINGECKO_API_URL}/search", params={'query': user_input}).json() + coins = search_results.get('coins', []) + + if coins: + # Only take the first 10 results to display + detailed_coins = [{'id': coin['id'], 'name': coin['name'], + 'symbol': coin['symbol'], 'image': coin.get('thumb', '')} + for coin in coins[:10]] + show_crypto_list(message, detailed_coins, "Search Results:") + else: + bot.send_message(message.chat.id, "Sorry, I couldn't find any cryptocurrency matching your search.") + show_main_menu(message) + except requests.RequestException as e: + logger.error(f"Error searching for cryptocurrency: {e}") + bot.send_message(message.chat.id, "An error occurred while searching for the cryptocurrency.") + show_main_menu(message) def main() -> None: - app = Application.builder().token(BOT_TOKEN).build() - - conv_handler = ConversationHandler( - entry_points=[CommandHandler("start", start)], - states={ - MAIN_MENU: [CallbackQueryHandler(button_click)], - CHOOSING_CRYPTO: [CallbackQueryHandler(button_click)], - CHOOSING_CURRENCY: [CallbackQueryHandler(button_click)], - TYPING_SEARCH: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)], - }, - fallbacks=[CommandHandler("start", start)], - per_message=False - ) - - app.add_handler(conv_handler) - app.add_handler(CommandHandler("help", help_command)) - app.add_error_handler(error) - - print('Starting bot...') - app.run_polling(poll_interval=3) + logger.info('Starting bot...') + bot.polling() if __name__ == '__main__': - main() + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index b12c62f..25ffdd2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -python-telegram-bot==20.5 -requests==2.31.0 +pyTelegramBotAPI +requests