feat: implement true concurrent processing

Add separate thread pools and async tasks for both text and Excel processing,
enabling simultaneous handling of multiple requests without blocking

bot_telegram.py

@@ -44,6 +44,14 @@ class TelegramBot:
         self.ai_url = f"{self.base_url}/api/v1/questions/text"
         self.excel_url = f"{self.base_url}/api/v1/questions/excel"
 
+        # Separate executors for text and Excel
+        self.text_executor = ThreadPoolExecutor(max_workers=20, thread_name_prefix="text_worker")
+        self.excel_executor = ThreadPoolExecutor(max_workers=10, thread_name_prefix="excel_worker")
+        
+        # Track active processes
+        self.active_requests = {}
+        self.active_excel_files = {}
+
         # Start Telegram Bot
         self.app = Application.builder().token(self.bot_token).build()
         self.setup_handlers()
@@ -94,9 +102,10 @@ class TelegramBot:
             await update.message.reply_text("🔑 Please enter the secret password to access the bot.")
 
     async def handle_message(self, update: Update, context: CallbackContext):
-        """Handles all incoming messages."""
+        """Handles all incoming messages concurrently"""
         user_id = update.message.from_user.id
         user_message = update.message.text.strip()
+        message_id = update.message.message_id
 
         # If user is waiting to enter a password, validate it
         if user_id in AWAITING_PASSWORD:
@@ -105,7 +114,8 @@ class TelegramBot:
 
         # If user is authenticated, process AI request
         if user_id in AUTHENTICATED_USERS:
-            await self.chat_with_ai(update, context)
+            # Create task for processing
+            asyncio.create_task(self.chat_with_ai(update, context))
         else:
             await update.message.reply_text("❌ You are not authenticated. Please enter the password first.")
 
@@ -129,38 +139,45 @@ class TelegramBot:
             await update.message.reply_text("❌ Wrong password. Try again.")
 
     async def chat_with_ai(self, update: Update, context: CallbackContext):
-        """Handles messages and sends them to the AI API."""
-        user_id = update.message.from_user.id
-
-        if user_id not in AUTHENTICATED_USERS:
-            await update.message.reply_text("❌ You are not authenticated. Please enter the password first.")
-            return
-
-        # if not self.auth_token:
-        #     self.authenticate()
-        #
-        # if not self.auth_token:
-        #     await update.message.reply_text("Authentication failed. Please try again later.")
-        #     return
-
+        """Process text messages asynchronously"""
+        message_id = update.message.message_id
         user_message = update.message.text
-        # Send immediate acknowledgment
-        processing_msg = await update.message.reply_text("Processing your request...")
 
         try:
-            # Run the API request in a separate thread to avoid blocking
+            # Send immediate acknowledgment
+            processing_msg = await update.message.reply_text(
+                f"🤔 Processing your request...\n"
+                f"Request ID: #{message_id}"
+            )
+
+            # Track this request
+            self.active_requests[message_id] = {
+                'type': 'text',
+                'status': 'processing',
+                'start_time': time.time()
+            }
+
+            # Process in thread pool
             response = await asyncio.get_event_loop().run_in_executor(
-                self.executor,
+                self.text_executor,
                 self._make_api_request,
                 user_message
             )
-            
-            # Update the processing message with the response
-            await processing_msg.edit_text(response)
+
+            # Update with response
+            await processing_msg.edit_text(
+                f"✅ Response for #{message_id}:\n{response}"
+            )
 
         except Exception as e:
-            logging.error(f"Error processing request: {e}")
-            await processing_msg.edit_text(f"❌ Error: {str(e)}")
+            logging.error(f"Error processing text request: {e}")
+            await processing_msg.edit_text(
+                f"❌ Error processing request #{message_id}: {str(e)}"
+            )
+        finally:
+            if message_id in self.active_requests:
+                self.active_requests[message_id]['status'] = 'completed'
+                self.active_requests[message_id]['end_time'] = time.time()
 
     def _make_api_request(self, user_message):
         """Make the actual API request in a separate thread"""
