app.py

import spaces

from dotenv import load_dotenv
load_dotenv()

import re
from urllib.parse import urlparse, parse_qs
import pandas as pd
import unicodedata as uni
import emoji
from langchain_openai import ChatOpenAI
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_loaders import DataFrameLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from tokopedia import request_product_id, request_product_review
import gradio as gr

shop_id = ""
item_id = ""
item = {}
LIMIT = 1000  # Limit to 1000 reviews so that processing does not take too long

import logging

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()],
)

logger = logging.getLogger(__name__)


def scrape(product_id, max_reviews=LIMIT):
    all_reviews = []
    page = 1
    has_next = True

    logger.info("Extracting product reviews...")
    while has_next and len(all_reviews) <= max_reviews:
        response = request_product_review(product_id, page=page)
        data = response.json()["data"]["productrevGetProductReviewList"]
        reviews = data["list"]
        all_reviews.extend(reviews)
        has_next = data["hasNext"]
        page += 1

    reviews_df = pd.json_normalize(all_reviews)
    reviews_df.rename(columns={"message": "comment"}, inplace=True)
    reviews_df = reviews_df[["comment"]]
    logger.info(reviews_df.head())
    return reviews_df


def get_product_id(URL):
    parsed_url = urlparse(URL)
    *_, SHOP, PRODUCT_KEY = parsed_url.path.split("/")
    product_id = request_product_id(SHOP, PRODUCT_KEY).json()["data"]["pdpGetLayout"][
        "basicInfo"
    ]["id"]
    logger.info(product_id)
    return product_id


# Clean
def clean(df):
    df = df.dropna().copy().reset_index(drop=True)  # drop reviews with empty comments
    df = df[df["comment"] != ""].reset_index(drop=True)  # remove empty reviews
    df["comment"] = df["comment"].apply(lambda x: clean_text(x))  # clean text
    df = df[df["comment"] != ""].reset_index(drop=True)  # remove empty reviews
    logger.info("cleaned")
    return df


def clean_text(text):
    text = uni.normalize("NFKD", text)  # normalise characters
    text = emoji.replace_emoji(text, "")  # remove emoji
    text = re.sub(r"(\w)\1{2,}", r"\1", text)  # repeated chars
    text = re.sub(r"[ ]+", " ", text).strip()  # remove extra spaces
    return text


# LLM
OpenAIModel = "gpt-3.5-turbo"
llm = ChatOpenAI(model=OpenAIModel, temperature=0.1)

# Embeddings
embeddings = HuggingFaceEmbeddings(model_name="LazarusNLP/all-indobert-base-v2")

cache_URL = ""
db = None
qa = None
cache = {}


@spaces.GPU
async def generate(URL, query):
    global cache_URL, db, qa, cache

    if URL == "" or query == "":
        return "Empty input"
    else:
        try:
            product_id = get_product_id(URL)

            if URL not in cache:
                # Get reviews
                try:
                    reviews = scrape(product_id)
                    # Clean reviews
                    cleaned_reviews = clean(reviews)
                    # Load data
                    loader = DataFrameLoader(
                        cleaned_reviews, page_content_column="comment"
                    )
                    documents = loader.load()
                except Exception as e:
                    return "Error getting reviews: " + str(e)
                else:
                    # Split text
                    text_splitter = RecursiveCharacterTextSplitter(
                        chunk_size=1000, chunk_overlap=50
                    )
                    docs = text_splitter.split_documents(documents)

                    # Vector store
                    db = FAISS.from_documents(docs, embeddings)

                    # Store in cache
                    cache[URL] = (docs, db)

            # Retrieve from cache
            docs, db = cache[URL]

            # Chain to answer questions
            qa = RetrievalQA.from_chain_type(llm=llm, retriever=db.as_retriever())
            res = await qa.ainvoke(query)

            # Process result
            return res["result"]
        except:
            return "URL tidak valid"


# Gradio
product_box = gr.Textbox(label="URL Produk", placeholder="URL produk dari Tokopedia")
query_box = gr.Textbox(
    lines=2,
    label="Kueri",
    placeholder="Contoh: Apa yang orang katakan tentang kualitas produknya?, Bagaimana pendapat orang yang kurang puas dengan produknya?",
)

gr.Interface(
    fn=generate,
    inputs=[product_box, query_box],
    outputs=[gr.Textbox(label="Jawaban")],
    title="RingkasUlas",
    description="Bot percakapan yang bisa meringkas ulasan-ulasan produk di Tokopedia Indonesia (https://tokopedia.com/). Harap bersabar, bot ini dapat memakan waktu agak lama saat mengambil ulasan dari Tokopedia dan menyiapkan jawabannya.",
    allow_flagging="never",
).launch(debug=True)