构建 AI 商品监控工具

python --version

mkdir product-monitor && cd product-monitor
python -m venv .venv
source .venv/bin/activate
pip install crawlbase openai

export CRAWLBASE_TOKEN="your_crawlbase_token"
export LLM_API_KEY="your_llm_api_key"

import os
from crawlbase import ScraperAPI

scraper = ScraperAPI({"token": os.environ["CRAWLBASE_TOKEN"]})

def collect_product(url):
    # 'amazon-product-details' is one of the maintained parsers.
    response = scraper.get(url, {"scraper": "amazon-product-details"})
    body = response["json"]["body"]

    reading = {
        "url": url,
        "name": body.get("name"),
        "price": body.get("rawPrice"),
        "currency": body.get("currency"),
        "in_stock": body.get("inStock"),
        "rating": body.get("rating"),
    }

    if reading["price"] is None or reading["name"] is None:
        raise ValueError(f"Parse returned no price/name for {url}")

    return reading

import sqlite3
from datetime import datetime, timezone

DB = "readings.db"

def init_db():
    con = sqlite3.connect(DB)
    con.execute(
        """CREATE TABLE IF NOT EXISTS readings (
            url TEXT, name TEXT, price REAL, currency TEXT,
            in_stock INTEGER, rating REAL, taken_at TEXT)"""
    )
    con.commit()
    con.close()

def save_reading(r):
    con = sqlite3.connect(DB)
    con.execute(
        "INSERT INTO readings VALUES (?, ?, ?, ?, ?, ?, ?)",
        (r["url"], r["name"], r["price"], r["currency"],
         int(bool(r["in_stock"])), r["rating"],
         datetime.now(timezone.utc).isoformat()),
    )
    con.commit()
    con.close()

def last_two(url):
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    rows = con.execute(
        "SELECT * FROM readings WHERE url = ? ORDER BY taken_at DESC LIMIT 2",
        (url,),
    ).fetchall()
    con.close()
    return [dict(row) for row in rows]

PRICE_THRESHOLD = 0.03  # 3% move counts as meaningful

def detect_changes(current, previous):
    changes = []

    old_price, new_price = previous["price"], current["price"]
    if old_price and new_price:
        delta = (new_price - old_price) / old_price
        if abs(delta) >= PRICE_THRESHOLD:
            changes.append({
                "field": "price",
                "old": old_price,
                "new": new_price,
                "pct": round(delta * 100, 1),
            })

    if previous["in_stock"] != current["in_stock"]:
        changes.append({
            "field": "in_stock",
            "old": bool(previous["in_stock"]),
            "new": bool(current["in_stock"]),
        })

    if previous["rating"] and current["rating"]:
        if abs(current["rating"] - previous["rating"]) >= 0.2:
            changes.append({
                "field": "rating",
                "old": previous["rating"],
                "new": current["rating"],
            })

    return changes

import os
import json
from openai import OpenAI

client = OpenAI(api_key=os.environ["LLM_API_KEY"])

def summarize_changes(product_name, changes):
    prompt = (
        f"Product: {product_name}\n"
        f"Detected changes (JSON): {json.dumps(changes)}\n\n"
        "Write one short sentence summarizing what changed. "
        "State only what the data shows. Do not speculate or "
        "add numbers that are not present."
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system",
             "content": "You summarize product data changes factually."},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content.strip()

def send_alert(message):
    # Swap this for Slack, email, or a webhook in production.
    print(f"[ALERT] {message}")

from collect import collect_product
from store import init_db, save_reading, last_two
from detect import detect_changes
from alert import summarize_changes, send_alert

WATCHLIST = [
    "https://www.example-store.com/product/abc",
    "https://www.example-store.com/product/xyz",
]

def run_once():
    init_db()
    for url in WATCHLIST:
        try:
            reading = collect_product(url)
        except Exception as exc:
            print(f"Skipped {url}: {exc}")
            continue

        save_reading(reading)
        history = last_two(url)
        if len(history) < 2:
            continue  # first reading, nothing to compare

        current, previous = history[0], history[1]
        changes = detect_changes(current, previous)
        if changes:
            summary = summarize_changes(reading["name"], changes)
            send_alert(summary)

if __name__ == "__main__":
    run_once()

python monitor.py
# [ALERT] The price of "Acme Widget" dropped 7.4% from $129 to $119, and it is back in stock.

# crontab -e, then add (runs at the top of every hour):
0 * * * * cd /path/to/product-monitor && .venv/bin/python monitor.py >> monitor.log 2>&1