الدليل العملي لاستخراج بيانات والترت: من كتالوج المنتجات إلى تجاوز الحماية

لماذا يصعب استخراج بيانات والترت؟

إذا كنت تعمل في فريق استخبارات التجزئة أو تحليل أسعار CPG، فأنت تعرف أن والترت يُعد من أكثر المنصات تعقيدًا في مجال استخراج البيانات. الموقع يستخدم طبقات حماية متعددة، ويتغير هيكله باستمرار، ويمزج بين منتجات الطرف الأول (1P) وبائعي الطرف الثالث (3P Marketplace) في صفحة واحدة.

النتيجة؟ فشل الطلبات، حظر عناوين IP، وبيانات غير مكتملة تُفسد خطوط أنابيب البيانات الخاصة بك. هذا الدليل يقدم لك نهجًا عمليًا — من فهم بنية الكتالوج إلى كتابة كود إنتاجي ينجح فعليًا.

بنية كتالوج منتجات والترت

قبل كتابة أي كود، يجب أن تفهم الطرق الثلاث الرئيسية للوصول إلى بيانات المنتجات على والترت:

صفحات المنتجات الفردية (Item Pages)

كل منتج له URL بصيغة: /ip/{slug}/{itemId}. مثال:

https://www.walmart.com/ip/Apple-AirPods-Pro/395273703

الجزء المهم هو itemId (مثل 395273703) — فهو المعرف الفريد. أما الـ slug فيمكن أن يتغير أو يكون مفقودًا، لكن itemId يبقى ثابتًا. يمكنك حتى الوصول للمنتج عبر /ip/{itemId} مباشرة.

صفحات الفئات (Category Pages)

الفئات تتبع بنية شجرية: /cp/electronics/3944. كل صفحة فئة تعرض شبكة منتجات مع ترقيم صفحات. مشكلة صفحات الفئات أنها تحمل أول 40 منتجًا فقط في الـ HTML المبدئي، والباقي يُحمّل عبر AJAX — مما يعني أنك ستحتاج إلى إجراء طلبات إضافية أو الاعتماد على البحث.

صفحات البحث (Search)

والترت يعرض نتائج البحث عبر /search?q={query}. هذه الطريقة مفيدة لاستخراج بيانات فئات واسعة، لكنها تعاني من نفس مشكلة الترقيم والتحميل الكسول.

الخلاصة: للحصول على بيانات كاملة ودقيقة، صفحة المنتج الفردية هي مصدرك الأفضل. استخدم الفئات والبحث لاكتشاف itemId فقط.

حماية Akamai + PerimeterX: لماذا تحتاج بروكسيات سكنية

والترت يستخدم كلاً من Akamai Bot Manager وPerimeterX (الذي أصبح الآن HUMAN Security) كطبقات حماية أساسية. إليك ما تفعله كل طبقة:

طبقة الحماية	ما تكتشفه	الاستجابة
Akamai Bot Manager	بصمة المتصفح (TLS fingerprint)، أنماط الطلبات، إطارات HTTP/2	تحدي JavaScript (صورة Akamai)، حظر 403
PerimeterX / HUMAN	سلوك الماوس واللوحة المفاتيح، إعادة تشغيل الطلبات، كثافة الطلبات من نفس IP	تحدي CAPTCHA، كتلة كاملة، أو بيانات مزيفة

هذا يعني أن الطلبات من مراكز البيانات (datacenter IPs) تُحظر بسرعة — غالبًا بعد 5-15 طلبًا فقط. حتى مع رؤوس طلبات مثالية، بصمة TLS لمكتبة requests في بايثون تكشفها Akamai فورًا.

لماذا البروكسيات السكنية هي الحل الوحيد العملي

عناوين IP السكنية تنتمي لمزودي خدمة إنترنت حقيقيين (ISPs)، لذا فإن Akamai وPerimeterX تعتبرها زيارات من مستخدمين حقيقيين. مع دوران IP لكل طلب، يمكنك توزيع الطلبات عبر آلاف العناوين المختلفة.

مع ProxyHat، يمكنك تحديد الدولة والمدينة لتتناسب مع سوقك المستهدف:

# بروكسيات سكنية أمريكية مع دوران IP لكل طلب
http://user-country-US:PASSWORD@gate.proxyhat.com:8080

