How to Build a Labubu Bot with Residential Proxies: Step-by-Step Guide

pip install playwright
python -m playwright install

pip install apscheduler

cd path/to/your/project

popmart-bot (folder)
    - data (folder)
         - products.json
    - job-scheduler.py
    - main.py 
    - popmart-scraper.py
    - purchase-bot.py

import subprocess
import time
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta

# Maximum number of retries for scraper
MAX_RETRIES = 5
RETRY_DELAY = 10

# Scheduled time for daily scraper run
HOUR = 6
MINUTE = 0
scheduler = BlockingScheduler()

def run_daily_scraper():
    # This function runs the popmart-scraper.py script and schedules job-scheduler.py to run shortly after.
    print(f"\nRunning popmart-scraper at {datetime.now().strftime('%H:%M:%S')}")
    for attempt in range(1, MAX_RETRIES + 1):
        print(f"Attempt {attempt} to run scraper...")

        try:
            subprocess.run(["python3", "popmart-scraper.py"], check=True)
            print("New arrival scraper launched successfully.")
            
            # Schedule job-scheduler to run shortly after
            run_time = datetime.now() + timedelta(seconds=5)
            scheduler.add_job(run_job_scheduler, trigger='date', run_date=run_time)
            print(f"The job-scheduler.py will run at {run_time.strftime('%H:%M:%S')}")
            return  # Exit early on success

        except subprocess.CalledProcessError as e:
            print(f"Scraper failed (attempt {attempt}) with exit code {e.returncode}")
            if attempt < MAX_RETRIES:
                print(f"Retrying in {RETRY_DELAY} seconds...")
                time.sleep(RETRY_DELAY)

    print("All attempts to run the scraper failed. Check popmart-scraper.py for issues.")

def run_job_scheduler():
    print(f"\nRunning job-scheduler.py")
    try:
        subprocess.run(["python3", "job-scheduler.py"], check=True)
    except subprocess.CalledProcessError as e:
        print(f"Job scheduler failed with exit code {e.returncode}")
        print("Please check job-scheduler.py for issues.")

if __name__ == "__main__":
    print("main.py started...")
    run_daily_scraper()  # run once immediately on startup

    # Schedule scraper to run daily at configured time
    scheduler.add_job(run_daily_scraper, 'cron', hour=HOUR, minute=MINUTE)
    print(f"Daily scraper has been scheduled to run at {HOUR:02d}:{MINUTE:02d} every day.")
    
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("Scheduler stopped.")

import asyncio
import json
import os
from playwright.async_api import async_playwright
import sys

TARGET_KEYWORDS = ["THE MONSTERS", "Labubu"]
BASE_URL = "https://www.popmart.com"
OUTPUT_FILE = os.path.join("data", "products.json")

# Proxy config (replace with your credentials)
PROXY_SERVER = "http://us.decodo.com:10001"
PROXY_USERNAME = "username"
PROXY_PASSWORD = "password"

