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 = "AKIA24CLQSVTW2K4S6RP"       # ← 填你的
AWS_SECRET_KEY = "CwBJqlsYATNC920A+U6iv80WnP6Eso+ECfIalQ1W"             # ← 填你的
AWS_REGION = "us-east-1"

SELLER_ID = "A1H50Z0WYYLEGQ"
SKU = "bar grey 03"
MARKETPLACE_ID = "ATVPDKIKX0DER"
NEW_IMAGE_URL = "https://cunchu.site/temp/f1ae4d03-e59e-45c5-9a3b-15dc3775396a.png"

# ==================== 工具函数 ====================
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"]

def spapi_request(method, path, query, body=None, access_token=None):
    service = "execute-api"
    host = "sellingpartnerapi-na.amazon.com"
    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 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": "TABLE",  # ← 查询返回的是 TABLE，必须填这个
        "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🖼️ 正在修改主图...")
    print(f"   SKU: {SKU}")
    print(f"   新图片: {NEW_IMAGE_URL}")
    
    resp = patch_main_image(token, NEW_IMAGE_URL)
    print(f"\n📤 状态码: {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("   注意：亚马逊处理图片需要几分钟到几小时，前台更新可能有延迟。")
    elif resp.status_code == 400:
        print("\n⚠️ 请求参数有误，请检查 productType 或图片 URL 格式。")
    elif resp.status_code == 403:
        print("\n⛔ 权限不足，请确认 Product Listing 角色已审批通过。")
    else:
        print(f"\n⚠️ 未知状态码: {resp.status_code}")