@@ -196,128 +213,114 @@ class TelegramBot:
             return f"Connection error: {e}"
 
     async def handle_excel(self, update: Update, context: CallbackContext):
-        """Handles Excel file uploads asynchronously"""
-        user_id = update.message.from_user.id
+        """Handle Excel files concurrently"""
         message_id = update.message.message_id
+        
+        # Create task for processing
+        asyncio.create_task(self._process_excel_file(update, context, message_id))
 
-        if user_id not in AUTHENTICATED_USERS:
-            await update.message.reply_text("❌ You are not authenticated. Please enter the password first.")
-            return
-
-        # if not self.auth_token:
-        #     self.authenticate()
-        # 
-        # if not self.auth_token:
-        #     await update.message.reply_text("Authentication failed. Please try again later.")
-        #     return
-
-        document = update.message.document
-        if not document.file_name.endswith(('.xls', '.xlsx')):
-            await update.message.reply_text("Please send only Excel files (.xls or .xlsx)")
-            return
-
-        # Track this Excel file
-        self.active_excel_files[message_id] = {
-            'filename': document.file_name,
-            'start_time': time.time(),
-            'status': 'starting'
-        }
-
+    async def _process_excel_file(self, update, context, message_id):
+        """Process Excel file asynchronously"""
         try:
+            document = update.message.document
+            
             # Send initial processing message
             processing_msg = await update.message.reply_text(
-                f"📊 Starting Excel file processing...\n"
+                f"📊 Processing Excel file...\n"
                 f"File: {document.file_name}\n"
                 f"Request ID: #{message_id}"
             )
 
-            # Use semaphore to control concurrent Excel processing
-            async with self.excel_semaphore:
-                self.active_excel_files[message_id]['status'] = 'downloading'
-                await processing_msg.edit_text(
-                    f"📥 Downloading file...\n"
-                    f"File: {document.file_name}\n"
-                    f"Request ID: #{message_id}"
-                )
+            # Track this Excel file
+            self.active_excel_files[message_id] = {
+                'filename': document.file_name,
+                'status': 'downloading',
+                'start_time': time.time()
+            }
 
-                # Download file
-                file = await context.bot.get_file(document.file_id)
-                file_bytes = await file.download_as_bytearray()
+            # Download file
+            file = await context.bot.get_file(document.file_id)
+            file_bytes = await file.download_as_bytearray()
 
-                self.active_excel_files[message_id]['status'] = 'processing'
-                await processing_msg.edit_text(
-                    f"⚙️ Processing Excel file...\n"
-                    f"File: {document.file_name}\n"
-                    f"Request ID: #{message_id}"
-                )
+            # Update status to processing
+            self.active_excel_files[message_id]['status'] = 'processing'
+            await processing_msg.edit_text(
+                f"⚙️ Processing Excel file...\n"
+                f"File: {document.file_name}\n"
+                f"Request ID: #{message_id}"
+            )
 
-                # Process Excel in thread pool
-                result = await asyncio.get_event_loop().run_in_executor(
-                    self.excel_executor,
-                    self._process_excel_sync,
-                    file_bytes,
-                    document.file_name
-                )
+            # Process in thread pool
+            result = await asyncio.get_event_loop().run_in_executor(
+                self.excel_executor,
+                self._process_excel_sync,
+                file_bytes,
+                document.file_name
+            )
 
