Мониторинг цен конкурентов: парсер + Telegram-уведомления

Знать цены конкурентов — базовая потребность любого бизнеса. Вместо ежедневного ручного мониторинга можно автоматизировать процесс: скрипт собирает цены, сохраняет историю в SQLite, сравнивает с предыдущими значениями и отправляет уведомления в Telegram при изменениях. Разберём полный проект на Python.

Архитектура проекта

scraper.py — сбор цен с сайтов конкурентов
database.py — хранение истории цен в SQLite
notifier.py — отправка уведомлений в Telegram
main.py — точка входа, запуск по расписанию через cron
config.py — настройки: URL, селекторы, токены

Конфигурация: что мониторим

# config.py
import os

TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
TELEGRAM_CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]

DB_PATH = "prices.db"

# Список товаров для мониторинга
PRODUCTS = [
    {
        "name": "Разработка лендинга — Студия А",
        "url": "https://studio-a.ru/prices",
        "selector": ".price-landing .amount",
        "competitor": "Студия А"
    },
    {
        "name": "Разработка лендинга — Студия Б",
        "url": "https://studio-b.ru/uslugi/lending",
        "selector": "[data-service='landing'] .price",
        "competitor": "Студия Б"
    },
    {
        "name": "Интернет-магазин — Студия А",
        "url": "https://studio-a.ru/prices",
        "selector": ".price-shop .amount",
        "competitor": "Студия А"
    },
]

# Порог изменения для уведомления (в процентах)
PRICE_CHANGE_THRESHOLD = 5

База данных: SQLite для истории цен

# database.py
import sqlite3
from datetime import datetime
from config import DB_PATH

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS prices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT NOT NULL,
            competitor TEXT NOT NULL,
            price REAL NOT NULL,
            url TEXT NOT NULL,
            checked_at TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_product_date
        ON prices (product_name, checked_at)
    """)
    conn.commit()
    conn.close()

def save_price(product_name: str, competitor: str,
               price: float, url: str):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT INTO prices (product_name, competitor, price, url, checked_at) "
        "VALUES (?, ?, ?, ?, ?)",
        (product_name, competitor, price, url,
         datetime.now().isoformat())
    )
    conn.commit()
    conn.close()

def get_last_price(product_name: str) -> float | None:
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT price FROM prices WHERE product_name = ? "
        "ORDER BY checked_at DESC LIMIT 1",
        (product_name,)
    )
    row = cursor.fetchone()
    conn.close()
    return row[0] if row else None

def get_price_history(product_name: str, days: int = 30) -> list:
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT price, checked_at FROM prices "
        "WHERE product_name = ? "
        "AND checked_at >= datetime('now', ?) "
        "ORDER BY checked_at",
        (product_name, f"-{days} days")
    )
    rows = cursor.fetchall()
    conn.close()
    return rows

Скрапер: сбор цен

# scraper.py
import re
import asyncio
import httpx
from bs4 import BeautifulSoup

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36"
}

def extract_price(text: str) -> float | None:
    """Извлекает числовую цену из строки вида '150 000 ₽' или 'от 99900 руб.'"""
    cleaned = re.sub(r"[^\d.,]", "", text.replace(" ", ""))
    cleaned = cleaned.replace(",", ".")
    try:
        return float(cleaned)
    except ValueError:
        return None

async def fetch_price(client: httpx.AsyncClient,
                      product: dict) -> dict | None:
    try:
        response = await client.get(
            product["url"], headers=HEADERS, timeout=15.0
        )
        response.raise_for_status()

        soup = BeautifulSoup(response.text, "lxml")
        element = soup.select_one(product["selector"])

        if not element:
            print(f"Селектор не найден: {product['name']}")
            return None

        price = extract_price(element.text)
        if price is None:
            print(f"Не удалось извлечь цену: {element.text}")
            return None

        return {
            "name": product["name"],
            "competitor": product["competitor"],
            "price": price,
            "url": product["url"]
        }
    except Exception as e:
        print(f"Ошибка при парсинге {product['name']}: {e}")
        return None

async def scrape_all(products: list) -> list:
    semaphore = asyncio.Semaphore(3)

    async def limited_fetch(client, product):
        async with semaphore:
            result = await fetch_price(client, product)
            await asyncio.sleep(1)
            return result

    async with httpx.AsyncClient() as client:
        tasks = [limited_fetch(client, p) for p in products]
        results = await asyncio.gather(*tasks)

    return [r for r in results if r is not None]

Telegram-уведомления

# notifier.py
import httpx
from config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID

async def send_telegram(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    async with httpx.AsyncClient() as client:
        await client.post(url, json={
            "chat_id": TELEGRAM_CHAT_ID,
            "text": message,
            "parse_mode": "HTML"
        })

def format_price_change(name: str, old_price: float,
                        new_price: float, url: str) -> str:
    diff = new_price - old_price
    pct = (diff / old_price) * 100
    arrow = "\u2b06\ufe0f" if diff > 0 else "\u2b07\ufe0f"

    return (
        f"{arrow} {name}\n"
        f"Было: {old_price:,.0f} ₽\n"
        f"Стало: {new_price:,.0f} ₽\n"
        f"Изменение: {pct:+.1f}%\n"
        f"Ссылка"
    )

async def send_daily_report(changes: list, unchanged: int):
    if not changes:
        return

    header = f"Мониторинг цен\n{'=' * 20}\n\n"
    body = "\n\n".join(changes)
    footer = f"\n\nБез изменений: {unchanged} позиций"

    await send_telegram(header + body + footer)

Главный скрипт

# main.py
import asyncio
from config import PRODUCTS, PRICE_CHANGE_THRESHOLD
from database import init_db, save_price, get_last_price
from scraper import scrape_all
from notifier import format_price_change, send_daily_report

async def main():
    init_db()

    results = await scrape_all(PRODUCTS)
    changes = []
    unchanged = 0

    for item in results:
        old_price = get_last_price(item["name"])
        save_price(
            item["name"], item["competitor"],
            item["price"], item["url"]
        )

        if old_price is None:
            continue

        diff_pct = abs((item["price"] - old_price) / old_price * 100)

        if diff_pct >= PRICE_CHANGE_THRESHOLD:
            msg = format_price_change(
                item["name"], old_price,
                item["price"], item["url"]
            )
            changes.append(msg)
        else:
            unchanged += 1

    await send_daily_report(changes, unchanged)

    print(f"Проверено: {len(results)} позиций")
    print(f"Изменения: {len(changes)}, без изменений: {unchanged}")

if __name__ == "__main__":
    asyncio.run(main())

Настройка cron

Запускаем проверку ежедневно в 9:00 и 18:00:

# crontab -e
0 9,18 * * * cd /opt/price-monitor && /usr/bin/python3 main.py >> /var/log/price-monitor.log 2>&1

Для Windows используйте Планировщик задач или запускайте внутри Docker-контейнера с cron.

Развитие проекта

Базовый мониторинг можно расширить:

Веб-дашборд — добавьте Flask/FastAPI с графиками истории цен через Chart.js
Несколько каналов — отправляйте отчёты в разные Telegram-чаты для разных категорий
Playwright для SPA — если конкурент использует SPA, замените httpx на Playwright для тех URL
Экспорт в Google Sheets — автоматическая выгрузка через Google Sheets API
Алерты по порогу — уведомление, если цена конкурента упала ниже вашей