Spaces:
Sleeping
Sleeping
| import os | |
| import asyncio | |
| import threading | |
| from telegram import Update | |
| from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes | |
| from transformers import pipeline | |
| import gradio as gr | |
| # Telegram bot token (from environment variable) | |
| TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") | |
| # Load AI model for text generation | |
| generator = pipeline("text-generation", model="distilgpt2") | |
| # Telegram bot handlers | |
| async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): | |
| await update.message.reply_text("Hello! I'm a chatbot powered by an AI model. Send me a message, and I'll respond!") | |
| async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): | |
| user_message = update.message.text | |
| response = generator(user_message, max_length=50, num_return_sequences=1, truncation=True)[0]['generated_text'] | |
| await update.message.reply_text(response) | |
| async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): | |
| print(f"Error occurred: {context.error}") | |
| if "Conflict" in str(context.error): | |
| await context.bot.send_message( | |
| chat_id=update.effective_chat.id if update else None, | |
| text="Bot conflict detected. Restarting..." | |
| ) | |
| # Telegram bot logic | |
| async def telegram_bot(): | |
| if not TELEGRAM_BOT_TOKEN: | |
| print("Telegram bot token not found. Please set TELEGRAM_BOT_TOKEN.") | |
| return | |
| application = Application.builder().token(TELEGRAM_BOT_TOKEN).build() | |
| # Add handlers | |
| application.add_handler(CommandHandler("start", start)) | |
| application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) | |
| application.add_error_handler(error_handler) | |
| # Clear webhook and updates | |
| print("Clearing any existing webhook and updates...") | |
| try: | |
| await application.bot.delete_webhook(drop_pending_updates=True) | |
| updates = await application.bot.get_updates(timeout=1) | |
| if updates: | |
| last_update_id = updates[-1].update_id | |
| await application.bot.get_updates(offset=last_update_id + 1, timeout=1) | |
| except Exception as e: | |
| print(f"Error clearing updates: {e}") | |
| # Start bot polling | |
| print("Starting Telegram bot...") | |
| await application.run_polling(allowed_updates=Update.ALL_TYPES) | |
| # Gradio logic | |
| def gradio_app(): | |
| iface = gr.Interface(fn=lambda x: "Hello " + x, inputs="text", outputs="text") | |
| iface.launch() | |
| # Combine both | |
| if __name__ == "__main__": | |
| # Step 1: Start Gradio in a separate thread | |
| gradio_thread = threading.Thread(target=gradio_app) | |
| gradio_thread.start() | |
| # Step 2: Start Telegram bot in the main thread, handling event loop properly | |
| try: | |
| loop = asyncio.get_running_loop() | |
| except RuntimeError: | |
| loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(loop) | |
| loop.run_until_complete(telegram_bot()) | |