פיתוח מנתח קטלוג מתחרים — מחירים, מלאי, התראות

הקמת חנות מקוונת, מחירי המתחרים משתנים מדי יום, וניטור ידני גוזל שעות רבות מזמנם של המנהלים שלך. ללא איסוף אוטומטי, אתה מפסיד רווח: אינך מגיב לירידת מחיר של מתחרה או מפספס מוצרים חדשים במלאי שלהם. מנתח קטלוג מתחרים הוא כלי שאוסף מדי יום מחירים עדכניים, זמינות ומאפיינים למסד הנתונים שלך. אין צורך לבדוק אתרים ידנית: המערכת סורקת את הקטלוג אוטומטית, מתעדת שינויים ושולחת התראות. הניסיון שלנו — למעלה מ-10 שנים בפיתוח פתרונות כאלה, עשרות פרויקטים מוצלחים מקצה לקצה.

מדוע איסוף ידני אינו יעיל?

ניטור ידני של שלושה מתחרים עם 500 מוצרים כל אחד לוקח 2–3 שעות ביום. שגיאות, השמטות, נתונים מיושנים. מנתח אוטומטי פותר בעיות אלה: אוסף נתונים תוך דקות, עובד 24/7, לעולם לא מתעייף. חיסכון בזמן — עד 90% בהשוואה לאיסוף ידני. מחזיר את ההשקעה תוך 2–3 חודשים.

ניתוח האתר לפני הפיתוח

לפני כתיבת הקוד — ניתוח האתר המבוקש:

• מבנה כתובות URL של הקטלוג: פגינציה באמצעות ?page=N, גלילה אינסופית או ניווט עץ לפי קטגוריות
• עיבוד: HTML סטטי (מהיר ופשוט) או נתונים שנטענים באמצעות XHR/fetch (דורש יירוט או headless)
• הגנה: Cloudflare, הגבלת קצב, אימות
• תדירות עדכון הנתונים באתר — כמה מהר מוצרים חדשים מופיעים ומחירים משתנים

קבוצת השדות המינימלית האופיינית: SKU / מק"ט, כותרת, מחיר (רגיל + מבצע), זמינות, קטגוריה, כתובת URL של דף המוצר, תאריך גרידה. עבור נישות מסוימות, שדות חשובים כוללים: דירוג, מספר ביקורות, משקל/מידות, מותג.

יישום טכני

לאתרים סטטיים — httpx + parsel (או Cheerio עבור Node.js). בקשות אסינכרוניות, מאגר חיבורים של 10–20 עובדים, השהיה של 1–3 שניות בין בקשות לאותו דומיין.

import httpx import asyncio import random from parsel import Selector UA_POOL = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36', ] async def fetch_page(session: httpx.AsyncClient, url: str) -> str: headers = { 'User-Agent': random.choice(UA_POOL), 'Accept-Language': 'ru-RU,ru;q=0.9', } resp = await session.get(url, headers=headers, timeout=15) resp.raise_for_status() return resp.text async def parse_catalog_page(html: str, base_url: str) -> list[dict]: sel = Selector(html) products = [] for item in sel.css('.product-card'): price_raw = item.css('.price::text').get('').strip() price = int(''.join(c for c in price_raw if c.isdigit())) if price_raw else None products.append({ 'title': item.css('.product-title::text').get('').strip(), 'price': price, 'sku': item.attrib.get('data-sku'), 'url': base_url + item.css('a::attr(href)').get(''), 'in_stock': bool(item.css('.in-stock')), 'image_url': item.css('img::attr(src)').get(), }) return products 

עבור SPA עם XHR — יירוט בקשות API באמצעות Playwright. חנויות מקוונות מודרניות רבות, בעת פתיחת דף, מבצעות בקשת fetch ל-API שלהן שמחזיר JSON עם נתוני מוצר:

from playwright.async_api import async_playwright import json async def intercept_catalog_api(catalog_url: str) -> list[dict]: products = [] async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() async def handle_response(response): if '/api/catalog' in response.url and response.status == 200: try: data = await response.json() if 'products' in data: products.extend(data['products']) except Exception: pass page.on('response', handle_response) await page.goto(catalog_url, wait_until='networkidle') await browser.close() return products 

