pip install pangolinfo-api

POST /v1/amazon/scrape HTTP/1.1
Host: api.pangolinfo.com
Authorization: Bearer pgo_your_api_key_here
Content-Type: application/json

import requests
import json

API_KEY = "pgo_your_api_key_here"
BASE_URL = "https://api.pangolinfo.com/v1"

def get_product_details(site: str, asins: list):
    url = f"{BASE_URL}/amazon/product"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "site": site,
        "asins": asins,
        "output_format": "json",
        "include_variants": True,
        "include_bsr": True,
        "include_seller_info": True
    }
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()

result = get_product_details("amazon.com", ["B08N5WRWNW", "B08N5M7S6K", "B09V3KXJPB"])
for product in result.get("products", []):
    print(f"ASIN: {product['asin']} | Price: {product['price']['current']} | Rating: {product['rating']['average']}")

{
  "status": "success",
  "request_id": "req_abc123def456",
  "products": [{
    "asin": "B08N5WRWNW",
    "title": "Apple AirPods (3rd Generation)",
    "brand": "Apple",
    "price": {"current": 149.99, "currency": "USD", "list_price": 179.00},
    "rating": {"average": 4.5, "count": 84532},
    "bsr": {"rank": 42, "category": "Electronics"},
    "images": {"main": "https://m.media-amazon.com/images/..."},
    "features": ["Spatial audio", "Adaptive EQ"],
    "variants": [{"asin": "B08N5WRWNW", "attribute": "Color", "value": "White"}],
    "seller": {"name": "Amazon.com", "is_amazon": true, "fulfillment": "FBA"},
    "availability": {"status": "In Stock"},
    "collected_at": "2026-05-25T08:30:00Z"
  }],
  "credits_used": 3,
  "credits_remaining": 997
}

def search_keywords(site: str, keyword: str, pages: int = 1):
    url = f"{BASE_URL}/amazon/search"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    all_results = []
    for page in range(1, pages + 1):
        payload = {
            "site": site, "keyword": keyword, "page": page,
            "sort_by": "relevance", "output_format": "json",
            "include_sponsored": True, "include_bsr": True
        }
        response = requests.post(url, headers=headers, json=payload, timeout=45)
        data = response.json()
        all_results.extend(data.get("products", []))
    return all_results

products = search_keywords("amazon.com", "wireless earbuds", pages=2)
print(f"Total products: {len(products)}")

def get_best_sellers(site: str, node_id: str, pages: int = 1):
    url = f"{BASE_URL}/amazon/bestsellers"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {"site": site, "node_id": node_id, "pages": pages, "output_format": "json"}
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    return response.json()

result = get_best_sellers("amazon.com", "172282", pages=2)
for item in result.get("products", []):
    print(f"#{item['rank']} {item['title'][:50]}... | ${item['price']['current']}")

def get_reviews(site: str, asin: str, pages: int = 3):
    url = f"{BASE_URL}/amazon/reviews"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {"site": site, "asin": asin, "pages": pages, "sort_by": "recent", "include_images": True}
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    return response.json()

reviews = get_reviews("amazon.com", "B08N5WRWNW", pages=2)
for review in reviews.get("reviews", [])[:5]:
    print(f"⭐ {review['rating']}/5 | {review['title']} | by {review['author']}")

def submit_async_task(task_type: str, tasks: list):
    url = f"{BASE_URL}/amazon/tasks"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {
        "task_type": task_type,
        "tasks": tasks,
        "webhook_url": "https://your-domain.com/webhook/pangolinfo",
        "webhook_secret": "your_webhook_secret",
        "priority": "normal",
        "output_format": "json"
    }
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    return response.json()

asins = ["B08N5WRWNW", "B08N5M7S6K", ...]  # 100 ASINs
tasks = [{"site": "amazon.com", "asin": asin} for asin in asins]
result = submit_async_task("product_details", tasks)
print(f"Batch ID: {result['batch_id']} | Tasks: {result['task_count']}")