# بروكسيات سكنية من مدينة محددة
http://user-country-US-city-newyork:PASSWORD@gate.proxyhat.com:8080

متى تستخدم الجلسات الثابتة (Sticky Sessions)

بعض السيناريوهات تحتاج لنفس IP لفترة قصيرة — مثل إضافة منتجات للسلة أو تصفح متعدد الصفحات. استخدم علامة الجلسة:

# جلسة ثابتة لمدة تصل إلى 10 دقائق
http://user-session-abc123-country-US:PASSWORD@gate.proxyhat.com:8080

أسهل طريقة: تحليل __NEXT_DATA__ المدمج

هنا تكمن الحيلة الأهم في هذا الدليل. والترت يبني واجهته باستخدام Next.js، وهذا يعني أن كل صفحة منتج تحتوي على كائن JSON كامل مدمج في الـ HTML:

<script id="__NEXT_DATA__" type="application/json">
  { ... كل بيانات المنتج هنا ... }
</script>

هذا الكائن يحتوي على كل شيء: السعر، حالة المخزون، التقييمات، بيانات البائع، المتغيرات، وحتى بيانات التوصيل. لست بحاجة لتحليل HTML أو إجراء طلبات API منفصلة.

المقارنة بين طرق الاستخراج

الطريقة	الصعوبة	موثوقية البيانات	خطر الحظر	ملاحظات
تحليل HTML (CSS/XPath)	متوسطة	منخفضة	مرتفع	يتغير مع كل تحديث للواجهة
__NEXT_DATA__ JSON	منخفضة	عالية	منخفض	أفضل طريقة — بيانات منظمة وكاملة
واجهة برمجة التطبيقات الداخلية	عالية	عالية	مرتفع جدًا	تحتاج رموزًا مميزة صالحة وتوقيعات
Walmart Affiliate API	منخفضة	متوسطة	لا يوجد	بيانات محدودة، تحتاج تسجيل

مثال عملي: استخراج بيانات منتج كامل ببايثون

إليك كود كامل يجمع بين البروكسيات السكنية وتحليل __NEXT_DATA__:

import requests
import json
import re
import time
from urllib.parse import quote

PROXY_URL = "http://user-country-US:PASSWORD@gate.proxyhat.com:8080"

PROXIES = {
    "http": PROXY_URL,
    "https": PROXY_URL,
}

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/125.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
}

def fetch_product(item_id: str, session_id: str = None) -> dict:
    """جلب صفحة منتج والترت واستخراج __NEXT_DATA__"""
    url = f"https://www.walmart.com/ip/{item_id}"
    
    # استخدام جلسة ثابتة إذا طُلبت
    if session_id:
        proxy = f"http://user-session-{session_id}-country-US:PASSWORD@gate.proxyhat.com:8080"
        proxies = {"http": proxy, "https": proxy}
    else:
        proxies = PROXIES
    
    resp = requests.get(url, headers=HEADERS, proxies=proxies, timeout=30)
    resp.raise_for_status()
    
    # استخراج __NEXT_DATA__ من HTML
    pattern = r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>'
    match = re.search(pattern, resp.text)
    if not match:
        raise ValueError("لم يتم العثور على __NEXT_DATA__ — ربما تم حظرك")
    
    data = json.loads(match.group(1))
    return data

def parse_product(next_data: dict) -> dict:
    """تحليل بيانات المنتج من __NEXT_DATA__"""
    # التنقل عبر بنية Next.js
    props = next_data["props"]["pageProps"]
    
    # المنتج الأساسي
    product = props.get("initialData", {}).get("data", {})
    item = product.get("product", {})
    
    # بيانات أساسية
    result = {
        "item_id": item.get("id"),
        "name": item.get("name"),
        "brand": item.get("brand"),
        "category_path": item.get("category", {}).get("path"),
    }
    
    # السعر
    price_info = item.get("priceInfo", {})
    result["price"] = {
        "current": price_info.get("currentPrice", {}).get("price"),
        "original": price_info.get("originalPrice", {}).get("price"),
        "currency": price_info.get("currencyUnit", "USD"),
    }
    
    # المخزون والتوفر
    availability = item.get("availability", {})
    result["in_stock"] = availability.get("status") == "IN_STOCK"
    result["stock_level"] = availability.get("quantity", None)
    
    # التقييمات
    review_stats = item.get("reviewStatistics", {})
    result["ratings"] = {
        "average": review_stats.get("averageRating"),
        "count": review_stats.get("reviewCount"),
    }
    
    # بيانات البائع
    seller = item.get("seller", {})
    result["seller"] = {
        "id": seller.get("id"),
        "name": seller.get("name"),
        "is_marketplace": seller.get("isThirdParty", False),
    }
    
    return result

