| import gradio as gr |
| from playwright.sync_api import sync_playwright |
| import csv |
| import time |
| import random |
| import re |
| import pandas as pd |
| from io import StringIO |
|
|
| def clean_amazon_image(url): |
| """Remove ._AC_..._ part to get full-size image""" |
| if url: |
| return re.sub(r'\._AC_.*?\.', '.', url) |
| return url |
|
|
| def scrape_amazon(upc_list, min_delay=3, max_delay=6): |
| """Scrape Amazon for UPCs""" |
| upcs = [x.strip() for x in upc_list.split('\n') if x.strip()] |
| |
| if not upcs: |
| return "No UPCs provided!", None |
| |
| results = [] |
| |
| with sync_playwright() as p: |
| browser = p.chromium.launch(headless=True) |
| context = browser.new_context( |
| locale="en-US", |
| user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" |
| ) |
| page = context.new_page() |
| |
| progress_log = [] |
| |
| for i, upc in enumerate(upcs, start=1): |
| log_msg = f"Processing ({i}/{len(upcs)}): {upc}" |
| progress_log.append(log_msg) |
| |
| try: |
| page.goto(f"https://www.amazon.com/s?k={upc}", timeout=60000) |
| time.sleep(random.uniform(min_delay, max_delay)) |
| |
| item = page.query_selector("div[data-component-type='s-search-result']") |
| if not item: |
| progress_log.append(f"[NOT FOUND] {upc}") |
| continue |
| |
| asin = item.get_attribute("data-asin") |
| |
| title_el = item.query_selector("h2 span") |
| title = title_el.inner_text().strip() if title_el else "" |
| |
| img_el = item.query_selector("img.s-image") |
| image = img_el.get_attribute("src") if img_el else "" |
| image = clean_amazon_image(image) |
| |
| link_el = item.query_selector("h2 a") |
| link = "https://www.amazon.com" + link_el.get_attribute("href") if link_el else "" |
| |
| results.append({ |
| "SKU": upc, |
| "ASIN": asin, |
| "Title": title, |
| "Image": image, |
| "Amazon URL": link |
| }) |
| |
| progress_log.append(f"✓ Found: {title[:60]}") |
| |
| except Exception as e: |
| progress_log.append(f"[ERROR] {upc}: {str(e)}") |
| |
| browser.close() |
| |
| |
| if results: |
| df = pd.DataFrame(results) |
| csv_output = df.to_csv(index=False) |
| return "\n".join(progress_log), csv_output |
| else: |
| return "\n".join(progress_log), None |
|
|
| |
| with gr.Blocks(title="Amazon UPC Scraper") as demo: |
| gr.Markdown("# 🛒 Amazon UPC Scraper") |
| gr.Markdown("Enter UPCs (one per line) to scrape Amazon product data") |
| |
| with gr.Row(): |
| with gr.Column(): |
| upc_input = gr.Textbox( |
| label="UPC Codes (one per line)", |
| placeholder="Enter UPCs here...\n123456789\n987654321", |
| lines=10 |
| ) |
| |
| with gr.Row(): |
| min_delay = gr.Slider(1, 10, value=3, label="Min Delay (seconds)") |
| max_delay = gr.Slider(1, 15, value=6, label="Max Delay (seconds)") |
| |
| scrape_btn = gr.Button("🔍 Scrape Amazon", variant="primary") |
| |
| with gr.Column(): |
| log_output = gr.Textbox( |
| label="Progress Log", |
| lines=15, |
| interactive=False |
| ) |
| |
| csv_output = gr.File( |
| label="Download Results (CSV)", |
| interactive=False |
| ) |
| |
| scrape_btn.click( |
| fn=scrape_amazon, |
| inputs=[upc_input, min_delay, max_delay], |
| outputs=[log_output, csv_output] |
| ) |
| |
| gr.Markdown(""" |
| ### ⚠️ Important Notes: |
| - This scraper respects delays to avoid overwhelming Amazon servers |
| - Large batches may take time to process |
| - Results are provided as-is; verify data accuracy |
| """) |
|
|
| if __name__ == "__main__": |
| demo.launch() |
