Update main.py

main.py

@@ -13,39 +13,39 @@ import json
 
 app = FastAPI()
 
-async def kafka_consumer():
-    consumer_topic = 'hugging-face-raw-data-topic'
-    producer_topic = 'hugging-face-processed-data-topic'
-    bootstrap_server = '138.197.1.52:9092'
-    consumer = KafkaConsumer(
-        consumer_topic,
-        group_id='hugging-face-group',
-        bootstrap_servers=bootstrap_server,
-        auto_offset_reset='earliest', max_poll_records=50,
-        enable_auto_commit=False
-    )
+# async def kafka_consumer():
+#     consumer_topic = 'hugging-face-raw-data-topic'
+#     producer_topic = 'hugging-face-processed-data-topic'
+#     bootstrap_server = '138.197.1.52:9092'
+#     consumer = KafkaConsumer(
+#         consumer_topic,
+#         group_id='hugging-face-group',
+#         bootstrap_servers=bootstrap_server,
+#         auto_offset_reset='earliest', max_poll_records=50,
+#         enable_auto_commit=False
+#     )
 
-    try:
-        for message in consumer:
-            data_str = message.value.decode('utf-8')
-            data = json.loads(data_str)
-            obj_id = data.get('obj_id')
-            raw_data = {"text":data.get('text', str()), 'star_rating':data.get('rating', 5), "skip":False}
-            processed_data, has_sentiments = process_single_comment(raw_data)
-            print(f"Processed message with id : {obj_id}")
-            processed_data.pop('color_map', None)
-            processed_data_dict = {"processed_data":processed_data, "has_sentiments":has_sentiments,"obj_id":obj_id}
-            producer = KafkaProducer(bootstrap_servers=bootstrap_server)
-            msg = json.dumps(processed_data_dict)
-            producer.send(producer_topic, msg.encode('utf-8'))
-            producer.flush()
-    finally:
-        consumer.close()
+#     try:
+#         for message in consumer:
+#             data_str = message.value.decode('utf-8')
+#             data = json.loads(data_str)
+#             obj_id = data.get('obj_id')
+#             raw_data = {"text":data.get('text', str()), 'star_rating':data.get('rating', 5), "skip":False}
+#             processed_data, has_sentiments = process_single_comment(raw_data)
+#             print(f"Processed message with id : {obj_id}")
+#             processed_data.pop('color_map', None)
+#             processed_data_dict = {"processed_data":processed_data, "has_sentiments":has_sentiments,"obj_id":obj_id}
+#             producer = KafkaProducer(bootstrap_servers=bootstrap_server)
+#             msg = json.dumps(processed_data_dict)
+#             producer.send(producer_topic, msg.encode('utf-8'))
+#             producer.flush()
+#     finally:
+#         consumer.close()
 
-@app.on_event("startup")
-async def startup_event():
-    # Start the Kafka consumer on application startup
-    asyncio.create_task(kafka_consumer())
+# @app.on_event("startup")
+# async def startup_event():
+#     # Start the Kafka consumer on application startup
+#     asyncio.create_task(kafka_consumer())
 
 class TextRatingRequest(BaseModel):
     text: str