Competitor price monitorin real time

pip install requests python-dotenv fastapi uvicorn

# .env
MARKUDOWN_API_KEY=sua_chave_aqui
WEBHOOK_URL=https://seu-servidor.com/webhook
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX/YYY/ZZZ
PRICE_DROP_THRESHOLD=0.05
INTERVAL_MINUTES=60

[
  {
    "id": "nike-air-max-270",
    "name": "Nike Air Max 270",
    "our_price": 699.90,
    "competitors": [
      {
        "store": "Concorrente A",
        "url": "https://concorrente-a.com.br/nike-air-max-270"
      },
      {
        "store": "Concorrente B",
        "url": "https://concorrente-b.com.br/tenis/nike-air-max-270"
      }
    ]
  },
  {
    "id": "adidas-ultraboost-22",
    "name": "Adidas Ultraboost 22",
    "our_price": 849.90,
    "competitors": [
      {
        "store": "Concorrente A",
        "url": "https://concorrente-a.com.br/adidas-ultraboost-22"
      }
    ]
  }
]

PRICE_SCHEMA = [
    {"name": "product_name",        "type": "String", "active": True},
    {"name": "current_price",       "type": "Number", "active": True},
    {"name": "original_price",      "type": "Number", "active": True},
    {"name": "discount_percentage", "type": "Number", "active": True},
    {"name": "availability",        "type": "String", "active": True},
]

"""
register.py — Registra um monitor por URL de concorrente via /api/monitor.
Execute uma vez. O MarkUDown cuida do agendamento — sem scheduler local.
"""
import json
import os
import requests
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

MARKUDOWN_URL = "https://api.scrapetechnology.com"
API_KEY = os.getenv("MARKUDOWN_API_KEY")
WEBHOOK_URL = os.getenv("WEBHOOK_URL")
INTERVAL_MINUTES = int(os.getenv("INTERVAL_MINUTES", "60"))

PRODUCTS_FILE = Path("products.json")
SUBSCRIPTIONS_FILE = Path("subscriptions.json")


def register_monitors():
    products = json.loads(PRODUCTS_FILE.read_text(encoding="utf-8"))
    subscriptions = {}

    for product in products:
        for competitor in product["competitors"]:
            url = competitor["url"]
            store = competitor["store"]

            print(f"Registrando monitor: {store} — {url}")

            resp = requests.post(
                f"{MARKUDOWN_URL}/api/monitor",
                headers={"X-API-KEY": API_KEY},
                json={
                    "url": url,
                    "callback_url": WEBHOOK_URL,
                    "interval_minutes": INTERVAL_MINUTES,
                    "main_content": True,
                },
                timeout=30,
            )
            resp.raise_for_status()
            data = resp.json()
            sub_id = data["subscription_id"]

            subscriptions[sub_id] = {
                "product_id": product["id"],
                "product_name": product["name"],
                "our_price": product["our_price"],
                "store": store,
                "url": url,
            }
            print(f"  OK — subscription_id: {sub_id}")

    SUBSCRIPTIONS_FILE.write_text(
        json.dumps(subscriptions, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"\n{len(subscriptions)} monitors registrados em subscriptions.json")


if __name__ == "__main__":
    register_monitors()

"""
webhook.py — Recebe notificações do MarkUDown e alerta quando preço cai.
Execute: uvicorn webhook:app --host 0.0.0.0 --port 8090
"""
import json
import os
import time
from datetime import datetime
from pathlib import Path

import requests
from fastapi import FastAPI, Request
from dotenv import load_dotenv

load_dotenv()

MARKUDOWN_URL = "https://api.scrapetechnology.com"
API_KEY = os.getenv("MARKUDOWN_API_KEY")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
PRICE_DROP_THRESHOLD = float(os.getenv("PRICE_DROP_THRESHOLD", "0.05"))

SUBSCRIPTIONS_FILE = Path("subscriptions.json")
PRICES_DB_FILE = Path("prices_db.json")

PRICE_SCHEMA = [
    {"name": "product_name",        "type": "String", "active": True},
    {"name": "current_price",       "type": "Number", "active": True},
    {"name": "original_price",      "type": "Number", "active": True},
    {"name": "discount_percentage", "type": "Number", "active": True},
    {"name": "availability",        "type": "String", "active": True},
]

app = FastAPI()


def load_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}


