Запустил мониторинг вручную, получил таблицу в 400 строк и потратил два часа на то, чтобы найти в ней пять нужных позиций. Именно тогда решил: хватит, пусть за конкурентами следит агент.
Расскажу как это устроено у меня сейчас — с конкретными инструментами, граблями и парой мест, где я сначала пошёл не туда.
Что вообще делает такой агент
По сути это три части: что-то собирает данные, что-то их обрабатывает, что-то сообщает о результате. Звучит просто. На практике большая часть времени уходит на первое — сбор. Сайты конкурентов не горят желанием отдавать цены в удобном формате.
Мой агент раз в несколько часов обходит список URL, вытаскивает цену с каждой страницы, сравнивает с предыдущим значением и если изменение больше порога — пишет мне в Telegram. Всё. Никакой магии.
Но чтобы это работало стабильно, пришлось разобраться с несколькими нетривиальными вещами.
Инструменты, которые я выбрал и почему
Начал с Playwright — библиотека для управления браузером из кода. Сначала пробовал requests + BeautifulSoup, но половина целевых сайтов грузит цены через JavaScript, и статичный парсер видел пустые div-ы вместо цифр. Playwright запускает настоящий браузер, ждёт пока страница загрузится, и только потом берёт HTML.
Для логики агента взял Python и APScheduler — она умеет запускать функции по расписанию без отдельного cron. Данные храню в SQLite: таблица с URL, последней ценой и временем проверки.
ИИ-часть — GPT-4o через API OpenAI. Здесь объясню зачем он вообще нужен, потому что это не самоочевидно.
Зачем здесь вообще GPT
Первая версия агента работала без языковой модели. Я писал регулярные выражения для каждого сайта отдельно: на одном цена в span с классом price, на другом — в div с data-атрибутом, на третьем — вообще в JSON внутри script-тега. Работало, но каждый новый сайт требовал получаса ковыряния в DevTools.
Потом попробовал передавать HTML в GPT с простым промптом: "Найди цену товара на этой странице и верни только число". Модель справляется с большинством случаев без ручной настройки под конкретный сайт. Не со всеми — бывают страницы где цен несколько (за штуку, за упаковку, с НДС и без), и тут нужно уточнять промпт. Но базовые случаи закрываются автоматически.
Дело в том, что я не отдаю GPT всю страницу целиком. Полный HTML крупного интернет-магазина — это 200-300 килобайт мусора. Сначала вырезаю через Playwright только body, потом фильтрую через BeautifulSoup, удаляю скрипты и стили. В итоге модель получает 5-15 килобайт осмысленного текста. И быстрее, и дешевле.
Как устроен код
Покажу скелет — не весь проект, но достаточно чтобы было понятно как части соединяются.
import asyncio
import sqlite3
from datetime import datetime
from playwright.async_api import async_playwright
from openai import OpenAI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import httpx
client = OpenAI() # читает OPENAI_API_KEY из переменных окружения
DB_PATH = "prices.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS prices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT NOT NULL,
product_name TEXT,
price REAL,
checked_at TEXT
)
""")
conn.commit()
conn.close()
async def fetch_page_text(url: str) -> str:
"""Загружает страницу через Playwright, возвращает очищенный текст."""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# Блокируем картинки и шрифты — нам они не нужны, только тормозят
await page.route("**/*.{png,jpg,jpeg,gif,svg,woff,woff2}",
lambda route: route.abort())
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
await page.wait_for_timeout(2000) # ждём JS
# Берём только видимый текст, без тегов
text = await page.inner_text("body")
await browser.close()
return text[:8000] # обрезаем чтобы не перегружать модель
def extract_price_with_gpt(page_text: str, product_hint: str = "") -> float | None:
"""Просит GPT найти цену в тексте страницы."""
prompt = f"""На странице интернет-магазина есть информация о товаре.
Твоя задача — найти актуальную цену продажи (не зачёркнутую старую цену).
{"Товар: " + product_hint if product_hint else ""}
Верни только число — цену в рублях без знаков валюты, пробелов и слов.
Если цену найти не удалось — верни null.
Текст страницы:
{page_text}"""
response = client.chat.completions.create(
model="gpt-4o-mini", # mini достаточно для такой задачи
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=20
)
result = response.choices[0].message.content.strip()
try:
return float(result.replace(",", "."))
except (ValueError, TypeError):
return None
def get_last_price(url: str) -> float | None:
conn = sqlite3.connect(DB_PATH)
row = conn.execute(
"SELECT price FROM prices WHERE url = ? ORDER BY checked_at DESC LIMIT 1",
(url,)
).fetchone()
conn.close()
return row[0] if row else None
def save_price(url: str, price: float, product_name: str = ""):
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT INTO prices (url, product_name, price, checked_at) VALUES (?, ?, ?, ?)",
(url, product_name, price, datetime.now().isoformat())
)
conn.commit()
conn.close()
async def send_telegram_alert(message: str, bot_token: str, chat_id: str):
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
async with httpx.AsyncClient() as http:
await http.post(url, json={"chat_id": chat_id, "text": message})
async def check_price(url: str, product_name: str,
bot_token: str, chat_id: str,
change_threshold: float = 0.02):
"""Основная функция — проверяет одну позицию."""
page_text = await fetch_page_text(url)
current_price = extract_price_with_gpt(page_text, product_name)
if current_price is None:
print(f"[{product_name}] Не удалось извлечь цену")
return
last_price = get_last_price(url)
save_price(url, current_price, product_name)
if last_price is None:
print(f"[{product_name}] Первая запись: {current_price} ₽")
return
change = (current_price - last_price) / last_price
if abs(change) >= change_threshold:
direction = "выросла" if change > 0 else "упала"
message = (
f"📊 {product_name}\n"
f"Цена {direction}: {last_price:.0f} → {current_price:.0f} ₽ "
f"({change:+.1%})\n{url}"
)
await send_telegram_alert(message, bot_token, chat_id)
print(f"[{product_name}] Алерт отправлен: {message}")
# Список для мониторинга
WATCHLIST = [
{
"url": "https://example-shop.ru/product/widget-pro",
"name": "Widget Pro 500g"
},
{
"url": "https://another-store.ru/catalog/item/12345",
"name": "Аналог от конкурента"
},
]
async def run_all_checks(bot_token: str, chat_id: str):
for item in WATCHLIST:
await check_price(
url=item["url"],
product_name=item["name"],
bot_token=bot_token,
chat_id=chat_id
)
await asyncio.sleep(5) # пауза между запросами — вежливость и осторожность
async def main():
import os
bot_token = os.environ["TELEGRAM_BOT_TOKEN"]
chat_id = os.environ["TELEGRAM_CHAT_ID"]
init_db()
scheduler = AsyncIOScheduler()
scheduler.add_job(
run_all_checks,
"interval",
hours=4,
kwargs={"bot_token": bot_token, "chat_id": chat_id}
)
scheduler.start()
# Первый прогон сразу при запуске
await run_all_checks(bot_token, chat_id)
# Держим процесс живым
await asyncio.Event().wait()
if __name__ == "__main__":
asyncio.run(main())
Где я обжёгся
Первая проблема — блокировки. Несколько сайтов после третьего-четвёртого обращения начали отдавать капчу или 403. Решил частично: добавил случайные паузы между запросами и ротацию user-agent. Полностью это не помогает — если сайт серьёзно защищён, Playwright с обычными настройками не вытянет. Тут уже нужен либо платный прокси-сервис, либо официальный API магазина, если он есть.
С другой стороны, куда неожиданнее оказалась вторая проблема: GPT иногда галлюцинирует цены. Один раз вернул число из описания товара — объём упаковки в граммах, не цену. Добавил простую проверку: если извлечённое число меньше 10 или больше миллиона — скорее всего что-то пошло не так. Плюс логирую все ответы модели в отдельный файл, чтобы можно было потом посмотреть что происходило.
Третья проблема была совсем глупой. Запускал агента на домашнем компьютере, и он падал когда машина уходила в сон. Час убил прежде чем понял почему мониторинг молчал двое суток. Переехал на маленький VPS за 200 рублей в месяц — и всё.
Что получилось в итоге
Агент работает третью неделю. За это время поймал два реальных снижения цен у конкурентов — оба раза узнал раньше, чем увидел бы вручную. Расходы на OpenAI API за три недели — около 40 рублей: gpt-4o-mini стоит копейки на таких объёмах.
Самое ценное оказалось не в самих алертах, а в истории. Теперь видно как цены двигаются во времени, в какие дни конкуренты делают акции и насколько глубокие скидки. Это уже другой разговор — но данные для него собираются сами.