from flask import Flask, request, jsonify
import hmac, hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route('/webhook/pangolinfo', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Pangolinfo-Signature')
    expected = hmac.new(WEBHOOK_SECRET.encode(), request.data, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401
    
    data = request.json
    if data['event'] == 'task.completed':
        save_to_database(data['task_id'], data['result'])
    elif data['event'] == 'task.failed':
        log_failure(data['task_id'], data['error'])
    return jsonify({"status": "ok"}), 200

def check_task_status(batch_id: str):
    url = f"{BASE_URL}/amazon/tasks/{batch_id}"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = requests.get(url, headers=headers, timeout=30)
    return response.json()

batch_id = "batch_abc123"
while True:
    status = check_task_status(batch_id)
    print(f"Progress: {status['completed']}/{status['total']}")
    if status['status'] in ['completed', 'failed']:
        break
    time.sleep(10)

import psycopg2
from datetime import datetime

def save_products_to_db(products: list, db_config: dict):
    conn = psycopg2.connect(**db_config)
    cursor = conn.cursor()
    insert_sql = """
        INSERT INTO amazon_products (asin, site, title, brand, current_price, currency,
            rating_avg, rating_count, bsr_rank, collected_at)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON CONFLICT (asin, site) DO UPDATE SET
            title = EXCLUDED.title, current_price = EXCLUDED.current_price,
            rating_avg = EXCLUDED.rating_avg, collected_at = EXCLUDED.collected_at
    """
    for product in products:
        cursor.execute(insert_sql, (
            product['asin'], product['site'], product['title'],
            product.get('brand', ''), product.get('price', {}).get('current'),
            product.get('price', {}).get('currency', 'USD'),
            product.get('rating', {}).get('average'),
            product.get('rating', {}).get('count'),
            product.get('bsr', {}).get('rank'), datetime.now()
        ))
    conn.commit()
    cursor.close()
    conn.close()

import redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def publish_to_queue(channel: str, data: dict):
    redis_client.xadd(channel, {"data": json.dumps(data)})

@app.route('/webhook/pangolinfo', methods=['POST'])
def handle_webhook():
    data = request.json
    if data['event'] == 'task.completed':
        publish_to_queue("amazon:data:products", data['result'])
    return jsonify({"status": "ok"}), 200

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract_amazon_data(**context):
    batch_id = submit_async_task("product_details", task_list)
    return wait_for_completion(batch_id)

def transform_data(**context):
    raw_data = context['ti'].xcom_pull(task_ids='extract')
    return clean_data(raw_data)

def load_to_warehouse(**context):
    data = context['ti'].xcom_pull(task_ids='transform')
    bulk_insert(data)

with DAG('amazon_data_pipeline', start_date=datetime(2026, 1, 1),
     schedule_interval='0 6 * * *', catchup=False) as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract_amazon_data)
    transform = PythonOperator(task_id='transform', python_callable=transform_data)
    load = PythonOperator(task_id='load', python_callable=load_to_warehouse)
    extract >> transform >> load

class AmazonERPConnector:
    def __init__(self, api_key: str, erp_db):
        self.api_key = api_key
        self.erp = erp_db
        self.base_url = "https://api.pangolinfo.com/v1"
    
    def sync_product_catalog(self, site: str, category_node: str):
        products = self._fetch_category_products(site, category_node)
        erp_products = [{
            "sku": f"AMZ-{p['asin']}", "name": p['title'],
            "category": p['bsr']['category'], "sale_price": p['price']['current'],
            "supplier": p['brand'], "source_platform": "amazon"
        } for p in products]
        self.erp.bulk_upsert_products(erp_products)
        return len(erp_products)
    
    def update_competitive_pricing(self, site: str, sku_list: list):
        asins = [sku.replace("AMZ-", "") for sku in sku_list]
        products = self._fetch_products(site, asins)
        for product in products:
            self.erp.update_price_monitor(
                sku=f"AMZ-{product['asin']}",
                competitor_price=product['price']['current']
            )

{
  "mcpServers": {
    "pangolinfo-amazon": {
      "command": "npx",
      "args": ["-y", "@pangolinfo/amazon-scraper-mcp@latest"],
      "env": {"PANGOLINFO_API_KEY": "pgo_your_api_key_here"}
    }
  }
}

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class AmazonSearchInput(BaseModel):
    keyword: str = Field(description="Amazon search keyword")
    site: str = Field(default="amazon.com")
    pages: int = Field(default=1, ge=1, le=5)

class AmazonSearchTool(BaseTool):
    name = "amazon_search"
    description = "Search Amazon products and return structured data"
    args_schema = AmazonSearchInput
    def _run(self, keyword: str, site: str = "amazon.com", pages: int = 1):
        response = requests.post("https://api.pangolinfo.com/v1/amazon/search",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"site": site, "keyword": keyword, "pages": pages})
        return response.json()

from langchain.agents import initialize_agent, Tool
tools = [Tool(name="amazon_search", func=AmazonSearchTool()._run)]
agent = initialize_agent(tools, llm, agent="zero-shot-react-description")
result = agent.run("Find top 10 wireless earbuds under $100 with 4.5+ rating")

import functools, redis
from datetime import timedelta
cache = redis.Redis()

def cached_api_call(ttl_seconds: int = 300):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"pangolinfo:{func.__name__}:{hash(str(args)+str(kwargs))}"
            cached = cache.get(cache_key)
            if cached: return json.loads(cached)
            result = func(*args, **kwargs)
            cache.setex(cache_key, timedelta(seconds=ttl_seconds), json.dumps(result))
            return result
        return wrapper
    return decorator

import time
from functools import wraps

def rate_limit(max_per_minute: int = 100):
    min_interval = 60.0 / max_per_minute
    last_call_time = {}
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = func.__name__
            now = time.time()
            if key in last_call_time:
                elapsed = now - last_call_time[key]
                if elapsed < min_interval: time.sleep(min_interval - elapsed)
            last_call_time[key] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

def validate_product_data(product: dict) -> tuple[bool, list]:
    errors = []
    if not product.get('asin'): errors.append("Missing ASIN")
    if not product.get('title') or len(product['title']) < 5: errors.append("Invalid title")
    price = product.get('price', {}).get('current')
    if price is None or price <= 0: errors.append("Invalid price")
    return len(errors) == 0, errors