async def scrape_popmart():
    print("New arrivals scraping started...")

    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=True,
                proxy={"server": PROXY_SERVER}
                )
            
            context = await browser.new_context(
                proxy={
                    "server": PROXY_SERVER,
                    "username": PROXY_USERNAME,
                    "password": PROXY_PASSWORD
                }
            )
            page = await context.new_page()
            await page.goto("https://www.popmart.com/us/new-arrivals", timeout=30000)
            await page.wait_for_selector("div.index_title__jgc2z")

            # Try to close location popup if present
            try:
                await page.wait_for_selector("div.index_siteCountry___tWaj", timeout=15000)
                popup_selector = "div.index_siteCountry___tWaj"
                # Wait briefly (2 seconds) for popup to appear without failing if it doesn't
                await page.wait_for_selector(popup_selector, timeout=2000)
                await page.click(popup_selector)
                print("Closed location pop-up.")
            except Exception:
                # Popup not present -- continue normally
                print("No location pop-up detected.")

            # Close policy acceptance pop-up if present (after country pop-up)
            try:
                policy_selector = "div.policy_acceptBtn__ZNU71"

                # Wait until it's visible
                await page.wait_for_selector(policy_selector, timeout=8000, state="visible")

                # Get the element
                policy_btn = await page.query_selector(policy_selector)

                if policy_btn:
                    await asyncio.sleep(1)  # slight buffer for JS readiness
                    await policy_btn.click()
                    print("Clicked policy ACCEPT div.")
                else:
                    print("Could not find the policy ACCEPT div.")
            except Exception as e:
                print(f"Policy ACCEPT pop-up not detected or failed to click: {e}")

            results = []

            sections = await page.query_selector_all("div.index_title__jgc2z")

            for section in sections:
                release_date = (await section.text_content()).strip()

                # Get sibling product list container
                sibling = await section.evaluate_handle("el => el.nextElementSibling")
                product_cards = await sibling.query_selector_all("div.index_productCardCalendarContainer__B96oH")

                for card in product_cards:
                    # Product title
                    title_elem = await card.query_selector("div.index_title__9DEwH span")
                    title = await title_elem.text_content() if title_elem else ""
                    if not any(keyword.lower() in title.lower() for keyword in TARGET_KEYWORDS):
                        continue

                    # Release time
                    time_elem = await card.query_selector("div.index_time__EyE6b")
                    time_text = await time_elem.text_content() if time_elem else "N/A"

                    # Product URL
                    a_elem = await card.query_selector("a[href^='/us']")
                    href = await a_elem.get_attribute("href") if a_elem else None
                    full_url = f"{BASE_URL}{href}" if href else "N/A"

                    # Build entry
                    result = {
                        "title": title.strip(),
                        "release_date": release_date.strip(),  # Raw text like "Upcoming JUL 11"
                        "release_time": time_text.strip(),     # Raw text like "09:00"
                        "url": full_url
                    }
                    results.append(result)

            await browser.close()

            # Save to JSON
            os.makedirs("data", exist_ok=True)
            with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
                json.dump(results, f, indent=2, ensure_ascii=False)

            print(f"Scraped {len(results)} matching products. Saved to {OUTPUT_FILE}")
    except Exception as e:
        print(f"Error during scraping: {e}")
        sys.exit(1)  # Exit with error code 1 on failure


if __name__ == "__main__":
    asyncio.run(scrape_popmart())

import json
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
import subprocess
import os
import time

DATA_FILE = os.path.join("data", "products.json")
MAX_RETRIES = 5
RETRY_DELAY = 10

def parse_release_datetime(date_str, time_str):
    # Convert strings like "Upcoming JUL 11" and "09:00" into a datetime object. Assumes the current year.
    try:
        # Remove unwanted keywords
        for keyword in ["Upcoming", "In Stock"]:
            date_str = date_str.replace(keyword, "").strip()
        
        full_date_str = f"{date_str} {datetime.now().year} {time_str}"
        # Example: "JUL 11 2025 09:00"
        return datetime.strptime(full_date_str, "%b %d %Y %H:%M")
    except Exception as e:
        print(f"Failed to parse datetime from '{date_str} {time_str}': {e}")
        return None

