app.py

import spaces
import os
from dotenv import load_dotenv
import re
from urllib.parse import urlparse
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
import gradio as gr
import logging
import requests

# Load environment variables
load_dotenv()

# Set command line arguments for Gradio
os.environ["COMMANDLINE_ARGS"] = "--no-gradio-queue"

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

import http.client

http.client.HTTPConnection.debuglevel = 1
req_log = logging.getLogger("requests.packages.urllib3")
req_log.setLevel(logging.DEBUG)
req_log.propagate = True

# Constants
LIMIT = 1000  # Limit to 1000 reviews to avoid long processing times
OpenAIModel = "gpt-3.5-turbo"
shop_id = ""
item_id = ""
item = {}
cache_URL = ""
db = None
qa = None
cache = {}

import json

# Function to request product ID from Tokopedia
def request_product_id(shop_domain, product_key, url):
    endpoint = "https://gql.tokopedia.com/graphql/PDPGetLayoutQuery"
    payload = {
        "operationName": "PDPGetLayoutQuery",
        "variables": {
            "shopDomain": f"{shop_domain}",
            "productKey": f"{product_key}",
            "apiVersion": 1,
        },
        "query": "fragment ProductVariant on pdpDataProductVariant { errorCode parentID defaultChild children { productID } __typename } query PDPGetLayoutQuery($shopDomain: String, $productKey: String, $layoutID: String, $apiVersion: Float, $userLocation: pdpUserLocation, $extParam: String, $tokonow: pdpTokoNow, $deviceID: String) { pdpGetLayout(shopDomain: $shopDomain, productKey: $productKey, layoutID: $layoutID, apiVersion: $apiVersion, userLocation: $userLocation, extParam: $extParam, tokonow: $tokonow, deviceID: $deviceID) { requestID name pdpSession basicInfo { id: productID } components { name type position data { ...ProductVariant __typename } __typename } __typename } }",
    }

    headers = {
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36",
        "Referer": "https://www.tokopedia.com",
        "X-TKPD-AKAMAI": "pdpGetLayout",
    }

    return requests.request(
        method="POST", url=endpoint, json=payload, headers=headers, timeout=30
    )


# Function to request product reviews from Tokopedia
def request_product_review(product_id, page=1, limit=20):
    ENDPOINT = "https://gql.tokopedia.com/graphql/productReviewList"
    payload = {
        "operationName": "productReviewList",
        "variables": {
            "productID": f"{product_id}",
            "page": page,
            "limit": limit,
            "sortBy": "",
            "filterBy": "",
        },
        "query": """query productReviewList($productID: String!, $page: Int!, $limit: Int!, $sortBy: String, $filterBy: String) {
  productrevGetProductReviewList(productID: $productID, page: $page, limit: $limit, sortBy: $sortBy, filterBy: $filterBy) {
    productID
    list {
      id: feedbackID
      variantName
      message
      productRating
      reviewCreateTime
      reviewCreateTimestamp
      isReportable
      isAnonymous
      reviewResponse {
        message
        createTime
        __typename
      }
      user {
        userID
        fullName
        image
        url
        __typename
      }
      likeDislike {
        totalLike
        likeStatus
        __typename
      }
      stats {
        key
        formatted
        count
        __typename
      }
      badRatingReasonFmt
      __typename
    }
    shop {
      shopID
      name
      url
      image
      __typename
    }
    hasNext
    totalReviews
    __typename
  }
}
                        """,
    }
    headers = {
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36",
        "Referer": "https://www.tokopedia.com",
        "X-TKPD-AKAMAI": "productReviewList",
    }
    try:
        response = requests.post(ENDPOINT, json=payload, headers=headers, timeout=60)
        response.raise_for_status()
        logger.info(f"Request successful. Status code: {response.status_code}")
        return response
    except requests.exceptions.RequestException as e:
        logger.error(f"Request failed: {e}")
        return None


# Function to scrape reviews for a product
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)
        if not response:
            break
        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


# Function to extract product ID from URL
def get_product_id(URL):
    parsed_url = urlparse(URL)
    *_, shop, product_key = parsed_url.path.split("/")
    response = request_product_id(shop, product_key, URL)
    if response:
        product_id = response.json()["data"]["pdpGetLayout"]["basicInfo"]["id"]
        logger.info(f"Product ID: {product_id}")
        return product_id
    else:
        logger.error("Failed to get product ID")
        return None


# Function to clean the reviews DataFrame
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 reviews DataFrame")
    return df


# Function to clean individual text entries
def clean_text(text):
    text = uni.normalize("NFKD", text)  # Normalize characters
    text = emoji.replace_emoji(text, "")  # Remove emoji
    text = re.sub(r"(\w)\1{2,}", r"\1", text)  # Remove repeated characters
    text = re.sub(r"[ ]+", " ", text).strip()  # Remove extra spaces
    return text


# Initialize LLM and embeddings
llm = ChatOpenAI(model=OpenAIModel, temperature=0.1)
embeddings = HuggingFaceEmbeddings(model_name="LazarusNLP/all-indobert-base-v2")


# Function to generate a summary or answer based on reviews
@spaces.GPU
async def generate(URL, query):
    global cache_URL, db, qa, cache

    if not URL or not query:
        return "Input kosong"
    try:
        product_id = get_product_id(URL)
        if not product_id:
            return "Gagal mendapatkan product ID"

        if URL not in cache:
            reviews = scrape(product_id)
            if reviews.empty:
                return "Tidak ada ulasan ditemukan"

            cleaned_reviews = clean(reviews)
            loader = DataFrameLoader(cleaned_reviews, page_content_column="comment")
            documents = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000, chunk_overlap=50
            )
            docs = text_splitter.split_documents(documents)
            db = FAISS.from_documents(docs, embeddings)
            cache[URL] = (docs, db)
        else:
            docs, db = cache[URL]

        qa = RetrievalQA.from_chain_type(llm=llm, retriever=db.as_retriever())
        res = await qa.ainvoke(query)
        return res["result"]
    except Exception as e:
        logger.error(f"Error in generating response: {e}")
        return "Gagal mendapatkan review dari URL"


# Set up Gradio interface
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)