Upload rabbit_mq_investor_report.py

rabbit_mq_investor_report.py

@@ -0,0 +1,141 @@
+import os
+import pika
+import json
+from datetime import datetime
+
+import requests
+import sys
+sys.path.append("/shared_disk/kushal/db_str_chunking/new_ws_structured_code")
+from db_structured_chunking.structure_chunking.API_backend.iqeq_app_latest import upload_documents
+from db_structured_chunking.structure_chunking.API_backend.rabbitmq_config_investor_report import RABBITMQ
+ 
+# RabbitMQ connection parameters
+connection_params = pika.ConnectionParameters(
+    host=RABBITMQ["HOST"],
+    port=RABBITMQ["PORT"],
+    virtual_host=RABBITMQ["VIRTUAL_HOST"],
+    credentials=pika.PlainCredentials(
+        RABBITMQ["USERNAME"], RABBITMQ["PASSWORD"]
+    )
+)
+
+# Define the names of the queues
+EXCHANGE_NAME = RABBITMQ["EXCHANGE_NAME"]
+EXCHANGE_TYPE = RABBITMQ["EXCHANGE_TYPE"]
+ 
+INPUT_FILE_QUEUE = RABBITMQ["QUEUES"]["INPUT_FILE_QUEUE"]
+FILE_RESPONSE_QUEUE = RABBITMQ["QUEUES"]["FILE_RESPONSE_QUEUE"]
+ 
+PRIORITY_ARGS = {'x-max-priority': 5}
+ 
+def on_message(channel, method_frame, header_frame, body):
+    processingStartTime = datetime.now().isoformat()
+
+    msg = json.loads(body)
+    print(f"::::::::::::::::{msg}")
+    filePath = msg.get("path")
+    result = upload_documents(filePath)
+    response_json_path = result["saved_json_path"]
+
+    response_message = {
+ 
+            "documentId": msg["documentId"],
+ 
+            "documentType": msg["documentType"],
+ 
+            "documentSubType": msg["documentSubType"],
+ 
+            "status": 200,
+ 
+            "accuracy": "",
+ 
+            "queueInsertTime": msg["queueInsertTime"],
+ 
+            "processingStartTime": "",
+ 
+            "processingEndTime": "",
+ 
+            "manualQaRequired": "",
+ 
+            "errorMessages": None,
+ 
+            "result": {"path":response_json_path} 
+ 
+        }
+ 
+    print(f"::::::::::::::::::::::::::::::::::\n{response_message}")
+ 
+    # Publish the response to the response queue
+    channel.basic_publish(
+        exchange=EXCHANGE_NAME,
+        routing_key=FILE_RESPONSE_QUEUE,
+        body=json.dumps(response_message)
+    )
+ 
+    # Acknowledge the original message
+    channel.basic_ack(delivery_tag=method_frame.delivery_tag)
+ 
+ 
+def main():
+    # Establish connection to RabbitMQ
+    try:
+        connection = pika.BlockingConnection(connection_params)
+        channel = connection.channel()
+        print("Connection to RabbitMQ established successfully!")
+    except Exception as e:
+        print(f"Failed to connect to RabbitMQ:{e}")
+        return
+ 
+    channel.exchange_declare(
+        exchange=EXCHANGE_NAME,
+        exchange_type=EXCHANGE_TYPE,
+        durable=True
+    )
+ 
+    # Declare the queues
+    channel.queue_declare(
+        queue = INPUT_FILE_QUEUE,
+        durable=True,
+        arguments=PRIORITY_ARGS
+    )
+ 
+    channel.queue_declare(
+        queue=FILE_RESPONSE_QUEUE,
+        durable=True,
+        arguments=PRIORITY_ARGS
+    )
+    print(f"Queues '{INPUT_FILE_QUEUE}' and '{FILE_RESPONSE_QUEUE}' are ready.")
+ 
+    # Bind each queue to the exchange with a routing key matching the queue name
+    channel.queue_bind(
+        queue=INPUT_FILE_QUEUE,
+        exchange=EXCHANGE_NAME,
+        routing_key=INPUT_FILE_QUEUE
+    )
+    channel.queue_bind(
+        queue=FILE_RESPONSE_QUEUE,
+        exchange=EXCHANGE_NAME,
+        routing_key=FILE_RESPONSE_QUEUE
+    )
+ 
+    print(f"Exchange '{EXCHANGE_NAME}' declared and queues bound.")
+ 
+    # Start consuming messages from the request queue
+    channel.basic_consume(
+        queue=INPUT_FILE_QUEUE,
+        on_message_callback=on_message
+    )
+ 
+    print("Waiting for messages. To exit press CTRL+C")
+ 
+    try:
+        channel.start_consuming()
+    except KeyboardInterrupt:
+        channel.stop_consuming()
+    finally:
+        connection.close()
+        print("Connection closed.")
+ 
+ 
+if __name__ == '__main__':
+    main()