مراقبة أسعار المنافسين
قم بإعداد مراقبة تلقائية للأسعار عبر مواقع المنافسين باستخدام FourA API.
ما ستقوم ببنائه
سكربت Python يقوم بما يلي:
- جلب صفحات المنتجات من قائمة عناوين URL للمنافسين
- استخراج بيانات الأسعار من HTML
- تسجيل النتائج في ملف CSV
- التشغيل وفق جدول زمني
المتطلبات الأساسية
- مفتاح FourA API (احصل على واحد من هنا)
- Python 3.8+
- حزمتا
requestsوbeautifulsoup4
pip install requests beautifulsoup4
الخطوة 1: تحديد أهدافك
أنشئ قائمة بعناوين URL الخاصة بالمنتجات لمراقبتها:
targets = [
{"name": "Competitor A - Widget", "url": "https://competitor-a.com/widget", "selector": ".price"},
{"name": "Competitor B - Widget", "url": "https://competitor-b.com/products/widget", "selector": "[data-price]"},
{"name": "Competitor C - Widget", "url": "https://competitor-c.com/item/123", "selector": ".product-price span"},
]
الخطوة 2: جلب الصفحات عبر FourA
import requests
import time
SINGLE_URL = "https://eu.api.foura.ai/api/single/"
BROWSER_URL = "https://eu.api.foura.ai/api/browser/"
API_KEY = "YOUR_API_KEY"
HEADERS = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
# A wait doesn't clear these: they reset with the period (credits, traffic)
# or at midnight UTC (the daily Browser allowance).
STOP_LIMITS = {"plan_limit_credits", "plan_limit_bandwidth", "plan_limit_browser_daily"}
def fetch_page(url, use_browser=False, attempts=3):
if use_browser:
endpoint, field = BROWSER_URL, "body"
payload = {"url": url, "timeout_ms": 15000}
else:
endpoint, field = SINGLE_URL, "data"
payload = {"method": "GET", "url": url, "unblocker": True}
for _ in range(attempts):
resp = requests.post(endpoint, headers=HEADERS, json=payload)
if resp.status_code != 429:
return resp.json().get(field, "")
limit = resp.headers.get("X-FourA-Limit")
if limit in STOP_LIMITS:
raise RuntimeError(f"Plan limit reached: {limit}")
time.sleep(int(resp.headers.get("Retry-After", "5")))
raise RuntimeError(f"Still rate limited after {attempts} attempts: {url}")
إن استجابة 429 مع plan_limit_credits أو plan_limit_bandwidth أو plan_limit_browser_daily في X-FourA-Limit لن تُحل بالانتظار لبضع ثوانٍ، لذا تتوقف الدالة بدلاً من إعادة المحاولة. راجع Rate Limits لمعرفة كل حد وموعد إعادة ضبطه.
الخطوة 3: استخراج الأسعار
from bs4 import BeautifulSoup
import re
def extract_price(html, selector):
soup = BeautifulSoup(html, "html.parser")
element = soup.select_one(selector)
if not element:
return None
# Extract numeric price from text like "$49.99" or "49,99 EUR"
text = element.get_text(strip=True)
match = re.search(r'[\d,.]+', text)
return float(match.group().replace(',', '.')) if match else None
الخطوة 4: التشغيل وتسجيل النتائج
import csv
from datetime import datetime
def monitor_prices():
timestamp = datetime.now().isoformat()
results = []
for target in targets:
html = fetch_page(target["url"])
price = extract_price(html, target["selector"])
results.append({
"timestamp": timestamp,
"name": target["name"],
"url": target["url"],
"price": price
})
print(f"{target['name']}: {price}")
time.sleep(1) # Be polite
# Append to CSV
with open("prices.csv", "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["timestamp", "name", "url", "price"])
if f.tell() == 0:
writer.writeheader()
writer.writerows(results)
if __name__ == "__main__":
monitor_prices()
الخطوة 5: جدولة التشغيل
قم بتشغيل السكربت كل ساعة باستخدام cron:
crontab -e
# Add this line:
0 * * * * cd /path/to/project && python3 monitor.py >> monitor.log 2>&1
نصائح
- ابدأ باستخدام endpoint الفردي، وانتقل إلى المتصفح إذا كانت الصفحات تستخدم معالجة JavaScript
- أضف معالجة الأخطاء: تغيّر المواقع هياكل تصميمها. سجّل حالات الفشل بشكل منفصل.
- حافظ على تحديث المحددات: عند قيام منافس بإعادة تصميم موقعه، حدّث محدد CSS
- احترم المواقع: باعد بين requests، وتجنب ساعات الذروة، والتزم بملف robots.txt
الخطوات التالية
- اختيار الـ endpoint المناسب: اختر الطريقة الأفضل
- معالجة الأخطاء: التعامل مع حالات الفشل بسلاسة
- كشط موقع ديناميكي: التعامل مع الصفحات المعالجة عبر JavaScript