clean up

app.py

@@ -1,12 +1,8 @@
 import spaces
 import os
-os.environ["COMMANDLINE_ARGS"] = "--no-gradio-queue"
-
 from dotenv import load_dotenv
-load_dotenv()
-
 import re
-from urllib.parse import urlparse, parse_qs
+from urllib.parse import urlparse
 import pandas as pd
 import unicodedata as uni
 import emoji
@@ -16,15 +12,15 @@ 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
+import logging
+import requests
 
-shop_id = ""
-item_id = ""
-item = {}
-LIMIT = 1000  # Limit to 1000 reviews so that processing does not take too long
+# Load environment variables
+load_dotenv()
 
-import logging
+# Set command line arguments for Gradio
+os.environ["COMMANDLINE_ARGS"] = "--no-gradio-queue"
 
 # Configure logging
 logging.basicConfig(
@@ -32,13 +28,21 @@ logging.basicConfig(
     format="%(asctime)s [%(levelname)s] %(message)s",
     handlers=[logging.StreamHandler()],
 )
-
 logger = logging.getLogger(__name__)
 
-
-import requests
+# 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 = {}
 
 
+# Function to request product ID from Tokopedia
 def request_product_id(shop_domain, product_key):
     ENDPOINT = "https://gql.tokopedia.com/graphql/PDPGetLayoutQuery"
     payload = {
@@ -81,30 +85,22 @@ def request_product_id(shop_domain, product_key):
                         }
                         """,
     }
-
     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",
     }
-
     try:
-        response = requests.request(
-            method="POST",
-            url=ENDPOINT,
-            json=payload,
-            headers=headers,
-            timeout=60
-        )
-        response.raise_for_status()  # Raise an exception for non-2xx status codes
+        response = requests.post(ENDPOINT, json=payload, headers=headers, timeout=60)
+        response.raise_for_status()
         logger.info(f"Request successful. Status code: {response.status_code}")
-        # Process the response data
+        return response
     except requests.exceptions.RequestException as e:
         logger.error(f"Request failed: {e}")
-    else:
-        return response
+        return None
 
 
+# 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 = {
@@ -168,40 +164,36 @@ def request_product_review(product_id, page=1, limit=20):
 }
                         """,
     }
-
     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.request(
-            method="POST", url=ENDPOINT, json=payload, headers=headers, timeout=60
-        )
-        response.raise_for_status()  # Raise an exception for non-2xx status codes
+        response = requests.post(ENDPOINT, json=payload, headers=headers, timeout=60)
+        response.raise_for_status()
         logger.info(f"Request successful. Status code: {response.status_code}")
-        # Process the response data
+        return response
     except requests.exceptions.RequestException as e:
         logger.error(f"Request failed: {e}")
-    else:
-        return response
+        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)
-    data = response.json()["data"]["productrevGetProductReviewList"]
-    reviews = data["list"]
-    all_reviews.extend(reviews)
-    has_next = data["hasNext"]
-    page += 1
-
+    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"]]
@@ -209,97 +201,82 @@ def scrape(product_id, max_reviews=LIMIT):
     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("/")
-    product_id = request_product_id(SHOP, PRODUCT_KEY).json()["data"]["pdpGetLayout"][
-        "basicInfo"
-    ]["id"]
-    logger.info(product_id)
-    return product_id
+    _, shop, product_key = parsed_url.path.strip("/").split("/")
+    response = request_product_id(shop, product_key)
+    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
 
 
-# Clean
+# 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")
+    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)  # 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
+    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
 
 
-# LLM
-OpenAIModel = "gpt-3.5-turbo"
+# Initialize LLM and embeddings
 llm = ChatOpenAI(model=OpenAIModel, temperature=0.1)
-
-# Embeddings
 embeddings = HuggingFaceEmbeddings(model_name="LazarusNLP/all-indobert-base-v2")
 
-cache_URL = ""
-db = None
-qa = None
-cache = {}
-
 
+# Function to generate a summary or answer based on reviews
 @spaces.GPU
 async def generate(URL, query):
     global cache_URL, db, qa, cache
 
-    if URL == "" or query == "":
+    if not URL or not query:
         return "Input kosong"
-    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
+    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]
 
-            # 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 "Gagal mendapatkan review dari 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"
 
 
-# Gradio
+# Set up Gradio interface
 product_box = gr.Textbox(label="URL Produk", placeholder="URL produk dari Tokopedia")
 query_box = gr.Textbox(
     lines=2,