fix RabbitMQ

worker.py

@@ -132,26 +132,33 @@ class RabbitMQWorker:
             elif pattern == "topic_extraction":
                 data = body_dict.get("data")
                 input_files = data.get("input_files")
-                logger.info("[Worker %s] Found %d file(s) to process.", thread_id, len(input_files))
-                if not input_files or not isinstance(input_files, list):
-                    logger.error("[Worker %s] No input files provided for topic extraction.", thread_id)
+                logger.info("[Worker %s] Found %d file(s) to process for topic extraction.", thread_id, len(input_files))
+
+                topics_contexts = []
+                for file in input_files:
+                    try:
+                        pdf_url = file.get("url")
+                        logger.info("[Worker %s] Processing topic extraction for URL: %s", thread_id, pdf_url)
+                        result = self.topic_processor.process(pdf_url)
+                        context = {
+                            "key": file.get("key", ""),
+                            "body": result
+                        }
+                        topics_contexts.append(context)
+                    except Exception as e:
+                        err_str = f"Error processing topic extraction for file {file.get('key', '')}: {e}"
+                        logger.error(err_str)
+                        topics_contexts.append({"key": file.get("key", ""), "body": err_str})
+                data["topics_markdown"] = topics_contexts
+                body_dict["pattern"] = "topic_extraction_update_from_gpu_server"
+                body_dict["data"] = data
+                if self.publish_message(body_dict, headers):
+                    logger.info("[Worker %s] Successfully published topic extraction results to ml_server.", thread_id)
                     ch.basic_ack(delivery_tag=method.delivery_tag)
-                    return
-                # Use the first file's URL for topic extraction
-                pdf_url = input_files[0].get("url")
-                logger.info("[Worker %s] Processing topic extraction for URL: %s", thread_id, pdf_url)
-                try:
-                    topics_markdown = self.topic_processor.process(pdf_url)
-                    data["topics_markdown"] = topics_markdown
-                    body_dict["pattern"] = "topic_extraction_update_from_gpu_server"
-                    body_dict["data"] = data
-                    if self.publish_message(body_dict, headers):
-                        ch.basic_ack(delivery_tag=method.delivery_tag)
-                    else:
-                        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
-                except Exception as e:
-                    logger.error("Error processing topic extraction: %s", e)
+                else:
                     ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+                
+                logger.info("[Worker %s] Contexts: %s", thread_id, contexts)
 
             else:
                 ch.basic_ack(delivery_tag=method.delivery_tag, requeue=False)