Мониторинг цен конкурентов: автоматический сбор данных с сайтов на Playwright

pip install playwright APScheduler pandas
playwright install chromium

import asyncio
import sqlite3
from datetime import datetime
from playwright.async_api import async_playwright

class PriceMonitor:
    def __init__(self):
        self.products = [
            {'name': 'Товар А', 'url': 'https://shop1.example.com/product-a', 'selector': '.price-value'},
            {'name': 'Товар Б', 'url': 'https://shop2.example.com/product-b', 'selector': '.price'},
            {'name': 'Товар В', 'url': 'https://shop3.example.com/product-c', 'selector': '[data-price]'},
        ]

    async def scrape_price(self, product):
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page()

            try:
                await page.goto(product['url'], wait_until='networkidle', timeout=30000)
                element = await page.wait_for_selector(product['selector'], timeout=5000)
                price_text = await element.inner_text()
                price = self.parse_price(price_text)
                await browser.close()
                return {'name': product['name'], 'price': price, 'url': product['url']}
            except Exception as e:
                await browser.close()
                return {'name': product['name'], 'price': None, 'error': str(e)}

    @staticmethod
    def parse_price(text):
        import re
        match = re.search(r'(\d[\d\s.]*(?:,\d{2})?)', text.replace(' ', ''))
        return float(match.group(1).replace(' ', '')) if match else None

    async def run_monitoring(self):
        results = []
        for product in self.products:
            result = await self.scrape_price(product)
            results.append(result)
            await asyncio.sleep(2)  # Задержка между запросами
        return results

monitor = PriceMonitor()
results = asyncio.run(monitor.run_monitoring())
for r in results:
    print(f"{r['name']}: {r.get('price', 'N/A')} руб.")

import sqlite3
from datetime import datetime

def init_price_db():
    conn = sqlite3.connect('prices.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT,
            price REAL,
            url TEXT,
            scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    conn.close()

def save_price(product_name, price, url):
    conn = sqlite3.connect('prices.db')
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO price_history (product_name, price, url) VALUES (?, ?, ?)',
        (product_name, price, url)
    )
    conn.commit()
    conn.close()

init_price_db()

from apscheduler.schedulers.blocking import BlockingScheduler
import asyncio

async def scheduled_monitoring():
    results = await monitor.run_monitoring()
    for r in results:
        if r.get('price'):
            save_price(r['name'], r['price'], r['url'])
            check_price_alert(r['name'], r['price'])

def check_price_alert(product_name, current_price):
    """Проверка на изменение цены и отправка уведомления."""
    conn = sqlite3.connect('prices.db')
    cursor = conn.cursor()
    cursor.execute(
        'SELECT price FROM price_history WHERE product_name = ? ORDER BY scraped_at DESC LIMIT 1',
        (product_name,)
    )
    last_price = cursor.fetchone()
    conn.close()

    if last_price and last_price[0] != current_price:
        send_alert(product_name, last_price[0], current_price)

def send_alert(product_name, old_price, new_price):
    import requests
    change = ((new_price - old_price) / old_price) * 100
    message = f"⚠️ {product_name}: {old_price} → {new_price} ({change:+.1f}%)"

    # Отправка в Telegram
    bot_token = "YOUR_BOT_TOKEN"
    chat_id = "YOUR_CHAT_ID"
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    requests.post(url, json={'chat_id': chat_id, 'text': message})

# Запуск мониторинга каждый час
scheduler = BlockingScheduler()
scheduler.add_job(scheduled_monitoring, 'cron', hour='*/1')
scheduler.start()

import pandas as pd

def analyze_price_trends(product_name, days=30):
    conn = sqlite3.connect('prices.db')
    df = pd.read_sql_query('''
        SELECT scraped_at, price
        FROM price_history
        WHERE product_name = ? AND scraped_at >= datetime('now', '-{} days')
        ORDER BY scraped_at
    '''.format(days), conn, params=(product_name,))
    conn.close()

    if df.empty:
        return None

    df['scraped_at'] = pd.to_datetime(df['scraped_at'])
    df['price_change'] = df['price'].diff()
    df['min_price'] = df['price'].rolling(7).min()
    df['max_price'] = df['price'].rolling(7).max()

    return df.to_dict('records')

# Пример использования
# trends = analyze_price_trends('Товар А')
# print(f"Средняя цена: {df['price'].mean():.0f} руб.")
# print(f"Минимальная: {df['price'].min():.0f} руб.")