-                if result is None:
-                    self.active_excel_files[message_id]['status'] = 'failed'
-                    await processing_msg.edit_text(
-                        f"❌ Error processing Excel file\n"
-                        f"File: {document.file_name}\n"
-                        f"Request ID: #{message_id}"
-                    )
-                    return
-
-                # Send processed file
-                self.active_excel_files[message_id]['status'] = 'sending'
-                await context.bot.send_document(
-                    chat_id=update.effective_chat.id,
-                    document=io.BytesIO(result),
-                    filename=f'processed_{document.file_name}',
-                    caption=f"✅ Excel processing completed!\nRequest ID: #{message_id}"
-                )
-                await processing_msg.delete()
-                self.active_excel_files[message_id]['status'] = 'completed'
+            if result is None:
+                raise Exception("Failed to process Excel file")
+
+            # Send processed file
+            await context.bot.send_document(
+                chat_id=update.effective_chat.id,
+                document=io.BytesIO(result),
+                filename=f'processed_{document.file_name}',
+                caption=f"✅ Excel processing completed!\nRequest ID: #{message_id}"
+            )
+            await processing_msg.delete()
 
         except Exception as e:
             logging.error(f"Error handling Excel file: {e}")
-            self.active_excel_files[message_id]['status'] = 'error'
-            self.active_excel_files[message_id]['error'] = str(e)
-            await processing_msg.edit_text(
+            await update.message.reply_text(
                 f"❌ Error processing Excel file\n"
                 f"File: {document.file_name}\n"
                 f"Request ID: #{message_id}\n"
                 f"Error: {str(e)}"
             )
         finally:
-            # Calculate processing time
-            end_time = time.time()
-            processing_time = end_time - self.active_excel_files[message_id]['start_time']
-            self.active_excel_files[message_id]['processing_time'] = processing_time
-
-    async def excel_status_command(self, update: Update, context: CallbackContext):
-        """Show status of Excel file processing"""
-        if not self.active_excel_files:
-            await update.message.reply_text("No active Excel file processing")
-            return
+            if message_id in self.active_excel_files:
+                self.active_excel_files[message_id]['status'] = 'completed'
+                self.active_excel_files[message_id]['end_time'] = time.time()
+
+    async def status_command(self, update: Update, context: CallbackContext):
+        """Show status of all active processes"""
+        status_message = "🤖 Current Status:\n\n"
+
+        # Text requests status
+        if self.active_requests:
+            status_message += "📝 Text Requests:\n"
+            for msg_id, info in self.active_requests.items():
+                current_time = time.time()
+                processing_time = current_time - info['start_time']
+                status_message += (
+                    f"Request #{msg_id}:\n"
+                    f"├─ Status: {info['status']}\n"
+                    f"└─ Time: {processing_time:.1f}s\n\n"
+                )
 
-        status_message = "📊 Excel Processing Status:\n\n"
-        for msg_id, info in self.active_excel_files.items():
-            processing_time = time.time() - info['start_time']
-            status_message += (
-                f"Request #{msg_id}:\n"
-                f"├─ File: {info['filename']}\n"
-                f"├─ Status: {info['status']}\n"
-                f"├─ Time: {processing_time:.1f}s\n"
-                f"└─ {info.get('error', '')}\n\n"
-            )
+        # Excel files status
+        if self.active_excel_files:
+            status_message += "📊 Excel Files:\n"
+            for msg_id, info in self.active_excel_files.items():
+                current_time = time.time()
+                processing_time = current_time - info['start_time']
+                status_message += (
+                    f"File #{msg_id}:\n"
+                    f"├─ Name: {info['filename']}\n"
+                    f"├─ Status: {info['status']}\n"
+                    f"└─ Time: {processing_time:.1f}s\n\n"
+                )
+
+        if not self.active_requests and not self.active_excel_files:
+            status_message += "No active processes"
 
         await update.message.reply_text(status_message)
 
     def setup_handlers(self):
         """Set up Telegram command and message handlers."""
         self.app.add_handler(CommandHandler("start", self.start_command))
-        self.app.add_handler(CommandHandler("excelstatus", self.excel_status_command))
+        self.app.add_handler(CommandHandler("status", self.status_command))
         self.app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message))
         self.app.add_handler(MessageHandler(
             filters.Document.FileExtension("xlsx") | filters.Document.FileExtension("xls"),