pythonTest/test.py

import requests
import json
import datetime
import hashlib
import hmac
from urllib.parse import quote



# ==================== 配置区域 ====================
CLIENT_ID = "amzn1.application-oa2-client.9f775d35077a4c7598116e5a89b60222"
CLIENT_SECRET = "amzn1.oa2-cs.v1.6909777baa82b796555a431cd23709342e8e01434abcad1f8b59cadf048b51c3"
REFRESH_TOKEN = "Atzr|IwEBICh_fojh7qAfrePzSjOZhQxTnHRZCYOuWNvSAHGaBs1WfkjR4qjC1Dmv0uK3oyq_L-R9SRKK4cYmPwF1MQXQdqKlpzOAGItWeXnNzh-SyhHHISv0_vCt3D6hluVmIEmF2kA7ro_XF-PslD41g3Eq6bv0Qq4y4pQbkDj-9dudDuz7nku8UL-nc4EWgJCrHCJTAs8TK48Yvrfi3E35uv97tYhs6qb8D9d4dR9C7daaC49eoDqP88qyUHThQV1Kqon8xukx4AT2_wLqWNcabpADzuK-5w7D-vvFFtvzfkbmtfzoPeEdHPulG6scc59BPFTyBlGU7a3d1JSFg2ROHIlrljRm"

AWS_ACCESS_KEY = "amzn1.application-oa2-client.9f775d35077a4c7598116e5a89b60222"       # ← 填你的
AWS_SECRET_KEY = "amzn1.oa2-cs.v1.6909777baa82b796555a431cd23709342e8e01434abcad1f8b59cadf048b51c3"             # ← 填你的
AWS_REGION = "us-east-1"

SELLER_ID = "A1H50Z0WYYLEGQ"
SKU = "bar grey 03"
MARKETPLACE_ID = "ATVPDKIKX0DER"

# ==================== 工具函数 ====================
def sha256_hash(data): 
    return hashlib.sha256(data.encode("utf-8")).hexdigest()

def hmac_sha256(key, msg): 
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

def get_signature_key(secret_key, date_stamp, region, service):
    k_date = hmac_sha256(("AWS4" + secret_key).encode("utf-8"), date_stamp)
    k_region = hmac_sha256(k_date, region)
    k_service = hmac_sha256(k_region, service)
    return hmac_sha256(k_service, "aws4_request")

def get_access_token():
    url = "https://api.amazon.com/auth/o2/token"
    resp = requests.post(url, data={
        "grant_type": "refresh_token",
        "refresh_token": REFRESH_TOKEN,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    })
    resp.raise_for_status()
    return resp.json()["access_token"]

# ==================== 带 AWS 签名的请求 ====================
def spapi_request(method, path, query, body=None, access_token=None):
    service = "execute-api"
    host = "sellingpartnerapi-na.amazon.com"
    
    # path 已经包含编码后的 SKU，直接用
    uri = path
    
    payload = json.dumps(body, separators=(",", ":")) if body else ""
    
    t = datetime.datetime.now(datetime.timezone.utc)
    amz_date = t.strftime("%Y%m%dT%H%M%SZ")
    date_stamp = t.strftime("%Y%m%d")
    
    canonical_headers = (
        f"host:{host}\n"
        f"x-amz-access-token:{access_token}\n"
        f"x-amz-date:{amz_date}\n"
    )
    signed_headers = "host;x-amz-access-token;x-amz-date"
    payload_hash = sha256_hash(payload)
    
    canonical_request = (
        f"{method}\n{uri}\n{query}\n"
        f"{canonical_headers}\n{signed_headers}\n{payload_hash}"
    )
    
    algorithm = "AWS4-HMAC-SHA256"
    credential_scope = f"{date_stamp}/{AWS_REGION}/{service}/aws4_request"
    string_to_sign = (
        f"{algorithm}\n{amz_date}\n{credential_scope}\n"
        f"{sha256_hash(canonical_request)}"
    )
    
    signing_key = get_signature_key(AWS_SECRET_KEY, date_stamp, AWS_REGION, service)
    signature = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
    
    auth_header = (
        f"{algorithm} Credential={AWS_ACCESS_KEY}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, Signature={signature}"
    )
    
    headers = {
        "Host": host,
        "x-amz-access-token": access_token,
        "x-amz-date": amz_date,
        "Authorization": auth_header,
    }
    if body:
        headers["Content-Type"] = "application/json"
    
    url = f"https://{host}{uri}?{query}"
    return requests.request(method, url, headers=headers, data=payload or None)

# ==================== 查询商品 ====================
def get_listings_item(access_token):
    # 对 SKU 进行 URL 编码：空格 → %20
    encoded_sku = quote(SKU, safe="")
    path = f"/listings/2021-08-01/items/{SELLER_ID}/{encoded_sku}"
    query = f"marketplaceIds={MARKETPLACE_ID}&includedData=summaries,attributes,issues,offers"
    return spapi_request("GET", path, query, access_token=access_token)

# ==================== 修改主图 ====================
def patch_main_image(access_token, image_url):
    encoded_sku = quote(SKU, safe="")
    path = f"/listings/2021-08-01/items/{SELLER_ID}/{encoded_sku}"
    query = f"marketplaceIds={MARKETPLACE_ID}"
    
    body = {
        "productType": "PRODUCT",
        "patches": [
            {
                "op": "replace",
                "path": "/attributes/main_product_image_locator",
                "value": [
                    {
                        "marketplace_id": MARKETPLACE_ID,
                        "media_location": image_url
                    }
                ]
            }
        ]
    }
    return spapi_request("PATCH", path, query, body=body, access_token=access_token)

# ==================== 主程序 ====================
if __name__ == "__main__":
    print("🔑 获取 Access Token...")
    token = get_access_token()
    print("✅ 令牌获取成功")
    
    print(f"\n📤 查询商品信息（SKU: {SKU}）...")
    resp = get_listings_item(token)
    print(f"状态码: {resp.status_code}")
    
    try:
        data = resp.json()
        print(json.dumps(data, indent=2, ensure_ascii=False))
    except Exception:
        print(resp.text)
    
    if resp.status_code == 200:
        print("\n📋 关键信息：")
        print(f"  productType: {data.get('productType', 'N/A')}")
        summaries = data.get('summaries', [])
        if summaries:
            print(f"  标题: {summaries[0].get('itemName', 'N/A')}")
        
        attrs = data.get('attributes', {})
        main_img = attrs.get('main_product_image_locator', [])
        if main_img:
            print(f"  当前主图: {main_img[0].get('value', {}).get('link', 'N/A')}")
        else:
            print("  当前主图: 未设置")