Python动态网页库存监控与Discord通知：Selenium实战指南

import asyncio
import aiohttp
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException
import time # For sleep in the main loop

# Discord Webhook URL (替换为你的实际URL)
webhook_url = 'YOUR_DISCORD_WEBHOOK_URL'

async def send_webhook_message(content):
    """
    异步发送Discord Webhook消息。
    """
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(webhook_url, json={"content": content}) as response:
                if response.status == 204:
                    print("Discord message sent successfully.")
                else:
                    print(f"Error sending Discord message. Status code: {response.status}, Response: {await response.text()}")
        except aiohttp.ClientError as e:
            print(f"Error sending Discord message: {e}")

def _check_stock_selenium_sync(url, size_to_check):
    """
    同步的Selenium函数，用于检查特定尺码的库存。
    此函数将在单独的线程中运行。
    """
    options = webdriver.ChromeOptions()
    options.add_argument('--headless')  # 无头模式，浏览器不在UI中显示
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-dev-shm-usage')
    options.add_argument('--disable-gpu') # 禁用GPU硬件加速，在某些环境下可能需要
    options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36') # 设置User-Agent

    # 确保 ChromeDriver 路径正确，如果不在PATH中，请指定 executable_path
    driver = webdriver.Chrome(options=options) 

    is_in_stock = False
    try:
        driver.get(url)
        print(f"Navigated to {url}")

        # 等待尺码选择器元素加载。根据网站结构，这些通常是带有特定类名的<li>或<a>元素
        # 这里我们等待所有包含 'swatchanchor' 类的链接元素出现
        WebDriverWait(driver, 15).until(
            EC.presence_of_all_elements_located((By.CSS_SELECTOR, "a.swatchanchor"))
        )
        print("Size swatches loaded.")

        # 查找特定尺码（例如“40”）并且是可选择的元素
        # 网站结构通常是：<li class="selectable"><a class="swatchanchor" title="40">...</a></li>
        # 我们可以通过CSS选择器或XPath来定位

        # CSS Selector: 查找父级为 'selectable' 的 'li' 元素，其子级 'a.swatchanchor' 的 title 属性包含 '40'
        # 或者更直接地查找所有 'swatchanchor'，然后过滤

        # 尝试查找所有尺码选项
        size_elements = driver.find_elements(By.CSS_SELECTOR, "a.swatchanchor")

        for element in size_elements:
            title = element.get_attribute('title')
            parent_li_class = element.find_element(By.XPATH, "./..").get_attribute('class') # 获取父级<li>的class

            if size_to_check in title:
                print(f"Found size element for {size_to_check} with title: '{title}' and parent class: '{parent_li_class}'")
                if 'selectable' in parent_li_class:
                    is_in_stock = True
                    break
                else:
                    print(f"Size {size_to_check} found but is not selectable (class: {parent_li_class}).")

        if is_in_stock:
            print(f"尺码 {size_to_check} 有货。")
        else:
            print(f"尺码 {size_to_check} 缺货或未找到。")

    except TimeoutException:
        print(f"Timeout waiting for page elements on {url}. Page might not have loaded correctly or elements not found within time.")
    except NoSuchElementException:
        print(f"Specific elements (like parent li) not found for {size_to_check}.")
    except Exception as e:
        print(f"An unexpected error occurred during Selenium operation: {e}")
    finally:
        driver.quit() # 确保关闭浏览器实例
    return is_in_stock

async def check_stock(url, size):
    """
    异步包装器，在单独的线程中运行同步的Selenium库存检查。
    """
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, _check_stock_selenium_sync, url, size)

product_data = [
    {'url': 'https://www.courir.com/fr/p/ugg-tasman-1499533.html', 'size': '40'}, # 目标尺码改为 '40'
]

async def main():
    previous_stock_status = {}  # 存储每个产品的上一次库存状态

    while True:
        result_message = ""

        for product_info in product_data:
            url = product_info['url']
            size = product_info['size']

            print(f"\nChecking stock for {url} (size: {size})...")
            current_stock_status = await check_stock(url, size) # 调用异步的Selenium检查函数

            # 获取上一次的库存状态，如果不存在则默认为False（缺货）
            last_status = previous_stock_status.get(url, {}).get(size, False)

            # 检查：如果之前缺货（last_status为False）且现在有货（current_stock_status为True）
            if not last_status and current_stock_status:
                message = f"? {url} - 尺码 {size} 现在有货啦！"
                result_message += message + "\n"
                print(message)
            elif last_status and not current_stock_status:
                # 如果之前有货，现在缺货，也可以选择发送通知
                message = f"⚠️ {url} - 尺码 {size} 已经售罄。"
                # result_message += message + "\n" # 根据需求决定是否通知缺货
                print(message)
            else:
                print(f"尺码 {size} 状态未改变 ({'有货' if current_stock_status else '缺货'})。")

            # 更新当前产品的库存状态
            if url not in previous_stock_status:
                previous_stock_status[url] = {}
            previous_stock_status[url][size] = current_stock_status

        # 如果有任何库存更新，发送Discord消息
        if result_message.strip(): # 确保消息不为空白
            print("\nSending Discord message...")
            await send_webhook_message(result_message)
            print("Discord message sent.")
        else:
            print("\nNo stock updates to send.")

        # 设置检查间隔
        check_interval_seconds = 600 # 10分钟
        print(f"Waiting {check_interval_seconds} seconds before next check...")
        await asyncio.sleep(check_interval_seconds)

# 运行主函数
if __name__ == "__main__":
    asyncio.run(main())