亚马逊 Hot New Releases 数据完整指南：实时新品榜单采集与选品应用

import requests
import json
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

API_ENDPOINT = "https://api.pangolinfo.com/v1/amazon/new-releases"
API_KEY = "your_api_key_here"  # tool.pangolinfo.com 控制台获取

# 目标监控类目（node_id 对应 Amazon 类目节点）
TARGET_CATEGORIES = {
    "home_kitchen":   "1055398",   # Home & Kitchen
    "pet_supplies":   "2619533011",# Pet Supplies
    "sports_outdoor": "3375251",   # Sports & Outdoors
    "tools_home":     "228013",    # Tools & Home Improvement
    "kitchen_dining": "284507",    # Kitchen & Dining
}

def fetch_new_releases(category_name: str, node_id: str, country: str = "US") -> dict:
    """
    采集指定类目的 Hot New Releases 榜单数据

    Args:
        category_name: 类目名称（用于日志标记）
        node_id:       Amazon 类目节点 ID
        country:       站点代码，支持 US/UK/DE/JP/CA

    Returns:
        dict: 含排名列表和采集元信息的结构化 JSON
    """
    resp = requests.post(
        API_ENDPOINT,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "country":  country,
            "node_id":  node_id,
            "list_type": "hot_new_releases",  # 或 "best_sellers" / "movers_shakers"
            "fields": [
                "rank", "asin", "title", "brand", "price",
                "bsr", "rating", "review_count",
                "date_first_available", "category_path"
            ]
        },
        timeout=30
    )
    resp.raise_for_status()
    data = resp.json()
    print(f"[{category_name}] 获取 {len(data.get('items', []))} 个新品上榜商品")
    return {"category": category_name, "node_id": node_id, "data": data}

def batch_fetch_all_categories(max_workers: int = 5) -> list:
    """并发采集所有目标类目，提升效率"""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(fetch_new_releases, name, node_id): name
            for name, node_id in TARGET_CATEGORIES.items()
        }
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"采集失败 [{futures[future]}]: {e}")
    return results

def find_fast_rising_new_products(results: list, min_reviews: int = 0, max_reviews: int = 50) -> list:
    """
    筛选"评论少但已上榜"的早期高潜力新品
    这类商品上量快但口碑积累刚开始，是竞品跟进或差异化参考的最佳目标

    Args:
        min_reviews: 最少评论数（默认0）
        max_reviews: 最多评论数（筛选早期新品）
    """
    early_stage = []
    for result in results:
        for item in result["data"].get("items", []):
            review_count = item.get("review_count", 0)
            rank = item.get("rank", 999)
            if min_reviews <= review_count <= max_reviews and rank <= 20:
                early_stage.append({
                    "category": result["category"],
                    "rank":     rank,
                    "asin":     item["asin"],
                    "title":    item["title"][:50],
                    "reviews":  review_count,
                    "rating":   item.get("rating", "N/A"),
                    "price":    item.get("price", "N/A"),
                    "launched": item.get("date_first_available", "N/A")
                })

    # 按榜单排名升序排列
    early_stage.sort(key=lambda x: x["rank"])
    return early_stage

if __name__ == "__main__":
    print(f"[{datetime.now():%H:%M:%S}] 开始批量采集 Hot New Releases 数据...
")

    all_results = batch_fetch_all_categories()

    # 筛选早期高潜力新品（进前20但评论数低于50）
    hot_early = find_fast_rising_new_products(all_results, max_reviews=50)

    print(f"
发现 {len(hot_early)} 个早期高潜力新品（榜单前20 + 评论 <50）：")
    for item in hot_early[:10]:
        print(f"  [{item['category']}] #{item['rank']} {item['title']} | 评论:{item['reviews']} | 价格:{item['price']}")

    # 保存完整数据
    with open(f"hot_new_releases_{datetime.now():%Y%m%d_%H%M}.json", "w", encoding="utf-8") as f:
        json.dump(all_results, f, ensure_ascii=False, indent=2)