# مثال الاستخدام
data = fetch_product("395273703")
product_info = parse_product(data)
print(json.dumps(product_info, indent=2, ensure_ascii=False))

لاحظ أن بنية __NEXT_DATA__ قد تتغير بين تحديثات والترت. المفتاح هو فحص الكائن يدويًا أولًا لتحديد المسارات الصحيحة:

# حفظ __NEXT_DATA__ لفحصه يدويًا
def save_next_data(item_id: str, filename: str = None):
    data = fetch_product(item_id)
    if not filename:
        filename = f"walmart_{item_id}_next_data.json"
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    print(f"تم الحفظ في {filename}")
    
    # طباعة المفاتيح الرئيسية لفهم البنية
    props = data["props"]["pageProps"]
    print("مفاتيح pageProps:", list(props.keys()))

save_next_data("395273703")

التعامل مع Walmart Marketplace (بائعو الطرف الثالث)

هناك فرق جوهري بين منتجات والترت المباشرة (1P) ومنتجات بائعي الطرف الثالث (3P Marketplace). فهم هذا الفرق ضروري لفرق استخبارات التجزئة.

كيف تميّز بين 1P و 3P

في كائن __NEXT_DATA__، ابحث عن:

• منتج 1P: البائع هو Walmart.com — seller.id يكون فارغًا أو يحتوي على معرف والترت، وseller.isThirdParty يساوي false.
• منتج 3P: البائع طرف ثالث — seller.isThirdParty يساوي true، وseller.name يكون اسم المتجر.

لماذا يهم هذا لفرق CPG؟

1. تسعير MAP: بائعو 3P قد يخالفون تسعير الحد الأدنى للإعلان (MAP). مراقبة هذا تحتاج لبيانات منفصلة عن 1P.
2. مخزون منافس: قد يكون المنتج غير متوفر على 1P لكن متوفر عبر 3P — والعكس صحيح.
3. جودة البيانات: بيانات 3P أقل موثوقية — أسماء المنتجات قد تحتوي على كلمات مفتاحية محشوة، والأسعار قد لا تشمل رسوم الشحن.

def parse_all_offers(next_data: dict) -> list:
    """استخراج جميع عروض البائعين من صفحة المنتج"""
    props = next_data["props"]["pageProps"]
    initial_data = props.get("initialData", {}).get("data", {})
    
    offers = []
    
    # العروض المتعددة في قسم BuyBox
    buybox = initial_data.get("buybox", {})
    items = buybox.get("items", [])
    
    for item in items:
        offer = {
            "seller_id": item.get("sellerId"),
            "seller_name": item.get("sellerName"),
            "is_3p": item.get("isThirdPartyItem", False),
            "price": item.get("price"),
            "shipping_cost": item.get("shippingPrice"),
            "delivery_date": item.get("fulfillment", {}).get("deliveryDate"),
            "in_stock": item.get("availability", {}).get("status") == "IN_STOCK",
        }
        offers.append(offer)
    
    return offers

# مثال: منتج له عروض متعددة
data = fetch_product("395273703")
all_offers = parse_all_offers(data)
for offer in all_offers:
    print(f"البائع: {offer['seller_name']} | 3P: {offer['is_3p']} | السعر: {offer['price']}")

جدولة الطلبات مع مراعاة حدود السرعة

والترت يفرض حدودًا صارمة على الطلبات. حتى مع بروكسيات سكنية، الإفراط في الطلبات يُفعّل حماية PerimeterX. إليك استراتيجية جدولة عملية:

حدود السرعة المُقترحة

• ببروكسيات سكنية دوّارة: 1-2 طلب لكل ثانية لكل عملية (مع IP مختلف لكل طلب).
• بجلسة ثابتة: طلب واحد كل 3-5 ثوانٍ لكل جلسة.
• حد يومي آمن: لا تتجاوز 50,000 طلب لكل حساب/عملية في اليوم.