def save_json(path: Path, data: dict) -> None:
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def extract_price(url: str) -> dict | None:
    """Chama /api/extract e aguarda o job completar."""
    resp = requests.post(
        f"{MARKUDOWN_URL}/api/extract",
        headers={"X-API-KEY": API_KEY},
        json={
            "url": url,
            "extract_query": "nome do produto, preco atual, preco original, desconto e disponibilidade",
            "schema": PRICE_SCHEMA,
            "extraction_scope": "single_page",
        },
        timeout=30,
    )
    resp.raise_for_status()
    job_id = resp.json()["job_id"]

    for _ in range(40):
        time.sleep(3)
        result = requests.get(
            f"{MARKUDOWN_URL}/api/extract/{job_id}",
            headers={"X-API-KEY": API_KEY},
            timeout=10,
        ).json()
        if result.get("status") == "completed":
            items = result.get("items", [])
            if items:
                data = items[0].get("extracted_data", [{}])
                return data[0] if data else None
    return None


def send_slack_alert(meta: dict, current_price: float, previous_price: float) -> None:
    if not SLACK_WEBHOOK_URL:
        return
    drop_pct = round((previous_price - current_price) / previous_price * 100, 1)
    our_price = meta["our_price"]
    gap_pct = round((our_price - current_price) / our_price * 100, 1)
    lines = [
        ":rotating_light: *Queda de preco detectada!*",
        f"*Produto:* {meta['product_name']}",
        f"*Concorrente:* {meta['store']}",
        f"*Preco anterior:* R$ {previous_price:.2f}",
        f"*Novo preco:* R$ {current_price:.2f} ({drop_pct}% de queda)",
        f"*Nosso preco:* R$ {our_price:.2f} ({abs(gap_pct)}% {'acima' if gap_pct > 0 else 'abaixo'})",
        f"*Link:* {meta['url']}",
    ]
    requests.post(SLACK_WEBHOOK_URL, json={"text": "\n".join(lines)}, timeout=10)


@app.post("/webhook")
async def handle_monitor_event(request: Request):
    payload = await request.json()

    if not payload.get("changed"):
        return {"ok": True}

    subscription_id = payload.get("subscription_id")
    url = payload.get("url")

    subscriptions = load_json(SUBSCRIPTIONS_FILE)
    meta = subscriptions.get(subscription_id)
    if not meta:
        return {"ok": True}

    print(f"Mudanca detectada: {meta['store']} — {url}")

    extracted = extract_price(url)
    if not extracted:
        return {"ok": True}

    current_price = extracted.get("current_price")
    if not current_price:
        return {"ok": True}

    print(f"  Preco atual: R$ {current_price:.2f}")

    db = load_json(PRICES_DB_FILE)
    key = f"{meta['product_id']}::{meta['store']}"
    previous_price = db.get(key, {}).get("current_price")

    if previous_price and previous_price > 0:
        change_ratio = (previous_price - current_price) / previous_price
        if change_ratio >= PRICE_DROP_THRESHOLD:
            print(f"  ALERTA: R$ {previous_price:.2f} -> R$ {current_price:.2f}")
            send_slack_alert(meta, current_price, previous_price)

    db[key] = {
        **extracted,
        "url": url,
        "store": meta["store"],
        "product_id": meta["product_id"],
        "checked_at": datetime.now().isoformat(),
    }
    save_json(PRICES_DB_FILE, db)
    return {"ok": True}

# 1. Instalar dependências
pip install -r requirements.txt

# 2. Registrar os monitors (rode uma vez)
python register.py

# 3. Subir o servidor de webhook
uvicorn webhook:app --host 0.0.0.0 --port 8090