app.py

import gradio as gr
import shap
from transformers import pipeline
from transformers import DistilBertTokenizer, DistilBertForSequenceClassification
import streamlit as st
import sys
import os
import pandas as pd
import json
from confluent_kafka import Consumer
from ast import literal_eval

consumer = Consumer(
         {'bootstrap.servers': 'pkc-41973.westus2.azure.confluent.cloud:9092',
        "group.id": "group_data_h",
        'security.protocol':'SASL_SSL',
        'sasl.mechanisms':'PLAIN',
        'sasl.username':'AIZHFU6TZHAQC5E3',
        'sasl.password':os.environ.get("confluent_ingreso"),
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': True })

consumer.subscribe(['factored_datathon_amazon_review_1'])
i=0
received=[]
df = {}

model_name = "sohan-ai/sentiment-analysis-model-amazon-reviews"
tokenizer = DistilBertTokenizer.from_pretrained("distilbert-base-uncased")
model = DistilBertForSequenceClassification.from_pretrained(model_name)


input_text="Awaiting Reviews"

def interpretation_function(text):
    inputs = tokenizer(text, return_tensors="pt")
    outputs = model(**inputs)
    predicted_label = "positive" if outputs.logits.argmax().item() == 1 else "negative"
    return {"Review": text, "interpretation": predicted_label}

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            input_text = gr.Textbox(label="Sentiment Analysis", value=input_text)
            with gr.Row():
                interpret = gr.Button("Interpret Review")
        with gr.Column():
                interpretation = gr.components.Interpretation(input_text)
        demo.load(interpretation_function, input_text, interpretation,every=60)
        

    interpret.click(interpretation_function, input_text, interpretation)
demo.queue(api_open=False)
try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        user = msg.value()
        if user is not None:
            nus=literal_eval(user.decode('utf8'))
            dato=json.loads(json.dumps(nus, indent=4))
            df[i] = dato
            df_t=pd.DataFrame.from_dict(df, orient='index')
            
            
            input_text = df_t.iloc[[i],[2]]

            i += 1
            
except SystemExit:
  print('closing the consumer')
  consumer.close()


if __name__ == "__main__":
    demo.launch()