沃尔玛商品详情 API 与库存 API 联动：实时库存同步与低库存预警实现

pip install requests python-dotenv pandas redis tenacity  # redis 用于缓存，tenacity 用于重试

# .env 文件
WALMART_CLIENT_ID=你的ClientID
WALMART_CLIENT_SECRET=你的ClientSecret
WALMART_ENV=sandbox  # sandbox/production
REDIS_HOST=localhost
REDIS_PORT=6379
# 预警配置
LOW_STOCK_THRESHOLD=10  # 低库存阈值（可按商品类目自定义）
WECHAT_WEBHOOK=你的企业微信机器人Webhook

import os
import requests
import redis
import json
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential
from datetime import datetime, timedelta
# 加载环境变量
load_dotenv()
# 初始化 Redis 缓存（存储 Access Token、商品基础信息）
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST"),
    port=int(os.getenv("REDIS_PORT")),
    decode_responses=True
)
def get_walmart_access_token():
    """获取 OAuth 2.0 Access Token（缓存 55 分钟，避免频繁请求）"""
    cache_key = "walmart_access_token"
    cached_token = redis_client.get(cache_key)
    if cached_token:
        return cached_token
    
    # 重新获取 Token
    env = os.getenv("WALMART_ENV", "sandbox")
    token_url = "https://sandbox-api.walmart.com/v3/token" if env == "sandbox" else "https://api.walmart.com/v3/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": os.getenv("WALMART_CLIENT_ID"),
        "client_secret": os.getenv("WALMART_CLIENT_SECRET")
    }
    
    try:
        response = requests.post(token_url, data=payload, timeout=10)
        response.raise_for_status()
        token_data = response.json()
        access_token = token_data["access_token"]
        # 缓存 55 分钟（Token 有效期 1 小时，预留 5 分钟缓冲）
        redis_client.setex(cache_key, 55 * 60, access_token)
        return access_token
    except Exception as e:
        raise Exception(f"获取 Access Token 失败：{str(e)}")
def send_wechat_alert(content):
    """发送企业微信预警通知"""
    webhook = os.getenv("WECHAT_WEBHOOK")
    if not webhook:
        print("未配置企业微信 Webhook，预警未发送")
        return
    
    payload = {
        "msgtype": "text",
        "text": {
            "content": f"【沃尔玛低库存预警】\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{content}"
        }
    }
    
    try:
        requests.post(webhook, json=payload, timeout=5)
        print("预警通知已发送至企业微信")
    except Exception as e:
        print(f"发送预警失败：{str(e)}")

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_product_base_info(product_id, access_token):
    """调用商品详情 API 获取基础信息（缓存 24 小时，减少重复请求）"""
    cache_key = f"walmart_product_{product_id}"
    cached_info = redis_client.get(cache_key)
    if cached_info:
        return json.loads(cached_info)
    
    # 调用商品详情 API
    env = os.getenv("WALMART_ENV", "sandbox")
    api_url = f"https://sandbox-api.walmart.com/v3/items/{product_id}" if env == "sandbox" else f"https://api.walmart.com/v3/items/{product_id}"
    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
    params = {"fields": "itemId,title,brand,variants,attributes.categoryPath"}
    
    response = requests.get(api_url, headers=headers, params=params, timeout=15)
    response.raise_for_status()
    product_data = response.json()
    
    # 缓存商品基础信息（24 小时更新一次）
    redis_client.setex(cache_key, 24 * 3600, json.dumps(product_data))
    return product_data
def batch_get_product_base_info(product_ids, access_token):
    """批量获取商品基础信息（适配多商品场景）"""
    product_info_dict = {}
    for product_id in product_ids:
        try:
            product_info = get_product_base_info(product_id, access_token)
            product_info_dict[product_id] = product_info
        except Exception as e:
            print(f"获取商品 {product_id} 基础信息失败：{str(e)}")
    return product_info_dict

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_product_inventory(product_ids, access_token):
    """调用库存 API 批量拉取商品库存（支持单商品/SKU 库存查询）"""
    env = os.getenv("WALMART_ENV", "sandbox")
    api_url = "https://sandbox-api.walmart.com/v3/inventory" if env == "sandbox" else "https://api.walmart.com/v3/inventory"
    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
    
    # 构造请求参数：批量商品 ID
    params = {
        "itemIds": ",".join(product_ids),
        "includeVariants": "true"  # 支持多规格商品的 SKU 库存查询
    }
    
    response = requests.get(api_url, headers=headers, params=params, timeout=20)
    response.raise_for_status()
    inventory_data = response.json()
    return inventory_data
