أتمتة تتبع أسعار المنتجات بـ Playwright و Discord — السكربت الكامل

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

DISCORD_WEBHOOK = os.environ["DISCORD_WEBHOOK_URL"]

async def get_price(page, product):
    await page.goto(product["url"], wait_until="domcontentloaded", timeout=30000)
    await page.wait_for_selector(product["selector"], timeout=15000)
    text = await page.inner_text(product["selector"])
    digits = "".join(c for c in text if c.isdigit() or c == ".")
    return float(digits)

async def notify(product, current_price):
    message = {
        "content": (
            f"السعر نزل! **{product['name']}**\n"
            f"السعر الحالي: {current_price} EGP\n"
            f"الحد المستهدف: {product['target_price']} EGP\n"
            f"{product['url']}"
        )
    }
    async with httpx.AsyncClient(timeout=10) as client:
        await client.post(DISCORD_WEBHOOK, json=message)

async def main():
    with open("products.json") as f:
        products = json.load(f)

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0.0.0 Safari/537.36"
            )
        )
        page = await context.new_page()

        for product in products:
            try:
                price = await get_price(page, product)
                if price <= product["target_price"]:
                    await notify(product, price)
                    print(f"[ALERT] {product['name']}: {price}")
                else:
                    print(f"[OK] {product['name']}: {price} > {product['target_price']}")
            except Exception as e:
                print(f"[ERROR] {product['name']}: {e}")

        await browser.close()

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