אם ה-API מחזיר JSON ישירות — ניתן לקרוא לו ישירות תוך עקיפת הדפדפן, מהיר פי 10–20. כדי למצוא את נקודת הקצה — השתמש בכרטיסיית DevTools Network תוך גלישה ידנית בקטלוג.

כיצד עובד ניתוח SPA עם XHR?

ב-SPA, האתגר המרכזי אינו ה-HTML אלא בקשות ה-API שטוענות נתונים. אנו מיירטים בקשות אלה באמצעות Playwright ומקבלים JSON נקי. זה אמין יותר מניתוח DOM שנוצר דינמית. אם ה-API פתוח — אנו קוראים לו ישירות, חוסכים במשאבים.

פגינציה וסריקה מלאה

עבור פגינציה באמצעות import httpx import asyncio import random from parsel import Selector UA_POOL = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36', ] async def fetch_page(session: httpx.AsyncClient, url: str) -> str: headers = { 'User-Agent': random.choice(UA_POOL), 'Accept-Language': 'ru-RU,ru;q=0.9', } resp = await session.get(url, headers=headers, timeout=15) resp.raise_for_status() return resp.text async def parse_catalog_page(html: str, base_url: str) -> list[dict]: sel = Selector(html) products = [] for item in sel.css('.product-card'): price_raw = item.css('.price::text').get('').strip() price = int(''.join(c for c in price_raw if c.isdigit())) if price_raw else None products.append({ 'title': item.css('.product-title::text').get('').strip(), 'price': price, 'sku': item.attrib.get('data-sku'), 'url': base_url + item.css('a::attr(href)').get(''), 'in_stock': bool(item.css('.in-stock')), 'image_url': item.css('img::attr(src)').get(), }) return products — סריקה רציפה עד דף ריק:

async def scrape_full_catalog(base_url: str) -> list[dict]: all_products = [] page_num = 1 async with httpx.AsyncClient() as session: while True: url = f'{base_url}?page={page_num}' html = await fetch_page(session, url) products = await parse_catalog_page(html, base_url) if not products: break all_products.extend(products) page_num += 1 await asyncio.sleep(random.uniform(1.5, 3.0)) # вежливая задержка return all_products 

עבור עץ קטגוריות — ראשית אסוף רקורסיבית את כל כתובות ה-URL של הקטגוריות, ולאחר מכן סרוק כל קטגוריה עם פגינציה.

אחסון ועדכונים מצטברים

