阿里巴巴国际站店铺商品接口实战：外贸全店商品拉取+站点适配+异常降级（Python合规版）

import requests
import hmac
import hashlib
import time
import json
from datetime import datetime, timedelta

# 阿里巴巴国际站alibaba.com 店铺全商品接口 非国内1688
class AlibabaGlobalShopProduct:
    def __init__(self, app_key, app_secret, refresh_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.refresh_token = refresh_token
        self.api_gateway = "https://gw.open.alibaba.com/api"
        self.access_token = ""
        self.token_safe_expire = None
        self.refresh_global_token()

    def refresh_global_token(self):
        """国际站专属令牌刷新，提前80秒续期规避过期"""
        token_url = "https://gw.open.alibaba.com/oauth/token"
        payload = {
            "grant_type":"refresh_token","appkey":self.app_key,
            "refresh_token":self.refresh_token
        }
        res = requests.post(token_url,data=payload,timeout=12).json()
        self.access_token = res["data"]["access_token"]
        expire = int(res["data"]["expires_in"])
        self.token_safe_expire = datetime.now() + timedelta(seconds=expire-80)

    def create_global_sign(self,params):
        """国际站HMAC-SHA256官方签名，区别1688MD5签名"""
        sort_param = sorted(params.items(),key=lambda x:x[0])
        sign_str = "".join([f"{k}{v}" for k,v in sort_param])
        sign = hmac.new(self.app_secret.encode(),sign_str.encode(),hashlib.sha256).hexdigest().upper()
        return sign

    def get_page_product(self,shop_id,page=1,page_size=30,product_status=1):
        """单页获取店铺商品：1上架 2下架 3审核中"""
        if datetime.now() >= self.token_safe_expire:
            self.refresh_global_token()
        params = {
            "method":"com.alibaba.product.foreign.list.get",
            "appkey":self.app_key,"access_token":self.access_token,
            "timestamp":str(int(time.time()*1000)),"v":"1.0",
            "shopId":shop_id,"pageNo":page,"pageSize":page_size,"status":product_status
        }
        params["sign"] = self.create_global_sign(params)
        try:
            resp = requests.post(self.api_gateway,data=params,timeout=15)
            result = resp.json()
            if result.get("code") != 0:
                return {"code":-1,"msg":result.get("message")}
            data = result["data"]
            product_list = []
            # 外贸跨境字段结构化封装
            for item in data.get("productList",[]):
                product_list.append({
                    "foreign_id":item["productId"],"title_en":item["enTitle"],
                    "title_cn":item.get("cnTitle"),"fob_price":item["fobPrice"],
                    "trade_moq":item["minOrderQty"],"supply_country":item["originCountry"],
                    "product_status":item["status"],"main_img":item["mainImg"]
                })
            time.sleep(0.5)
            return {"code":200,"total":data["total"],"page":page,"data":product_list}
        except Exception as e:
            return {"code":-2,"msg":f"请求异常:{str(e)}"}

    def get_shop_all_product(self,shop_id,status=1):
        """一键获取店铺全部商品，自动分页闭环"""
        all_data = []
        curr_page = 1
        while True:
            res = self.get_page_product(shop_id,curr_page,30,status)
            if res["code"] !=200 or len(res["data"])==0:
                break
            all_data.extend(res["data"])
            if len(res["data"]) <30:break
            curr_page +=1
        return {"shop_id":shop_id,"total":len(all_data),"product_all":all_data}

# 调用示例
if __name__ == "__main__":
    api = AlibabaGlobalShopProduct(
        app_key="你的国际站AppKey",
        app_secret="你的国际站AppSecret",
        refresh_token="商家授权RefreshToken"
    )
    # 传入国际站店铺ID，获取全店上架外贸商品
    result = api.get_shop_all_product(shop_id="GL20260512001")
    print(json.dumps(result,ensure_ascii=False,indent=2))