Upload 8 files

API_main.py

@@ -0,0 +1,49 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from price_analysis import market_price_estimation
+
+app = FastAPI(title="Market Price Analysis API")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+class MarketEstimation(BaseModel):
+    product_name: str
+    cost_price: int
+    user_price: int
+
+
+@app.get("/")
+async def root():
+    return {
+        "message": "Welcome to the Market Prices Estimation API!",
+        "version": "1.0",
+        "endpoints": {
+            "/": "This welcome message",
+            "/market-prices-estimation/": "POST endpoint for price analysis"
+        }
+    }
+
+
+@app.post("/market-prices-estimation/")
+async def market_prices_estimation_endpoint(request: MarketEstimation):
+    try:
+        response = market_price_estimation(request.product_name, request.cost_price, request.user_price)
+
+        if not isinstance(response, dict):
+            raise ValueError("market_price_estimation must return a dictionary")
+
+        return JSONResponse(status_code=200, content=response)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

Amazon_scraper.py

@@ -0,0 +1,122 @@
+import requests
+from bs4 import BeautifulSoup
+import re
+import random
+import time
+from difflib import SequenceMatcher
+
+
+def extract_prefix_and_number(text):
+    match = re.search(r'([A-Za-z]+)(\d+)', text)
+    if match:
+        return match.group(1), match.group(2)
+    return None, None  # No valid match
+
+
+def similarity(a, b):
+    a_lower, b_lower = a.lower(), b.lower()
+    a_prefix, a_number = extract_prefix_and_number(a_lower)
+    b_prefix, b_number = extract_prefix_and_number(b_lower)
+    if not a_prefix or not b_prefix or not a_number or not b_number:
+        return 0
+    if (a_prefix != b_prefix) or (a_number != b_number):
+        return 0
+    if a_number not in b_lower:
+        return 0
+    base_similarity = SequenceMatcher(None, a_lower, b_lower).ratio()
+    return base_similarity
+
+
+def parse_amazon_page(content, product_name, your_cost):
+    soup = BeautifulSoup(content, 'html.parser')
+    price_digit_limit = len(f"{your_cost}")
+    product_prices = []
+    products = soup.findAll("div", attrs={"data-component-type": "s-search-result"})
+
+    for product in products[:20]:
+        title = product.find("h2", attrs={"class": "a-size-base-plus"})
+        if not title:
+            continue
+
+        spans = title.findAll("span")
+        for span in spans:
+            name = span.text.strip()
+            similarity_score = similarity(product_name, name)
+            if similarity_score >= 0.0:
+                # Get product link
+                product_link = ""
+                link_tag = title.find_parent("a")
+                if link_tag and 'href' in link_tag.attrs:
+                    product_link = "https://www.amazon.eg" + link_tag['href']
+
+                # Get image link
+                image_link = ""
+                img_tag = product.find("img", attrs={"class": "s-image"})
+                if img_tag and 'src' in img_tag.attrs:
+                    image_link = img_tag['src']
+
+                price_tag = product.find("span", attrs={"class": "a-price-whole"})
+                if price_tag:
+                    raw_price = price_tag.text.strip()
+                    numeric_price = re.sub(r"[^\d]", "", raw_price)
+
+                    if not numeric_price:
+                        continue
+
+                    integer_part = numeric_price.split('.')[0]
+                    if ((len(integer_part) == price_digit_limit) or (len(integer_part) == price_digit_limit + 1)) and (
+                            int(integer_part) > int(your_cost)):
+                        product_prices.append((name, numeric_price, product_link, image_link))
+
+    if not product_prices:
+        print("Warning: No valid prices found on Amazon.")
+
+    return product_prices
+
+
+def scrape_amazon(product_name, your_cost, queue, max_retries=3, retry_delay=3):
+    url = f"https://www.amazon.eg/s?k={product_name.replace(' ', '+')}&language=en"
+    print(f"Fetching: {url}")
+
+    headers = {
+        "User-Agent": random.choice([
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.5481.77 Safari/537.36",
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36",
+            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"
+        ]),
+        "Accept-Language": "en-US,en;q=0.9",
+        "Referer": "https://www.google.com/",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+        "Connection": "keep-alive",
+    }
+
+    for attempt in range(max_retries):
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+
+            if response.status_code in [506, 503]:
+                print(
+                    f"Error {response.status_code}. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
+                time.sleep(retry_delay)
+                continue
+
+            if response.status_code == 200:
+                print("Page fetched successfully with status code: 200")
+
+                results = parse_amazon_page(response.content, product_name, your_cost)
+                queue.put(("amazon", results))
+
+                print("Amazon results sent to queue")  # Fix: Now this line runs
+                return results  # Fix: Ensures function exits properly
+
+            else:
+                print(f"Unexpected status code: {response.status_code}")
+                queue.put([])
+
+        except requests.exceptions.RequestException as e:
+            print(f"An error occurred: {e}. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
+            time.sleep(retry_delay)
+
+    print("Failed to fetch Amazon data after multiple attempts.")
+    queue.put([])  # Ensure the queue gets an empty list if all retries fail

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.9-slim
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+COPY . /code/
+
+EXPOSE 7860
+
+CMD ["uvicorn", "API_main:app", "--host", "0.0.0.0", "--port", "7860"]

Jumia_scraper.py

@@ -0,0 +1,85 @@
+import requests
+from bs4 import BeautifulSoup
+import re
+import random
+import time
+from Amazon_scraper import similarity
+import multiprocessing
+
+
+def parse_jumia_page(content, product_name, your_cost):
+    soup = BeautifulSoup(content, 'html.parser')
+    price_digit_limit = len(f"{your_cost}")
+    product_prices = []
+    articles = soup.findAll("article", attrs={"class": "prd"})
+
+    for article in articles[:20]:
+        title = article.find("h3", attrs={"class": "name"})
+        if not title:
+            continue
+
+        name = title.text.strip()
+        similarity_score = similarity(product_name, name)
+        if similarity_score >= 0.0:
+            # Get product link
+            product_link = ""
+            link_tag = article.find("a")
+            if link_tag and 'href' in link_tag.attrs:
+                product_link = link_tag['href']
+
+            # Get image link with correct class name
+            image_link = ""
+            img_tag = article.find("img", attrs={"class": "img-c"})
+            if img_tag and 'data-src' in img_tag.attrs:
+                image_link = img_tag['data-src']
+
+            price_tag = article.find("div", attrs={"class": "prc"})
+            if price_tag:
+                raw_price = price_tag.text.strip()
+                numeric_price = re.sub(r"[^\d.]", "", raw_price)
+                numeric_price = numeric_price.split(".")[0]
+                if price_digit_limit:
+                    if ((len(numeric_price) == price_digit_limit) or (
+                            len(numeric_price) == price_digit_limit + 1)) and (int(numeric_price) > int(your_cost)):
+                        product_prices.append((name, numeric_price, product_link, image_link))
+                    continue
+
+    return product_prices
+
+def scrape_jumia(product_name, your_cost, queue, max_retries=5, retry_delay=5):
+    url = f"https://www.jumia.com.eg/catalog/?q={product_name.replace(' ', '+')}"
+    # print(url)
+    headers = {
+        "User-Agent": random.choice([
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.5481.77 Safari/537.36",
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36",
+            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"
+        ]),
+        "Accept-Language": "en-US,en;q=0.9"
+    }
+
+    for attempt in range(max_retries):
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            if response.status_code in [506, 503]:
+                print(
+                    f"Error {response.status_code}. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
+                time.sleep(retry_delay)
+                continue
+            if response.status_code == 200:
+                print(f"Page fetched successfully with status code: {response.status_code}")
+                # time.sleep(1.5)
+                results = parse_jumia_page(response.content, product_name, your_cost)
+                queue.put(("jumia", results))
+                print("jumia results sent to queue")
+                return results
+            else:
+                print(f"Unexpected status code: {response.status_code}")
+                queue.put([])
+        except requests.exceptions.RequestException as e:
+            print(f"An error occurred: {e}. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
+            time.sleep(retry_delay)
+            queue.put([])  # Empty list after retries failed
+
+
+

main.py

@@ -0,0 +1,40 @@
+from parallel_execution import scrape_product_multiprocessing
+from Amazon_scraper import scrape_amazon
+from Jumia_scraper import scrape_jumia
+import numpy as np
+from price_analysis import market_price_estimation,get_products_list
+from Amazon_scraper import scrape_amazon
+from PIL import Image  # To open the saved image for preview
+
+if __name__ == '__main__':
+    product_name = 'soundcore r50 nc'
+    cost_price = 1000
+    user_price = 1500
+
+    # Get prices from web scrapers
+    response = market_price_estimation(product_name, cost_price, user_price)
+    print(response)
+    # print("Scraped Prices:", prices)
+    # min_price, avg_price, max_price = get_MinMaxAverage(prices)
+    # Generate image
+    # image_buffer = plot_your_price(user_price, min_price,max_price,avg_price)
+
+    # if image_buffer is None:
+    #     print("Error: Could not generate price comparison plot.")
+    # else:
+    #     # Save to a file
+    #     with open("price_comparison.png", "wb") as f:
+    #         f.write(image_buffer.getvalue())
+    #
+    #     print("Image saved as 'price_comparison.png'")
+    #
+    #     # Open image for preview (optional)
+    #     img = Image.open("price_comparison.png")
+    #     img.show()
+
+    #
+    # recommendations = recommend_price(min_price,avg_price,max_price,user_price, cost_price,prices)
+    # for key, value in recommendations.items():
+    #     print(f"{key}: {value}")
+    # ans = get_prices_analysis(prices,cost_price,user_price)
+    # print(ans)

parallel_execution.py

@@ -0,0 +1,43 @@
+import multiprocessing
+from  Jumia_scraper import scrape_jumia
+from Amazon_scraper import scrape_amazon
+import time
+
+def scrape_product_multiprocessing(product_name,your_cost):
+    queue = multiprocessing.Queue()
+
+    # Create processes
+    p1 = multiprocessing.Process(target=scrape_amazon, args=(product_name, your_cost, queue))
+    p2 = multiprocessing.Process(target=scrape_jumia, args=(product_name, your_cost, queue))
+
+    # Start processes
+    p1.start()
+    p2.start()
+
+    # Wait for processes to complete
+    p1.join()
+    p2.join()
+
+    # Debugging: Check the queue size after processes finish
+    print(f"Queue size after both processes finish: {queue.qsize()}")
+
+    # Retrieve results from queue
+    results_amazon = []
+    results_jumia = []
+
+    # Check if queue has results
+    while not queue.empty():
+        try:
+
+           site, results = queue.get()
+        # print(f"Results from {site}: {results}")  # Debugging output
+           if site == 'amazon':
+              results_amazon = results
+           elif site == 'jumia':
+              results_jumia = results
+
+        except Exception as e:
+            print(e)
+
+    # Return results and total time
+    return results_amazon, results_jumia

price_analysis.py

@@ -0,0 +1,267 @@
+from parallel_execution import scrape_product_multiprocessing
+import numpy as np
+import matplotlib.pyplot as plt
+import io
+from typing import List, Dict, Any
+import json
+import re
+
+prices_list = []
+# min_price=0
+# avg_price=0
+# max_price = 0
+from typing import List, Any, Tuple
+
+
+def loop_prices(source_list: List[List[Any]], source_name: str) -> Tuple[List[dict], List[int]]:
+    products = []
+    prices_list = []
+
+    for item in source_list:
+        try:
+            product_name = item[0]
+            price = int(item[1])
+            product_link = item[2] if len(item) > 2 else ""
+            image_link = item[3] if len(item) > 3 else ""
+
+            product = {
+                "product_name": product_name,
+                "source": source_name,
+                "price": price,
+                "product link": product_link,
+                "image link": image_link
+            }
+            products.append(product)
+            prices_list.append(price)
+        except (ValueError, IndexError, TypeError):
+            print(f"Error: Skipping invalid price data from {source_name}")
+
+    return products, prices_list
+
+def get_products_list(product_name: str, your_cost: float):
+    amazon, jumia = scrape_product_multiprocessing(product_name, your_cost)
+
+    products = []
+    all_prices = []
+
+    amazon_products, amazon_prices = loop_prices(amazon, "Amazon")
+    jumia_products, jumia_prices = loop_prices(jumia, "Jumia")
+
+    products.extend(amazon_products)
+    products.extend(jumia_products)
+
+    all_prices.extend(amazon_prices)
+    all_prices.extend(jumia_prices)
+
+    return products, all_prices
+
+# def generate_prompt(product_name, your_cost):
+#     products, _ = get_products_list(product_name, your_cost)
+#
+#     prompt = (f"Filter the given product list to include only closely related products."
+#               f" The response should only be the filtered product list in valid JSON format, without any explanations or additional text."
+#               f" Ensure the output is a properly formatted JSON array of dictionaries."
+#               f" Real product name: {product_name}\nProduct list: {json.dumps(products, ensure_ascii=False, indent=2)}\n"
+#               f"Example JSON output:\n"
+#               f"["
+#               f"{{\"product_name\": \"Soundcore R50i NC Wireless Bluetooth Headphones - Black\", \"source\": \"Amazon\", \"price\": 1150}},"
+#               f"{{\"product_name\": \"Soundcore R50i NC True Wireless Earbuds 10mm Drivers with Big Bass, Bluetooth 5.3, 45H Playtime, IP54-Sweatguard Waterproof, AI Clear Calls with 4 Mics, 22 Preset EQs via App-Black\", \"source\": \"Amazon\", \"price\": 1390}},"
+#               f"{{\"product_name\": \"Soundcore R50i NC True Wireless Earbuds 10mm Drivers with Big Bass, Bluetooth 5.3, 45H Playtime, IP54-Sweatguard Waterproof, AI Clear Calls with 4 Mics, 22 Preset EQs via App-White\", \"source\": \"Amazon\", \"price\": 1713}},"
+#               f"{{\"product_name\": \"Soundcore R50i NC True Wireless Earbuds with Big Bass, Bluetooth 5.3, 45H Playtime, IP54-Sweatguard Waterproof, AI Clear Calls with 4 Mics, 22 Preset EQs via App-GREEN Local warranty\", \"source\": \"Amazon\", \"price\": 1550}}"
+#               f"]"
+#               )
+#
+#     return prompt
+
+# def get_filtered_product_list(product_name, your_cost):
+#     prompt = generate_prompt(product_name, your_cost)
+#
+#     genai.configure(api_key="AIzaSyAzp-WRPAi4IaALmpjyRh2yo0qsPmFMxdI")
+#     model = genai.GenerativeModel("gemini-2.0-flash")
+#     response = model.generate_content(prompt)
+#
+#     try:
+#         # Parse response into JSON
+#         filtered_products = json.loads(response.text)
+#         if isinstance(filtered_products, list):
+#             return filtered_products
+#         else:
+#             return []
+#     except json.JSONDecodeError:
+#         return []
+
+
+# def extract_json(response_text):
+#     match = re.search(r"\[.*\]", response_text, re.DOTALL)  # Extracts JSON part
+#     if match:
+#         return match.group(0)
+#     return None
+#
+# def extract_prices(response_text):
+#     json_text = extract_json(response_text)
+#     if not json_text:
+#         return "Invalid API response: No JSON found"
+#
+#     try:
+#         products = json.loads(json_text)  # Convert JSON string to Python list
+#         prices = [product["price"] for product in products if "price" in product]
+#         return prices
+#     except json.JSONDecodeError as e:
+#         return f"Invalid JSON format: {e}"
+
+def remove_outliers(prices, multiplier=1.0):
+    if not prices:
+        print("Warning: The prices list is empty. Returning an empty list.")
+        return []
+    try:
+        prices = list(map(int, prices))
+    except ValueError:
+        print("Error: Could not convert prices to integers. Check data format.")
+        return []
+
+    if len(prices) < 2:
+        print("Warning: Not enough data points to compute outliers.")
+        return prices
+
+    Q1 = np.percentile(prices, 25)
+    Q3 = np.percentile(prices, 75)
+    IQR = Q3 - Q1
+    lower_bound = Q1 - multiplier * IQR
+    upper_bound = Q3 + multiplier * IQR
+    return [price for price in prices if lower_bound <= price <= upper_bound]
+
+
+def get_MinMaxAverage(updated_price_list):
+    filtered_prices = remove_outliers(updated_price_list)
+
+    if not filtered_prices:
+        print("Error: No valid prices available.")
+        return None, None, None
+
+    try:
+        filtered_prices = list(map(int, filtered_prices))
+        mini = np.min(list(map(int, updated_price_list)))
+        maxi = np.max(list(map(int, updated_price_list)))
+        average = round(np.mean(filtered_prices), 2)
+    except ValueError:
+        print("Error: Could not compute min/max/average due to invalid data.")
+        return None, None, None
+
+    return mini, average, maxi
+
+def normalize(price, min_price, max_price):
+    if min_price is None or max_price is None:
+        print("Error: Cannot normalize due to missing price data.")
+        return np.pi / 2
+    if min_price == max_price:
+        return np.pi / 2
+    return np.pi - ((price - min_price) / (max_price - min_price) * np.pi)
+
+
+
+def plot_your_price(your_price, min_price,max_price,avg_price):
+
+
+    if min_price is None or max_price is None or avg_price is None:
+        print("Error: Cannot plot due to missing price data.")
+        return None  # Return None if the image cannot be generated
+
+    fig, ax = plt.subplots(figsize=(8, 4), subplot_kw={'projection': 'polar'})
+
+    # Create the three segments (Min = Red, Mid = Yellow, Max = Green)
+    ax.barh(1, np.pi / 3, left=2 * np.pi / 3, color='red', height=0.5)
+    ax.barh(1, np.pi / 3, left=np.pi / 3, color='yellow', height=0.5)
+    ax.barh(1, np.pi / 3, left=0, color='green', height=0.5)
+
+    # Plot your price marker
+    norm_price = normalize(your_price, min_price, max_price)
+    ax.plot([norm_price, norm_price], [0, 1], color="black", linewidth=3, marker="o", markersize=10)
+
+    # Labels
+    ax.text(np.pi + 0.1, 1.2, f"Min: {int(min_price)}", ha="center", fontsize=10, color="black", fontweight="bold")
+    ax.text(np.pi / 2, 1.2, f"Avg: {int(avg_price)}", ha="center", fontsize=10, color="black", fontweight="bold")
+    ax.text(-0.1, 1.2, f"Max: {int(max_price)}", ha="center", fontsize=10, color="black", fontweight="bold")
+
+    # Final styling
+    plt.title("Your Price Compared to Market", fontsize=12, fontweight="bold", color="black")
+    ax.set_xticks([])
+    ax.set_yticks([])
+    ax.set_frame_on(False)
+    plt.show()
+    # Save the plot to an in-memory buffer
+    img_buffer = io.BytesIO()
+    plt.savefig(img_buffer, format="png", bbox_inches="tight")  # Save the figure to the buffer
+    plt.close(fig)
+    img_buffer.seek(0)
+
+    return img_buffer
+
+def recommend_price(min_price, avg_price, max_price, user_price, user_cost, price_list):
+    # Calculate quartiles
+    q1 = float(np.percentile(price_list, 25))
+    q3 = float(np.percentile(price_list, 75))
+
+    # Price Adjustment Suggestion based on market range
+    if user_price < min_price:
+        price_suggestion = f"Your price is too low. Consider increasing it to at least {min_price}."
+    elif user_price > max_price:
+        price_suggestion = f"Your price is too high. Consider lowering it below {max_price}."
+    else:
+        price_suggestion = "Your price is competitive in the market."
+
+    # Market Competitiveness Rating
+    if user_price < q1:
+        competitiveness = "Very Cheap (Consider increasing your price!)"
+    elif q1 <= user_price <= q3:
+        competitiveness = "Competitive (Good price in the market)"
+    else:
+        competitiveness = "Expensive (Consider lowering your price)"
+
+    # Recommended Selling Price Range (ensuring at least 10% profit)
+    recommended_price = max(user_cost * 1.1, q1)  # Ensure minimum profit of 10%
+    recommended_range = (round(recommended_price, 2), round(q3, 2))
+
+    # Relationship between User Price and Average Price
+    if user_price < avg_price:
+        avg_relation = f"Your price is below the average market price ({avg_price}). You may have room to increase it."
+    elif user_price > avg_price:
+        avg_relation = f"Your price is above the average market price ({avg_price}). Ensure your product quality justifies the price."
+    else:
+        avg_relation = "Your price matches the average market price."
+
+    # Profit Calculation
+    profit_margin = user_price - user_cost
+    profit_percentage = (profit_margin / user_cost) * 100 if user_cost > 0 else 0
+
+    return {
+        "min_price": min_price,
+        "max_price": max_price,
+        "avg_price": avg_price,
+        "user_price": user_price,
+        "price_suggestion": price_suggestion,
+        "competitiveness": competitiveness,
+        "recommended_range": recommended_range,
+        "avg_relation": avg_relation,
+        "profit_margin": f"{round(profit_margin, 2)} EGP",
+        "profit_percentage": f"{round(profit_percentage, 2)}%"
+    }
+
+
+def get_prices_analysis(prices, cost_price, user_price):
+    prices = [float(p) for p in prices]
+    min_price, avg_price, max_price = get_MinMaxAverage(prices)
+    min_price, avg_price, max_price = int(min_price), float(avg_price), int(max_price)
+    # image_buffer = plot_your_price(user_price, min_price, max_price, avg_price)
+    recommendations = recommend_price(min_price, avg_price, max_price, user_price, cost_price, prices)
+    recommendations["recommended_range"] = tuple(map(float, recommendations["recommended_range"]))
+    return recommendations
+
+
+def market_price_estimation(product_name , cost_price , user_price):
+    products,prices  = get_products_list(product_name,cost_price)
+    recommendations = get_prices_analysis(prices, cost_price, user_price)
+    response={
+        "products": products,
+        "recommendations": recommendations,
+    }
+    return response