Создание агента для отслеживания цен с помощью Crawl API

TL;DR

Агент по отслеживанию цен — это небольшая цепочка: чтение одобренных URL продуктов, извлечение конкретного поля цены, хранение наблюдений, сравнение с историей и оповещение о значительных изменениях.
Nstdata Crawl обрабатывает загрузку страниц и может возвращать Markdown или HTML. Ваше приложение должно реализовать сопоставление продуктов, извлечение цен, планирование, сохранение и уведомления.
Отклоняйте отсутствующие цены, изменения валюты и несовпадающие варианты перед сравнением значений. Храните исходный URL и временную метку с каждым наблюдением.
Пример, который можно запустить, использует текстовый маркер, специфичный для сайта, и SQLite, чтобы его ограничения были видны; адаптируйте парсер к фактической разметке целевого сайта.

Предположим, что розничная команда хочет знать, когда цена на публичной странице продукта изменится. Cron-задание, которое загружает HTML, может работать на статической странице, но производственный монитор также требует рендеринга, проверки извлечения, истории, дедупликации и политики оповещения. Nstdata Crawl может предоставить содержимое страницы; решения о мониторинге принадлежат приложению. Этот учебник использует иллюстративные данные о продуктах и не делает обязательств относительно живых цен конкурентов.

Что такое агент по отслеживанию цен?

Агент по отслеживанию цен превращает наблюдения в небольшое решение: "Изменился ли тот же продукт, в той же валюте и варианте, достаточно, чтобы уведомить рецензента?" Он отличается от базового скрипта для сбора данных тем, что сохраняет состояние, применяет правила валидации и объясняет свои уведомления. Он не может вывести коммерческое намерение конкурента из изменения цены. Более низкое отображаемое значение может быть акцией, альтернативным вариантом, предложением, специфичным для региона, или ошибкой анализа.

Для каждого продукта определите стабильный product_id, ожидаемое имя или SKU, исходный URL, валюту и интервал проверки. Не объединяйте вариации продуктов без предупреждения. Если целевой сайт предоставляет лицензированную ленту или официальный API, который поставляет необходимые данные, оцените этот путь в первую очередь.

import os
import re
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation

from nstdata_ai_crawl import Format, NstDataClient, ScrapeRequestDto

DB_PATH = "prices.sqlite3"
PRODUCTS = [
    {
        "product_id": "widget-standard",
        "product_name": "Стандартный виджет",
        "source_url": "https://shop.example.com/products/widget",
        "currency": "USD",
    }
]


def fetch_markdown(client: NstDataClient, url: str) -> str:
    result = client.submit_scrape_task_sync(
        ScrapeRequestDto(
            url=url,
            formats=[Format.MARKDOWN],
            onlyMainContent=True,
            timeout=60000,
        )
    )
    body = result.data.get_markdown()
    if not body or not body.strip():
        raise ValueError("Нет доступного содержимого страницы продукта")
    return body


def parse_price(markdown: str, currency: str) -> Decimal:
    # Замените этот специфичный для цели маркер на протестированный парсер для вашего сайта.
    pattern = rf"(?m)^Цена:\s*{re.escape(currency)}\s+([0-9]+(?:\.[0-9]{{2}})?)\s*$"
    matches = re.findall(pattern, markdown)
    if len(matches) != 1:
        raise ValueError(f"Ожидался один маркер цены {currency}, найдено {len(matches)}")
    try:
        amount = Decimal(matches[0])
    except InvalidOperation as exc:
        raise ValueError("Недопустимая десятичная цена") from exc
    if amount <= 0:
        raise ValueError("Цена должна быть положительной")
    return amount


def init_db(db: sqlite3.Connection) -> None:
    db.execute("""
        CREATE TABLE IF NOT EXISTS observations (
            id INTEGER PRIMARY KEY,
            product_id TEXT NOT NULL,
            observed_price TEXT,
            currency TEXT NOT NULL,
            observed_at TEXT NOT NULL,
            source_url TEXT NOT NULL,
            extraction_status TEXT NOT NULL
        )
    """)
    db.commit()


def previous_price(db: sqlite3.Connection, product_id: str, currency: str):
    row = db.execute(
        """SELECT observed_price FROM observations
           WHERE product_id = ? AND currency = ? AND extraction_status = 'ok'
           ORDER BY id DESC LIMIT 1""",
        (product_id, currency),
    ).fetchone()
    return Decimal(row[0]) if row else None


def save(db, product, observed_at, status, price=None):
    db.execute(
        """INSERT INTO observations
           (product_id, observed_price, currency, observed_at, source_url, extraction_status)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (
            product["product_id"],
            str(price) if price is not None else None,
            product["currency"],
            observed_at,
            product["source_url"],
            status,
        ),
    )
    db.commit()


def main() -> None:
    with sqlite3.connect(DB_PATH) as db:
        init_db(db)
        with NstDataClient(os.environ["NSTDATA_API_TOKEN"]) as client:
            for product in PRODUCTS:
                observed_at = datetime.now(timezone.utc).isoformat()
                try:
                    markdown = fetch_markdown(client, product["source_url"])
                    price = parse_price(markdown, product["currency"])
                    baseline = previous_price(
                        db, product["product_id"], product["currency"]
                    )
save(db, product, observed_at, "ok", price)
                    if baseline is None:
                        print(f"Базовый уровень записан: {product['product_id']} {price}")
                        continue
                    delta = price - baseline
                    pct = abs(delta / baseline) * 100 if baseline else Decimal("0")
                    if abs(delta) >= Decimal("5") and pct >= Decimal("3"):
                        print(
                            f"ALERT {product['product_id']}: {baseline} -> {price} "
                            f"{product['currency']} ({pct:.2f}%); "
                            f"source={product['source_url']} at={observed_at}"
                        )
                except Exception as exc:
                    save(db, product, observed_at, "error")
                    print(f"ОШИБКА {product['product_id']}: {exc}")


if __name__ == "__main__":
    main()

Создание агента мониторинга цен с помощью Crawl API

TL;DR

Что такое агент по отслеживанию цен?

Популярные статьи

Создание агента мониторинга цен с помощью Crawl API

Crawl + LLM: Суммируйте любой веб-сайт по запросу

Как работает агент по отслеживанию цен

Crawl API и сбор веб-данных

Извлечение цены и историческое отслеживание

Решения агента и уведомления

Пошаговая инструкция: создайте агент по отслеживанию цен с помощью Python

Шаг 1: Настройте проект и список URL продуктов

Шаг 2: Загружайте страницы продуктов с помощью Nstdata Crawl

Пример: Соберите страницу продукта

Шаг 3: Извлечение и нормализация цен

Шаг 4: Хранение истории цен в SQLite

Шаг 5: Обнаружение значительных изменений цен

Шаг 6: Запланируйте проверки и генерируйте уведомления

Почему стоит использовать Nstdata Crawl для мониторинга цен?

Основные преимущества Nstdata Crawl

Пример интеграции Python SDK

Тестирование, затраты и распространенные проблемы

Распространенные ошибки и устранение неполадок

Как измерить качество мониторинга

Часто задаваемые вопросы

Как я могу создать агент мониторинга цен?

Может ли API Crawl автоматически извлекать цены?

Как часто следует проверять цены?

Как следует хранить исторические цены продуктов?