def launch_purchase_bot(product):
    # Launch purchase-bot.py with retry logic
    url = product.get("url")
    title = product.get("title")
    
    for attempt in range(MAX_RETRIES + 1):  # +1 for initial attempt
        print(f"Launching purchase bot for '{title}' (attempt {attempt + 1}/{MAX_RETRIES + 1}) at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        try:
            # Run the purchase bot and wait for it to complete
            result = subprocess.run(
                ["python3", "purchase-bot.py", url],
                capture_output=True,
                text=True,
                timeout=300  # 5 minute timeout
            )
            
            if result.returncode == 0:
                print(f"✅ Purchase bot succeeded for '{title}' on attempt {attempt + 1}")
                return  # Success - exit the retry loop
            else:
                print(f"Purchase bot failed for '{title}' on attempt {attempt + 1}")
                print(f"Return code: {result.returncode}")
                print(f"STDOUT: {result.stdout}")
                print(f"STDERR: {result.stderr}")
                
        except subprocess.TimeoutExpired:
            print(f"⏰ Purchase bot timed out for '{title}' on attempt {attempt + 1}")
        except Exception as e:
            print(f"💥 Exception running purchase bot for '{title}' on attempt {attempt + 1}: {e}")
        
        # If this wasn't the last attempt, wait before retrying
        if attempt < MAX_RETRIES:
            print(f"⏳ Waiting {RETRY_DELAY} seconds before retry...")
            time.sleep(RETRY_DELAY)
    
    print(f"All {MAX_RETRIES + 1} attempts failed for '{title}'.")

def schedule_all_jobs_from_json(json_path):
    scheduler = BackgroundScheduler()
    job_count = 0

    with open(json_path, "r", encoding="utf-8") as f:
        products = json.load(f)

    for product in products:
        run_time = parse_release_datetime(product["release_date"], product["release_time"])
        if not run_time:
            continue
        if run_time < datetime.now():
            continue
        
        scheduler.add_job(launch_purchase_bot, "date", run_date=run_time, args=[product])
        print(f"🧸 Scheduled '{product['title']}' for {run_time}")
        job_count += 1

    if job_count == 0:
        print("No upcoming valid jobs found in JSON. Nothing scheduled.")
        return

    scheduler.start()
    print("Scheduler started. Jobs will run at their scheduled times.")

    try:
        # Keep the scheduler alive
        while True:
            pass
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("Scheduler stopped.")

if __name__ == "__main__":
    schedule_all_jobs_from_json(DATA_FILE)

import sys
import asyncio
from playwright.async_api import async_playwright

# Proxy config (replace with your credentials)
PROXY_SERVER = "http://us.decodo.com:10001"
PROXY_USERNAME = "username"
PROXY_PASSWORD = "password"

async def run(url):
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=False, # Visible browser for purchase
                proxy={"server": PROXY_SERVER}
                ) 
            context = await browser.new_context( # Create a new incognito context
                proxy={
                    "server": PROXY_SERVER,
                    "username": PROXY_USERNAME,
                    "password": PROXY_PASSWORD
                }
            )  
            page = await browser.new_page()
            await page.goto(url)

            # Try to close location popup if present
            try:
                await page.wait_for_selector("div.index_siteCountry___tWaj", timeout=15000)
                popup_selector = "div.index_siteCountry___tWaj"
                # Wait briefly (2 seconds) for popup to appear without failing if it doesn't
                await page.wait_for_selector(popup_selector, timeout=2000)
                await page.click(popup_selector)
                print("Closed location pop-up.")
            except Exception:
                # Popup not present -- continue normally
                print("No location pop-up detected.")

            # Close policy acceptance pop-up if present (after country pop-up)
            try:
                policy_selector = "div.policy_acceptBtn__ZNU71"

                # Wait until it's visible
                await page.wait_for_selector(policy_selector, timeout=8000, state="visible")

                # Get the element
                policy_btn = await page.query_selector(policy_selector)

                if policy_btn:
                    await asyncio.sleep(1)  # slight buffer for JS readiness
                    await policy_btn.click()
                    print("Clicked policy ACCEPT div.")
                else:
                    print("Could not find the policy ACCEPT div.")
            except Exception as e:
                print(f"Policy ACCEPT pop-up not detected or failed to click: {e}")

            
            # Wait for ADD TO BAG button and click it
            add_to_bag_selector = "div.index_usBtn__2KlEx.index_red__kx6Ql.index_btnFull__F7k90"
            
            # Wait and click button safely
            try:
                await page.wait_for_selector(add_to_bag_selector, timeout=15000)  # 15 seconds timeout
                await page.click(add_to_bag_selector)
                print("Clicked 'ADD TO BAG' button.")
            except Exception as e:
                print(f"Failed to find or click 'ADD TO BAG' button: {e}")
                await browser.close()
                return 1  # Return error code
            
            await asyncio.sleep(3)  # Give it time to process

            # Go to the shopping cart page
            try:
                await page.goto("https://www.popmart.com/us/largeShoppingCart")
                print("Navigated to shopping cart.")
                # Click the checkbox to select all items
                await page.click("div.index_checkbox__w_166")
                # Keep the browser open to allow manual checkout
                print("Browser will stay open for manual checkout. Close it when done.")
                #await asyncio.Future()  # Keeps script running indefinitely until manually closed
                await page.wait_for_event("close", timeout=0)  # Wait until user closes the visible browser tab
            except Exception as e:
                print(f"Failed during checkout preparation: {e}")
                return 1  # Return error code
            finally:
                await context.close() # Clean up incognito session
                await browser.close() # Fully shut down Playwright
            
            return 0  # Success
            
    except Exception as e:
        print(f"Fatal error in purchase bot: {e}")
        return 1  # Return error code

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 purchase-bot.py <product_url>")
        sys.exit(1)
    
    product_url = sys.argv[1]
    exit_code = asyncio.run(run(product_url))
    sys.exit(exit_code)

python main.py