Следене на цени на конкуренти
Настройте автоматизирано следене на цените в уебсайтовете на конкурентите с помощта на 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"
}
def fetch_page(url, use_browser=False):
if use_browser:
resp = requests.post(BROWSER_URL, headers=HEADERS, json={
"url": url,
"timeout_ms": 15000
})
if resp.status_code == 429:
time.sleep(5)
return fetch_page(url, use_browser)
return resp.json().get("body", "")
else:
resp = requests.post(SINGLE_URL, headers=HEADERS, json={
"method": "GET",
"url": url,
"unblocker": True
})
if resp.status_code == 429:
time.sleep(5)
return fetch_page(url, use_browser)
return resp.json().get("data", "")
Стъпка 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
Съвети
- Започете с single endpoint, преминете към browser, ако страниците използват JavaScript рендиране
- Добавете обработка на грешки: сайтовете променят дизайна си. Записвайте неуспешните опити отделно.
- Поддържайте селекторите актуални: когато конкурент промени дизайна си, актуализирайте CSS селектора
- Уважавайте сайтовете: правете паузи между заявките, избягвайте пиковите часове, спазвайте robots.txt
Следващи стъпки
- Избор на правилния endpoint: Изберете най-добрия подход
- Обработка на грешки: Справяйте се с грешките елегантно
- Скрапване на динамичен уебсайт: Справяне със страници, рендирани с JavaScript