def link_product_and_inventory(product_info_dict, inventory_data):
    """联动商品基础信息与库存数据，生成结构化结果"""
    linked_result = []
    inventory_items = inventory_data.get("inventoryItems", [])
    
    for inventory in inventory_items:
        product_id = inventory.get("itemId")
        # 匹配商品基础信息
        product_base = product_info_dict.get(product_id)
        if not product_base:
            print(f"商品 {product_id} 未查询到基础信息，跳过关联")
            continue
        
        # 处理单规格商品
        if not product_base.get("variants"):
            linked_item = {
                "itemId": product_id,
                "title": product_base.get("title"),
                "brand": product_base.get("brand"),
                "category": product_base.get("attributes", {}).get("categoryPath", ""),
                "variantId": None,  # 无 SKU
                "variantSpec": "单规格",
                "availableQuantity": inventory.get("availableQuantity", 0),
                "stockStatus": inventory.get("status", "UNKNOWN"),
                "syncTime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            linked_result.append(linked_item)
        # 处理多规格商品（SKU 级库存）
        else:
            variants = product_base.get("variants", {}).get("variant", [])
            variant_map = {v.get("variantId"): v for v in variants}  # 构建 SKU-ID 映射
            
            # 库存 API 返回的 SKU 库存列表
            variant_inventories = inventory.get("variantInventories", {}).get("variantInventory", [])
            for var_inv in variant_inventories:
                variant_id = var_inv.get("variantId")
                variant_info = variant_map.get(variant_id, {})
                
                linked_item = {
                    "itemId": product_id,
                    "title": product_base.get("title"),
                    "brand": product_base.get("brand"),
                    "category": product_base.get("attributes", {}).get("categoryPath", ""),
                    "variantId": variant_id,
                    "variantSpec": f"颜色：{variant_info.get('attributes', {}).get('color', '未知')} | 尺寸：{variant_info.get('attributes', {}).get('size', '未知')}",
                    "availableQuantity": var_inv.get("availableQuantity", 0),
                    "stockStatus": var_inv.get("status", "UNKNOWN"),
                    "syncTime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                }
                linked_result.append(linked_item)
    
    return linked_result

def check_low_stock(linked_data, threshold=None):
    """检查低库存商品并触发预警"""
    threshold = threshold or int(os.getenv("LOW_STOCK_THRESHOLD", 10))
    low_stock_items = [item for item in linked_data if item["availableQuantity"] < threshold]
    
    if not low_stock_items:
        print(f"当前无低库存商品（阈值：{threshold}）")
        return
    
    # 生成预警内容
    alert_content = ""
    for item in low_stock_items:
        alert_content += (
            f"商品 ID：{item['itemId']}\n"
            f"商品名称：{item['title']}\n"
            f"规格：{item['variantSpec']}\n"
            f"当前库存：{item['availableQuantity']}\n"
            f"库存状态：{item['stockStatus']}\n"
            "---\n"
        )
    
    # 发送预警
    send_wechat_alert(alert_content)
    return low_stock_items
def save_inventory_to_csv(linked_data, file_path="walmart_inventory_sync.csv"):
    """将同步后的商品-库存数据保存为 CSV（便于运营查看）"""
    import pandas as pd
    df = pd.DataFrame(linked_data)
    # 追加模式写入，避免覆盖历史数据
    df.to_csv(file_path, mode="a", header=not os.path.exists(file_path), index=False, encoding="utf-8-sig")
    print(f"库存数据已保存至：{file_path}")
def save_to_mysql(linked_data):
    """将数据保存至 MySQL（生产环境推荐，需提前创建表）"""
    import pymysql
    from pymysql.err import OperationalError
    
    # 数据库配置（需自行补充）
    db_config = {
        "host": "localhost",
        "user": "root",
        "password": "your_password",
        "database": "walmart_inventory"
    }
    
    # 表结构示例（需提前创建）：
    # CREATE TABLE inventory_sync (
    #     id INT AUTO_INCREMENT PRIMARY KEY,
    #     itemId VARCHAR(20) NOT NULL,
    #     title VARCHAR(255) NOT NULL,
    #     brand VARCHAR(100),
    #     category VARCHAR(255),
    #     variantId VARCHAR(20),
    #     variantSpec VARCHAR(100),
    #     availableQuantity INT NOT NULL,
    #     stockStatus VARCHAR(20) NOT NULL,
    #     syncTime DATETIME NOT NULL
    # );
    
    try:
        conn = pymysql.connect(**db_config)
        cursor = conn.cursor()
        
        insert_sql = """
        INSERT INTO inventory_sync (itemId, title, brand, category, variantId, variantSpec, availableQuantity, stockStatus, syncTime)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
        
        data_tuples = [
            (
                item["itemId"], item["title"], item["brand"], item["category"],
                item["variantId"], item["variantSpec"], item["availableQuantity"],
                item["stockStatus"], item["syncTime"]
            ) for item in linked_data
        ]
        
        cursor.executemany(insert_sql, data_tuples)
        conn.commit()
        print(f"成功插入 {cursor.rowcount} 条库存同步数据")
    
    except OperationalError as e:
        print(f"MySQL 连接失败：{str(e)}")
    except Exception as e:
        conn.rollback()
        print(f"数据插入失败：{str(e)}")
    finally:
        if conn:
            conn.close()

def inventory_sync_main(product_ids, low_stock_threshold=None):
    """主流程：获取 Token → 批量获取商品基础信息 → 拉取库存 → 联动解析 → 预警 → 数据落地"""
    try:
        # 1. 获取 Access Token
        access_token = get_walmart_access_token()
        print("Access Token 获取成功")
        
        # 2. 批量获取商品基础信息
        print("开始获取商品基础信息...")
        product_info_dict = batch_get_product_base_info(product_ids, access_token)
        if not product_info_dict:
            print("未获取到有效商品基础信息，流程终止")
            return
        
        # 3. 拉取实时库存数据
        print("开始拉取库存数据...")
        inventory_data = get_product_inventory(product_ids, access_token)
        
        # 4. 联动商品与库存数据
        print("开始关联商品与库存数据...")
        linked_data = link_product_and_inventory(product_info_dict, inventory_data)
        print(f"成功关联 {len(linked_data)} 条商品-库存记录")
        
        # 5. 低库存检查与预警
        print("开始检查低库存商品...")
        low_stock_items = check_low_stock(linked_data, low_stock_threshold)
        
        # 6. 数据落地（CSV + MySQL）
        save_inventory_to_csv(linked_data)
        save_to_mysql(linked_data)
        
        return linked_data, low_stock_items
    
    except Exception as e:
        print(f"库存同步流程失败：{str(e)}")
        return None, None
# 定时任务：每 15 分钟同步一次（生产环境可使用 crontab 或 Airflow 调度）
def scheduled_sync(product_ids, interval_minutes=15):
    import time
    while True:
        print(f"\n=== 开始第 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 次库存同步 ===")
        inventory_sync_main(product_ids)
        print(f"=== 同步完成，下次同步将在 {interval_minutes} 分钟后执行 ===")
        time.sleep(interval_minutes * 60)
# 测试执行（替换为实际商品 ID 列表）
if __name__ == "__main__":
    # 沙箱测试商品 ID（生产环境替换为真实商品 ID）
    TEST_PRODUCT_IDS = ["123456789", "987654321", "112233445"]
    # 单次同步
    inventory_sync_main(TEST_PRODUCT_IDS, low_stock_threshold=10)
    # 定时同步（注释掉单次同步，开启定时任务）
    # scheduled_sync(TEST_PRODUCT_IDS, interval_minutes=15)