feat: add concurrent Excel file processing with status tracking

Implement multi-file Excel processing system with ThreadPoolExecutor,
detailed status tracking, and new /excelstatus command for monitoring progress

bot_telegram.py

@@ -55,6 +55,11 @@ class TelegramBot:
         # Create a ThreadPoolExecutor for handling concurrent requests
         self.executor = ThreadPoolExecutor(max_workers=10)
 
+        # Increase Excel workers for more concurrent processing
+        self.excel_executor = ThreadPoolExecutor(max_workers=10, thread_name_prefix="excel_worker")
+        self.excel_semaphore = asyncio.Semaphore(10)  # Allow 10 concurrent Excel processes
+        self.active_excel_files = {}
+
     # def authenticate(self):
     #     """Authenticate with the API and retrieve an access token."""
     #     payload = {"username": self.username, "password": self.password}
@@ -191,8 +196,9 @@ class TelegramBot:
             return f"Connection error: {e}"
 
     async def handle_excel(self, update: Update, context: CallbackContext):
-        """Handles Excel file uploads."""
+        """Handles Excel file uploads asynchronously"""
         user_id = update.message.from_user.id
+        message_id = update.message.message_id
 
         if user_id not in AUTHENTICATED_USERS:
             await update.message.reply_text("❌ You are not authenticated. Please enter the password first.")
@@ -210,105 +216,113 @@ class TelegramBot:
             await update.message.reply_text("Please send only Excel files (.xls or .xlsx)")
             return
 
-        # Send initial processing message
-        processing_msg = await update.message.reply_text("📊 Processing your Excel file... This may take a few minutes.")
+        # Track this Excel file
+        self.active_excel_files[message_id] = {
+            'filename': document.file_name,
+            'start_time': time.time(),
+            'status': 'starting'
+        }
 
         try:
-            # Run Excel processing in a separate thread
-            result = await asyncio.get_event_loop().run_in_executor(
-                self.executor,
-                self._process_excel,
-                update, context, document
+            # Send initial processing message
+            processing_msg = await update.message.reply_text(
+                f"📊 Starting Excel file processing...\n"
+                f"File: {document.file_name}\n"
+                f"Request ID: #{message_id}"
             )
-            
-            if isinstance(result, tuple):
-                file_content, error = result
-                if error:
-                    await processing_msg.edit_text(f"❌ Error: {error}")
-                else:
-                    await context.bot.send_document(
-                        chat_id=update.effective_chat.id,
-                        document=io.BytesIO(file_content),
-                        filename='rfp_responses.xlsx',
-                        caption="✅ Here's your processed Excel file with answers!"
+
+            # 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}"
+                )
+
+                # 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}"
+                )
+
+                # 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
+                )
+
+                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}"
                     )
-                    await processing_msg.delete()
+                    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'
+
         except Exception as e:
             logging.error(f"Error handling Excel file: {e}")
-            await processing_msg.edit_text(f"❌ Error processing Excel file: {str(e)}")
+            self.active_excel_files[message_id]['status'] = 'error'
+            self.active_excel_files[message_id]['error'] = str(e)
+            await processing_msg.edit_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
 
-    async def _process_excel(self, update, context, document):
-        """Process Excel file in a separate thread"""
-        try:
-            file = await context.bot.get_file(document.file_id)
-            file_bytes = await file.download_as_bytearray()
-
-            headers = {
-                "Authorization": f"Bearer {HF_TOKEN}",
-                "accept": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
-            }
-
-            async with aiohttp.ClientSession() as session:
-                form_data = aiohttp.FormData()
-                form_data.add_field('file',
-                                  file_bytes,
-                                  filename=document.file_name,
-                                  content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
-
-                async with session.post(self.excel_url, headers=headers, data=form_data) as response:
-                    if response.status == 200:
-                        return await response.read(), None
-                    else:
-                        error_text = await response.text()
-                        return None, error_text
+        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"
+            )
 
-        except Exception as e:
-            return None, str(e)
-
-    async def test_concurrent(self, update: Update, context: CallbackContext):
-        """Test concurrent processing"""
-        await update.message.reply_text("Sending 5 requests simultaneously...")
-        
-        tasks = []
-        for i in range(5):
-            msg = f"Test question {i+1}"
-            tasks.append(self.chat_with_ai(update, context))
-        
-        await asyncio.gather(*tasks)
-
-    async def load_test_command(self, update: Update, context: CallbackContext):
-        """Command to test concurrent processing"""
-        if not context.args or not context.args[0].isdigit():
-            await update.message.reply_text("Usage: /loadtest <number_of_requests>")
-            return
-            
-        num_requests = min(int(context.args[0]), 20)  # Limit to 20 for safety
-        await update.message.reply_text(f"Starting load test with {num_requests} concurrent requests...")
-        
-        start_time = time.time()
-        tasks = []
-        for i in range(num_requests):
-            tasks.append(self.chat_with_ai(
-                update, 
-                context, 
-                f"Test question {i+1}"
-            ))
-        
-        await asyncio.gather(*tasks)
-        duration = time.time() - start_time
-        
-        await update.message.reply_text(
-            f"Load test completed!\n"
-            f"Processed {num_requests} requests in {duration:.2f} seconds\n"
-            f"Average time per request: {(duration/num_requests):.2f} seconds"
-        )
+        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("loadtest", self.load_test_command))
+        self.app.add_handler(CommandHandler("excelstatus", self.excel_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"), self.handle_excel))
+        self.app.add_handler(MessageHandler(
+            filters.Document.FileExtension("xlsx") | filters.Document.FileExtension("xls"),
+            self.handle_excel
+        ))
 
     async def run(self):
         """Start the bot and listen for messages."""