image_download_script.py

import asyncio
import csv
import os
from PIL import Image
from io import BytesIO
from pathlib import Path
import pandas as pd
import aiohttp

# Constants
CSV_FILE = "extracted_full_data.csv"
CONCURRENT_DOWNLOADS = 25
IMAGE_COLUMN = "Image"
CATEGORY_ID_COLUMN = "categoryID"
PRODUCT_ID_COLUMN = "productID"

# Directory to store images
BASE_DIR = Path("downloaded_images")
BASE_DIR.mkdir(exist_ok=True)

semaphore = asyncio.Semaphore(CONCURRENT_DOWNLOADS)

async def save_image_metadata(image_data, save_path):
    try:
        image = Image.open(BytesIO(image_data))
        image.save(save_path)
        print(f"[✓] Saved: {save_path} | Format: {image.format}, Size: {image.size}, Mode: {image.mode}")
    except Exception as e:
        print(f"[!] Metadata save error at {save_path}: {e}")

async def download_and_save_image(session, url, category_id, product_id):
    async with semaphore:
        try:
            async with session.get(url) as response:
                if response.status == 200:
                    image_data = await response.read()
                    file_ext = url.split('.')[-1].split("?")[0]
                    folder = BASE_DIR / str(category_id)
                    folder.mkdir(parents=True, exist_ok=True)
                    save_path = folder / f"{product_id}.{file_ext}"
                    await save_image_metadata(image_data, save_path)
                else:
                    print(f"[x] Failed: {url} - HTTP {response.status}")
        except Exception as e:
            print(f"[x] Download error for {url}: {e}")

async def run_tasks():
    df = pd.read_csv(CSV_FILE)
    tasks = []

    async with aiohttp.ClientSession() as session:
        for _, row in df.iterrows():
            url = row.get(IMAGE_COLUMN)
            category_id = row.get(CATEGORY_ID_COLUMN)
            product_id = row.get(PRODUCT_ID_COLUMN)

            if pd.notnull(url) and pd.notnull(category_id) and pd.notnull(product_id):
                tasks.append(download_and_save_image(session, url, category_id, product_id))

        await asyncio.gather(*tasks)

# Safe wrapper to run in environments with already running event loop (e.g. Jupyter)
def run_async_main():
    try:
        asyncio.run(run_tasks())
    except RuntimeError as e:
        if "cannot be called from a running event loop" in str(e):
            loop = asyncio.get_event_loop()
            task = loop.create_task(run_tasks())
            loop.run_until_complete(task)
        else:
            raise

# Execute
run_async_main()