Create app.py

app.py

@@ -0,0 +1,172 @@
+# file: app.py
+
+import gradio as gr
+import requests
+import json
+import concurrent.futures
+from concurrent.futures import ThreadPoolExecutor
+from langchain_community.document_loaders import PyPDFLoader
+from langdetect import detect_langs
+from PyPDF2 import PdfReader
+from io import BytesIO
+import logging
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+data = False 
+seen = set()
+
+main_url = "https://similar-products-api.vercel.app/search/all"
+main_product = "Samsung Galaxy"
+
+API_URL = "https://api-inference.huggingface.co/models/google/flan-t5-xxl"
+headers = {"Authorization": f"Bearer {os.getenv('HUGGINGFACE_API_TOKEN')}"}
+
+logging.basicConfig(level=logging.INFO)
+
+def get_links(product):
+    params = {
+        "API_KEY": "12345",
+        "product": f"{product}",
+    }
+    response = requests.get(main_url, params=params)
+    if response.status_code == 200:
+        results = response.json()
+        return results
+    else:
+        return {}
+
+def language_preprocess(text):
+    try:
+        if detect_langs(text)[0].lang == 'en':
+            return True
+        return False
+    except Exception as e:
+        logging.error(f"Language detection error: {e}")
+        return False
+
+def relevant(product, similar_product, content):
+    try:
+        payload = {"inputs": f'''Do you think that the given content is similar to {similar_product} and {product}, just Respond True or False  \nContent for similar product:  {content}'''}
+        response = requests.post(API_URL, headers=headers, json=payload)
+        output = response.json()
+        return bool(output[0]['generated_text'])
+    except Exception as e:
+        logging.error(f"Relevance checking error: {e}")
+        return False
+
+def download_pdf(url, timeout=10):
+    try:
+        response = requests.get(url, timeout=timeout)
+        response.raise_for_status()
+        return BytesIO(response.content)
+    except requests.RequestException as e:
+        logging.error(f"PDF download error: {e}")
+        return None
+
+def extract_text_from_pages(pdf_file, pages):
+    reader = PdfReader(pdf_file)
+    extracted_text = ""
+    try:
+        for page_num in pages:
+            if page_num < len(reader.pages):
+                page = reader.pages[page_num]
+                extracted_text += page.extract_text() + "\n"
+            else:
+                logging.warning(f"Page {page_num} does not exist in the document.")
+        return extracted_text
+    except Exception as e:
+        logging.error(f"PDF text extraction error: {e}")
+        return 'हे चालत नाही'
+
+def process_link(link, similar_product):
+    if link in seen:
+        return None
+    seen.add(link)
+    try:
+        pdf_file = download_pdf(link)
+        if pdf_file:
+            text = extract_text_from_pages(pdf_file, [0, 2, 4])
+            if language_preprocess(text):
+                if relevant(main_product, similar_product, text):
+                    return link
+    except Exception as e:
+        logging.error(f"Error processing link: {e}")
+    return None
+
+def filtering(urls, similar_product):
+    res = []
+    with ThreadPoolExecutor() as executor:
+        futures = {executor.submit(process_link, link, similar_product): link for link in urls}
+        for future in concurrent.futures.as_completed(futures):
+            result = future.result()
+            if result is not None:
+                res.append(result)
+    return res
+
+def wikipedia_url(product):
+    api_url = "https://en.wikipedia.org/w/api.php"
+    params = {
+        "action": "opensearch",
+        "search": product,
+        "limit": 5,
+        "namespace": 0,
+        "format": "json"
+    }
+    try:
+        response = requests.get(api_url, params=params)
+        response.raise_for_status()
+        data = response.json()
+        if data and len(data) > 3 and len(data[3]) > 0:
+            return data[3]
+        else:
+            return []
+    except requests.RequestException as e:
+        logging.error(f"Error fetching Wikipedia URLs: {e}")
+        return []
+
+def preprocess_initial(product):
+    return get_links(product)
+
+def preprocess_filter(product, data):
+    for similar_product in data:
+        if similar_product != product:
+            if list(data[similar_product][0])[0] == 'duckduckgo':
+                s = set(('duckduckgo', 'google', 'archive'))
+                temp = []
+
+                for idx, item in enumerate(data[similar_product]):
+                    if list(item)[0] in s:
+                        urls = data[similar_product][idx][list(item)[0]]
+                        temp += filtering(urls, similar_product)
+                    else:
+                        temp += data[similar_product][idx][list(item)[0]]
+
+                data[similar_product] = temp
+                data[similar_product] += wikipedia_url(similar_product)
+            else:
+                urls = data[similar_product]
+                data[similar_product] = filtering(urls, similar_product)
+                data[similar_product] += wikipedia_url(similar_product)
+    logging.info('Filtering completed')
+    return data
+
+def main(product_name):
+    return preprocess_initial(product_name)
+
+def filter_links(product_name, initial_data):
+    return preprocess_filter(product_name, initial_data)
+
+with gr.Blocks() as demo:
+    product_name = gr.Textbox(label="Product Name")
+    get_links_btn = gr.Button("Get Links")
+    initial_links_output = gr.JSON()
+    filter_btn = gr.Button("Filter Links")
+    filtered_links_output = gr.JSON()
+
+    get_links_btn.click(fn=main, inputs=product_name, outputs=initial_links_output)
+    filter_btn.click(fn=filter_links, inputs=[product_name, initial_links_output], outputs=filtered_links_output)
+
+if __name__ == "__main__":
+    demo.launch()