CREATE TABLE competitor_products ( id SERIAL PRIMARY KEY, source VARCHAR(100) NOT NULL, -- 'competitor_a', 'competitor_b' external_id VARCHAR(255) NOT NULL, title TEXT NOT NULL, price DECIMAL(10,2), price_sale DECIMAL(10,2), in_stock BOOLEAN DEFAULT TRUE, category VARCHAR(500), url TEXT NOT NULL, image_url TEXT, attributes JSONB DEFAULT '{}', first_seen TIMESTAMPTZ DEFAULT NOW(), last_seen TIMESTAMPTZ DEFAULT NOW(), UNIQUE(source, external_id) ); CREATE TABLE competitor_price_history ( id BIGSERIAL PRIMARY KEY, product_id INT REFERENCES competitor_products(id), price DECIMAL(10,2), price_sale DECIMAL(10,2), in_stock BOOLEAN, scraped_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX ON competitor_price_history(product_id, scraped_at DESC); 

בסריקות הבאות — from playwright.async_api import async_playwright import json async def intercept_catalog_api(catalog_url: str) -> list[dict]: products = [] async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() async def handle_response(response): if '/api/catalog' in response.url and response.status == 200: try: data = await response.json() if 'products' in data: products.extend(data['products']) except Exception: pass page.on('response', handle_response) await page.goto(catalog_url, wait_until='networkidle') await browser.close() return products . רשומת היסטוריה נוצרת רק אם המחיר או הזמינות השתנו (השווה עם הרשומה האחרונה באמצעות ?page=N או אחסן async def scrape_full_catalog(base_url: str) -> list[dict]: all_products = [] page_num = 1 async with httpx.AsyncClient() as session: while True: url = f'{base_url}?page={page_num}' html = await fetch_page(session, url) products = await parse_catalog_page(html, base_url) if not products: break all_products.extend(products) page_num += 1 await asyncio.sleep(random.uniform(1.5, 3.0)) # вежливая задержка return all_products בטבלה הראשית).

תזמון והתראות

Celery Beat או Node.js cron. התדירות המומלצת לקטלוג מתחרה — כל 4–12 שעות, בהתאם לדינמיקת המחירים בנישה. עבור שווקים עם מחירים המשתנים במהירות — כל שעה עבור המיקומים המובילים.

התראה כאשר מחיר מתחרה יורד מתחת למחיר שלך — שאילתת SQL או טריגר PostgreSQL עם הודעה ל-Slack/Telegram באמצעות webhook. דוגמת שאילתה:

SELECT cp.title, cp.price AS competitor_price, mp.price AS my_price FROM competitor_products cp JOIN my_products mp ON mp.sku = cp.external_id WHERE cp.source = 'competitor_a' AND cp.price < mp.price AND cp.in_stock = TRUE ORDER BY (mp.price - cp.price) DESC; 

כיצד להגדיר התראות לירידות מחיר של מתחרים?

1. הגדר סף: CREATE TABLE competitor_products ( id SERIAL PRIMARY KEY, source VARCHAR(100) NOT NULL, -- 'competitor_a', 'competitor_b' external_id VARCHAR(255) NOT NULL, title TEXT NOT NULL, price DECIMAL(10,2), price_sale DECIMAL(10,2), in_stock BOOLEAN DEFAULT TRUE, category VARCHAR(500), url TEXT NOT NULL, image_url TEXT, attributes JSONB DEFAULT '{}', first_seen TIMESTAMPTZ DEFAULT NOW(), last_seen TIMESTAMPTZ DEFAULT NOW(), UNIQUE(source, external_id) ); CREATE TABLE competitor_price_history ( id BIGSERIAL PRIMARY KEY, product_id INT REFERENCES competitor_products(id), price DECIMAL(10,2), price_sale DECIMAL(10,2), in_stock BOOLEAN, scraped_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX ON competitor_price_history(product_id, scraped_at DESC); — התראה על ירידה של 5%.
2. הגדר webhook ב-Telegram/Slack.
3. הרץ את שאילתת ה-SQL לאחר כל סריקה ושלח את התוצאה.

אנו מיישמים לוגיקה זו כחלק מהמנתח: אתה מקבל הודעה במסנג'ר עם טבלת מוצרים שבהם המתחרה נעשה זול יותר.

כיצד להבטיח פעולה רציפה של המנתח?

אתרי מתחרים משתנים — המנתח נשבר מעת לעת. אנו מגדירים ניטור: התראה אם בסריקה האחרונה נאספו פחות מ-50% ממספר המוצרים הממוצע. כאשר המבנה משתנה, העדכון בדרך כלל לוקח 2–4 שעות. אנו מבטיחים תמיכה והתאמה לגרסאות חדשות של האתר.

מה כלול בעבודה?

• ניתוח מקיף של האתר המבוקש (מבנה, הגנה, API)
• פיתוח המנתח עם פגינציה, קטגוריות, עדכונים מצטברים
• הגדרת מסד נתונים לאחסון היסטוריית מחירים ומלאי
• תזמון (cron) והגדרת התראות (Telegram/Slack)
• תיעוד לתפעול וגישה
• הדרכת הצוות שלך לשימוש במערכת
• תמיכה באחריות למשך חודש אחד ומענה לתקלות תוך 2–4 שעות

אנו נעריך את הפרויקט שלך — צור קשר, נציע את הפתרון האופטימלי מקצה לקצה. הזמן פיתוח מנתח וקבל כלי שיביא תועלת אמיתית במאבק התחרותי. ויקיפדיה: גרידת אתרים