import time
import random
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("walmart_scraper")

class RateLimitedScraper:
    def __init__(self, proxy_user: str, proxy_pass: str, 
                 min_delay: float = 1.5, max_delay: float = 3.5,
                 max_retries: int = 3):
        self.proxy_user = proxy_user
        self.proxy_pass = proxy_pass
        self.min_delay = min_delay
        self.max_delay = max_delay
        self.max_retries = max_retries
        self.request_count = 0
        self.session = requests.Session()
        self.session.headers.update(HEADERS)
    
    def _get_proxy(self, session_id: str = None) -> dict:
        """إنشاء URL بروكسي مع دوران IP"""
        if session_id:
            proxy_url = (
                f"http://user-session-{session_id}-country-US"
                f":{self.proxy_pass}@gate.proxyhat.com:8080"
            )
        else:
            proxy_url = (
                f"http://user-country-US"
                f":{self.proxy_pass}@gate.proxyhat.com:8080"
            )
        return {"http": proxy_url, "https": proxy_url}
    
    def scrape_product(self, item_id: str) -> dict | None:
        """جلب وتحليل منتج مع إعادة المحاولة والجدولة"""
        for attempt in range(self.max_retries):
            try:
                proxies = self._get_proxy()
                url = f"https://www.walmart.com/ip/{item_id}"
                
                resp = self.session.get(
                    url, proxies=proxies, timeout=30
                )
                
                # التحقق من عدم الحظر
                if "robot" in resp.url.lower() or resp.status_code == 403:
                    logger.warning(f"حظر محتمل على {item_id}، محاولة {attempt+1}")
                    time.sleep(random.uniform(10, 20))
                    continue
                
                resp.raise_for_status()
                
                # استخراج __NEXT_DATA__
                match = re.search(
                    r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
                    resp.text
                )
                if not match:
                    logger.error(f"لا __NEXT_DATA__ في {item_id}")
                    continue
                
                next_data = json.loads(match.group(1))
                product = parse_product(next_data)
                
                self.request_count += 1
                logger.info(
                    f"[{self.request_count}] نجح: {item_id} - {product['name'][:50]}"
                )
                
                # تأخير عشوائي بين الطلبات
                delay = random.uniform(self.min_delay, self.max_delay)
                time.sleep(delay)
                
                return product
                
            except requests.exceptions.RequestException as e:
                logger.error(f"خطأ في {item_id}: {e}")
                time.sleep(random.uniform(5, 15))
        
        logger.error(f"فشل نهائي لـ {item_id} بعد {self.max_retries} محاولات")
        return None

# مثال الاستخدام
scraper = RateLimitedScraper(
    proxy_user="user", 
    proxy_pass="your_password",
    min_delay=2.0,
    max_delay=4.0
)

item_ids = ["395273703", "55164693", "825862932"]
results = []
for item_id in item_ids:
    product = scraper.scrape_product(item_id)
    if product:
        results.append(product)
    else:
        # تأخير أطول بعد الفشل
        time.sleep(random.uniform(30, 60))

print(f"تم استخراج {len(results)} من {len(item_ids)} منتج")

نصائح عملية للجدولة

• توزّع على مدار اليوم: لا تُرسل كل طلباتك في ساعة واحدة. وزّعها على 18-20 ساعة.
• غيّر User-Agent: استخدم مكتبة مثل fake-useragent لتدوير رأس User-Agent مع كل جلسة جديدة.
• احترم robots.txt: تحقق من https://www.walmart.com/robots.txt لمعرفة المسارات المحظورة.
• راقب معدل النجاح: إذا انخفض عن 85%، قلّل السرعة فورًا.
• استخدم جلسات ثابتة للمنتجات ذات الصلة: إذا كنت تستخرج منتجات من نفس الفئة، استخدم نفس الجلسة لمدة 5-10 طلبات ثم غيّرها.

مثال على استجابة __NEXT_DATA__ (مقتطف)

هذا مقتطف من كيف تبدو البيانات فعليًا داخل __NEXT_DATA__:

{
  "props": {
    "pageProps": {
      "initialData": {
        "data": {
          "product": {
            "id": "395273703",
            "name": "Apple AirPods Pro (2nd Generation)",
            "brand": "Apple",
            "priceInfo": {
              "currentPrice": { "price": 189.99, "currencyUnit": "USD" },
              "originalPrice": { "price": 249.99, "currencyUnit": "USD" }
            },
            "availability": {
              "status": "IN_STOCK",
              "quantity": null
            },
            "reviewStatistics": {
              "averageRating": 4.6,
              "reviewCount": 15234
            },
            "seller": {
              "id": "",
              "name": "Walmart.com",
              "isThirdParty": false
            },
            "category": { "path": "Electronics/Headphones/AirPods" }
          }
        }
      }
    }
  }
}

استخراج بيانات البحث بالجملة

لاكتشاف itemId للمنتجات، ستحتاج للبحث أو تصفح الفئات. إليك طريقة عملية:

def search_walmart(query: str, max_pages: int = 3) -> list[str]:
    """البحث في والترت واستخراج معرفات المنتجات"""
    item_ids = []
    
    for page in range(1, max_pages + 1):
        url = f"https://www.walmart.com/search?q={quote(query)}&page={page}"
        
        proxies = {
            "http": "http://user-country-US:PASSWORD@gate.proxyhat.com:8080",
            "https": "http://user-country-US:PASSWORD@gate.proxyhat.com:8080",
        }
        
        try:
            resp = requests.get(url, headers=HEADERS, proxies=proxies, timeout=30)
            resp.raise_for_status()
            
            # استخراج itemId من نتائج البحث
            pattern = r'"itemId":"(\d+)"'
            ids = re.findall(pattern, resp.text)
            item_ids.extend(ids)
            
            logger.info(f"صفحة {page}: وُجد {len(ids)} منتج")
            time.sleep(random.uniform(3, 6))
            
        except requests.RequestException as e:
            logger.error(f"فشل البحث في صفحة {page}: {e}")
            break
    
    # إزالة التكرارات مع الحفاظ على الترتيب
    seen = set()
    unique_ids = []
    for id in item_ids:
        if id not in seen:
            seen.add(id)
            unique_ids.append(id)
    
    return unique_ids

# اكتشاف منتجات ثم استخراج بياناتها
ids = search_walmart("wireless headphones", max_pages=2)
print(f"تم اكتشاف {len(ids)} منتج فريد")

الاعتبارات الأخلاقية والقانونية

قبل البدء في أي مشروع استخراج بيانات، ضع في اعتبارك:

• احترم robots.txt: والترت يحدد مسارات لا يجب الوصول إليها. راجع https://www.walmart.com/robots.txt.
• البيانات العامة فقط: لا تستخرج بيانات مستخدمين أو مراجعات شخصية.
• لا تضر بالخدمة: حدود السرعة موجودة لحماية البنية التحتية. احترامها يفيد الجميع.
• GDPR وCCPA: إذا كنت تخزن أي بيانات شخصية من المراجعات، تأكد من الامتثال للقوانين.
• شروط الخدمة: استخراج البيانات قد ينتهك شروط خدمة والترت. استشر فريقك القانوني.

النقاط الرئيسية

1. استخدم __NEXT_DATA__ المدمج في HTML — إنه أسرع وأكثر موثوقية من تحليل HTML أو استدعاء واجهات برمجة تطبيقات داخلية.

2. بروكسيات سكنية ليست رفاهية — إنها ضرورة. Akamai وPerimeterX يحظران عناوين مراكز البيانات بسرعة.

3. ميّز بين منتجات 1P و 3P Marketplace — فرق CPG تحتاج لهذا التمييز لمراقبة MAP وتحليل المنافسة.

4. جدول طلباتك: 1-2 طلب/ثانية مع بروكسيات دوّارة، وتأخير عشوائي بين 1.5-4 ثوانٍ.

5. افحص بنية __NEXT_DATA__ يدويًا أولًا قبل بناء خط أنابيب — والترت يحدّث واجهته بانتظام.

هل أنت مستعد لبدء استخراج بيانات والترت؟ جرّب ProxyHat الآن واحصل على بروكسيات سكنية من أكثر من 190 دولة مع دوران IP تلقائي. أو تعرّف على المزيد من حالات استخدام استخراج الويب التي تدعمها بنيتنا التحتية.