Un agente de alerta de precios de Amazon busca productos rastreados diariamente a través de la API de Scavio con la plataforma configurada en Amazon, compara los precios actuales con las líneas de base almacenadas y envía una alerta cuando un precio cae más de un umbral configurable.
Requisitos previos
- Python 3.9+
- Clave API de Scavio
- SQLite (stdlib)
Guia paso a paso
Paso 1: Definir productos rastreados
Seguimiento de productos mediante consulta de búsqueda. El agente encuentra el listado coincidente todos los días.
Python
TRACKED_PRODUCTS = [
{"query": "Sony WH-1000XM5 Wireless Headphones", "max_budget": 300},
{"query": "Kindle Paperwhite 16GB", "max_budget": 150},
{"query": "Apple AirPods Pro 2nd generation", "max_budget": 200}
]Paso 2: Busque en Amazon y obtenga el precio actual
Busque cada producto y extraiga el precio actual más bajo.
Python
import requests, re
API_KEY = "your-scavio-api-key"
def get_amazon_price(query: str) -> dict | None:
r = requests.post(
"https://api.scavio.dev/api/v1/search",
json={"query": query, "platform": "amazon", "num_results": 5},
headers={"x-api-key": API_KEY}, timeout=20
)
r.raise_for_status()
products = r.json().get("organic_results", [])
if not products:
return None
def parse_price(s):
m = re.search(r"[\d.]+", (s or "").replace(",",""))
return float(m.group()) if m else None
priced = [(parse_price(p.get("price","")), p) for p in products]
priced = [(price, p) for price, p in priced if price]
if not priced:
return None
price, product = min(priced, key=lambda x: x[0])
return {"title": product.get("title"), "price": price, "url": product.get("link"), "rating": product.get("rating")}Paso 3: Almacenar la línea base y detectar caídas
En la primera ejecución, almacene el precio base. En ejecuciones posteriores, compare y alerte.
Python
import sqlite3
from datetime import date
def init_db(path="amazon_alerts.db"):
conn = sqlite3.connect(path)
conn.execute("""
CREATE TABLE IF NOT EXISTS prices (
date TEXT, query TEXT, title TEXT, price REAL, url TEXT,
PRIMARY KEY (date, query)
)
""")
conn.commit()
return conn
def run_alerts(tracked: list, drop_threshold_pct: float = 10.0):
conn = init_db()
today = str(date.today())
for item in tracked:
query = item["query"]
data = get_amazon_price(query)
if not data:
continue
conn.execute("INSERT OR REPLACE INTO prices VALUES (?,?,?,?,?)",
(today, query, data["title"], data["price"], data["url"]))
conn.commit()
prev = conn.execute(
"SELECT price FROM prices WHERE query=? AND date < ? ORDER BY date DESC LIMIT 1",
(query, today)
).fetchone()
if prev and prev[0] > 0:
drop_pct = (prev[0] - data["price"]) / prev[0] * 100
if drop_pct >= drop_threshold_pct:
print(f"ALERT: {data['title'][:50]}")
print(f" ${prev[0]:.2f} -> ${data['price']:.2f} (-{drop_pct:.1f}%)")
print(f" {data['url']}")
print(f"Tracked: {query[:50]} | ${data['price']:.2f}")
run_alerts(TRACKED_PRODUCTS, drop_threshold_pct=10.0)Ejemplo en Python
Python
import requests
import sqlite3
import re
from datetime import date
API_KEY = "your-scavio-api-key"
TRACKED = [
"Sony WH-1000XM5 Wireless Headphones",
"Kindle Paperwhite 16GB",
"Apple AirPods Pro 2nd generation"
]
def get_price(query):
r = requests.post("https://api.scavio.dev/api/v1/search",
json={"query": query, "platform": "amazon", "num_results": 5},
headers={"x-api-key": API_KEY}, timeout=20)
r.raise_for_status()
products = r.json().get("organic_results", [])
def parse(s):
m = re.search(r"[\d.]+", (s or "").replace(",",""))
return float(m.group()) if m else None
priced = [(parse(p.get("price","")), p) for p in products]
priced = [(pr, p) for pr, p in priced if pr]
if not priced: return None
pr, p = min(priced, key=lambda x: x[0])
return {"title": p.get("title"), "price": pr, "url": p.get("link")}
def init_db():
conn = sqlite3.connect("amazon_alerts.db")
conn.execute("CREATE TABLE IF NOT EXISTS prices (date TEXT, query TEXT, title TEXT, price REAL, url TEXT, PRIMARY KEY (date, query))")
conn.commit()
return conn
def run():
conn = init_db()
today = str(date.today())
for query in TRACKED:
d = get_price(query)
if not d: continue
conn.execute("INSERT OR REPLACE INTO prices VALUES (?,?,?,?,?)", (today, query, d["title"], d["price"], d["url"]))
conn.commit()
prev = conn.execute("SELECT price FROM prices WHERE query=? AND date<? ORDER BY date DESC LIMIT 1", (query, today)).fetchone()
if prev and prev[0] > 0:
drop = (prev[0] - d["price"]) / prev[0] * 100
if drop >= 10:
print(f"PRICE DROP: {d['title'][:60]}\n ${prev[0]:.2f} -> ${d['price']:.2f} (-{drop:.1f}%)\n {d['url']}")
print(f"OK | {query[:50]} | ${d['price']:.2f}")
if __name__ == "__main__":
run()Ejemplo en JavaScript
JavaScript
const API_KEY = 'your-scavio-api-key';
async function getAmazonPrice(query) {
const res = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'x-api-key': API_KEY },
body: JSON.stringify({ query, platform: 'amazon', num_results: 5 })
});
const data = await res.json();
const products = data.organic_results ?? [];
const parsePrice = s => { const m = s?.replace(/,/g,'').match(/[\d.]+/); return m ? parseFloat(m[0]) : null; };
const priced = products.map(p => ({ price: parsePrice(p.price), ...p })).filter(p => p.price);
if (!priced.length) return null;
const cheapest = priced.reduce((a, b) => a.price < b.price ? a : b);
return { title: cheapest.title, price: cheapest.price, url: cheapest.link };
}
for (const q of ['Sony WH-1000XM5', 'Kindle Paperwhite']) {
const d = await getAmazonPrice(q);
if (d) console.log(`${d.title?.slice(0,50)} | $${d.price}`);
}Salida esperada
JSON
OK | Sony WH-1000XM5 Wireless Headphones | $299.99
PRICE DROP: Kindle Paperwhite 16GB - International Version
$139.99 -> $109.99 (-21.4%)
https://amazon.com/dp/B0CFPJYX...
OK | Apple AirPods Pro 2nd generation | $189.99