Xây dựng một tác nhân theo dõi giá với Crawl API

TL;DR

Một tác nhân theo dõi giá là một quy trình nhỏ: đọc các URL sản phẩm đã được phê duyệt, phân tích một trường giá cụ thể, lưu trữ các quan sát, so sánh với lịch sử và cảnh báo về những thay đổi có ý nghĩa.
Nstdata Crawl xử lý việc truy xuất trang và có thể trả về Markdown hoặc HTML. Ứng dụng của bạn phải thực hiện việc đối sánh sản phẩm, phân tích giá, lập lịch, lưu trữ và thông báo.
Từ chối các giá thiếu, thay đổi tiền tệ và biến thể không khớp trước khi so sánh giá trị. Giữ lại URL gốc và thời gian cho mỗi quan sát.
Ví dụ có thể chạy sử dụng một dấu hiệu văn bản cụ thể cho trang và SQLite nên các giới hạn của nó là rõ ràng; điều chỉnh bộ phân tích cho thực tế đánh dấu của từng trang mục tiêu.

Giả sử một nhóm bán lẻ muốn biết khi nào một trang sản phẩm công khai thay đổi giá. Một cron job tải HTML có thể hoạt động trên một trang tĩnh, nhưng một trình giám sát sản xuất cũng cần kết xuất, kiểm tra trích xuất, lịch sử, loại bỏ trùng lặp và chính sách cảnh báo. Nstdata Crawl có thể cung cấp nội dung trang; các quyết định giám sát thuộc về ứng dụng. Hướng dẫn này sử dụng dữ liệu sản phẩm minh họa và không đưa ra tuyên bố nào về giá cạnh tranh trực tiếp.

Tác Nhân Theo Dõi Giá Là Gì?

Một tác nhân theo dõi giá biến các quan sát thành một quyết định nhỏ: “Liệu sản phẩm giống nhau, trong cùng một loại tiền tệ và biến thể, đã chuyển động đủ để thông báo cho người xem?” Nó khác với một kịch bản quét cơ bản bằng cách duy trì trạng thái, áp dụng các quy tắc xác thực và giải thích các cảnh báo của nó. Nó không thể suy ra ý định thương mại của một đối thủ từ việc thay đổi giá. Một giá trị hiển thị thấp hơn có thể là một chương trình khuyến mãi, một biến thể thay thế, một đề nghị theo khu vực, hoặc một lỗi phân tích.

Đối với mỗi sản phẩm, định nghĩa một product_id ổn định, tên hoặc SKU mong đợi, URL gốc, tiền tệ, và khoảng thời gian kiểm tra. Không hợp nhất lặng lẽ các biến thể sản phẩm. Nếu trang mục tiêu cung cấp một nguồn cấp dữ liệu được cấp phép hoặc API chính thức cung cấp dữ liệu bạn cần, hãy đánh giá con đường đó trước.

Cách Tác Nhân Theo Dõi Giá Hoạt Động

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": "Widget Standard",
        "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("No usable product page content")
    return body


def parse_price(markdown: str, currency: str) -> Decimal:
    # Replace this target-specific marker with a tested parser for your site.
    pattern = rf"(?m)^Price:\s*{re.escape(currency)}\s+([0-9]+(?:\.[0-9]{{2}})?)\s*$"
    matches = re.findall(pattern, markdown)
    if len(matches) != 1:
        raise ValueError(f"Expected one {currency} price marker, found {len(matches)}")
    try:
        amount = Decimal(matches[0])
    except InvalidOperation as exc:
        raise ValueError("Invalid decimal price") from exc
    if amount <= 0:
        raise ValueError("Price must be positive")
    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"Baseline recorded: {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"ERROR {product['product_id']}: {exc}")


if __name__ == "__main__":
    main()

Xây dựng một tác nhân theo dõi giá với Crawl API

TL;DR

Tác Nhân Theo Dõi Giá Là Gì?

Cách Tác Nhân Theo Dõi Giá Hoạt Động

Bai viet pho bien

Xây dựng một tác nhân theo dõi giá với Crawl API

Crawl + LLM: Tóm tắt bất kỳ trang web nào theo yêu cầu

Crawl API và Thu Thập Dữ Liệu Web

Trích Xuất Giá và Theo Dõi Lịch Sử

Quyết Định và Cảnh Báo của Tác Nhân

Các Bước: Xây Dựng Một Tác Nhân Theo Dõi Giá Với Python

Bước 1: Thiết Lập Dự Án và Danh Sách URL Sản Phẩm

Bước 2: Lấy Trang Sản Phẩm Với Nstdata Crawl

Ví dụ: Thu Thập Một Trang Sản Phẩm

Bước 3: Trích xuất và Chuẩn hóa Giá

Bước 4: Lưu Lịch sử Giá trong SQLite

Bước 5: Phát hiện Thay đổi Giá Có Ý Nghĩa

Bước 6: Lập Lịch Kiểm Tra và Tạo Cảnh Báo

Tại sao sử dụng Nstdata Crawl để theo dõi giá?

Những lợi thế chính của Nstdata Crawl

Ví dụ tích hợp SDK Python

Kiểm tra, Chi phí và Các Vấn đề Thường Gặp

Các Lỗi Thường Gặp và Khắc Phục

Cách Đo Chất Lượng Giám Sát

Câu hỏi thường gặp

Làm thế nào để tôi xây dựng một tác nhân giám sát giá?

Liệu một API Crawl có thể tự động trích xuất giá không?

Giá nên được kiểm tra bao lâu một lần?

Nên lưu trữ giá sản phẩm lịch sử như thế nào?