Handle the input style

worker.py

@@ -87,23 +87,32 @@ class RabbitMQWorker:
             if headers.get("request_type") == "process_files":
                 ch.basic_ack(delivery_tag=method.delivery_tag)
                 contexts = []
+
                 body_dict = json.loads(body)
+
+                data = body_dict.get("data", {})
+                input_files = data.get("input_files", [])
                 
-                # Process files
-                for file in body_dict.get("input_files", []):
+                logger.info("[Worker %s] Found %d file(s) to process.", thread_id, len(input_files))
+
+                for files in input_files:
                     try:
-                        logger.info("[Worker %s] Processing %d input files.", thread_id, len(file))
-                        context = {"key": file["key"], "body": self.processor.process(file["url"], file["key"])}
+                        context = {
+                            "key": files["key"],
+                            "body": self.processor.process(files["url"], files["key"])
+                        }
                         contexts.append(context)
                     except Exception as e:
-                        print(f"Error processing file {file['key']}: {e}")
-                        contexts.append({"key": file["key"], "body": f"Error: {str(e)}"})
+                        err_str = f"Error processing file {files.get('key', '')}: {e}"
+                        logger.error(err_str)
+                        contexts.append({"key": files.get("key", ""), "body": err_str})
 
-                body_dict["md_context"] = contexts
+
+                data["md_context"] = contexts
+                # topics = data.get("topics", [])
 
                 headers["request_type"] = "question_extraction_update_from_gpu_server"
                 
-                
                 # Publish results
                 if self.publish_message(body_dict, headers):
                     logger.info("[Worker %s] Successfully published results to ml_server.", thread_id)