pythonTest/test2.py

import requests
import json
import datetime
import hashlib
import hmac
import time
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 = "sofa table black"
MARKETPLACE_ID = "ATVPDKIKX0DER"
NEW_IMAGE_URL = "https://cunchu.site/temp/f1ae4d03-e59e-45c5-9a3b-15dc3775396a.jpg"

# ==================== 工具函数 ====================
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"
    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{path}\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}{path}?{query}"
    return requests.request(method, url, headers=headers, data=payload or None)

# ==================== 通用图片修改函数 ====================
def patch_image(access_token, image_type, image_url):
    """
    image_type 可选值：
      "main"              → 主图
      "other_1" ~ "other_8" → 副图1~8
      "swatch"            → 变体图
    """
    encoded_sku = quote(SKU, safe="")
    path = f"/listings/2021-08-01/items/{SELLER_ID}/{encoded_sku}"
    query = f"marketplaceIds={MARKETPLACE_ID}"
    
    # 根据类型生成 path
    if image_type == "main":
        attr_path = "/attributes/main_product_image_locator"
    elif image_type.startswith("other_") and 1 <= int(image_type.split("_")[1]) <= 8:
        n = image_type.split("_")[1]
        attr_path = f"/attributes/other_product_image_locator_{n}"
    elif image_type == "swatch":
        attr_path = "/attributes/swatch_product_image_locator"
    else:
        raise ValueError(f"不支持的 image_type: {image_type}")
    
    body = {
        "productType": "TABLE",
        "patches": [
            {
                "op": "replace",
                "path": attr_path,
                "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("✅ 令牌获取成功")
    
    # 示例：增加/修改副图第3张
    new_images = {
        "other_1": "https://cunchu.site/temp/cd3a846e-f3b4-4858-a643-e1edc60cbe95.png"
    }
    
    for img_type, img_url in new_images.items():
        print(f"\n🖼️ 正在设置 {img_type}: {img_url}")
        resp = patch_image(token, img_type, img_url)
        print(f"📤 状态码: {resp.status_code}")
        try:
            timestamp = int(time.time())
            print(timestamp)
            print(json.dumps(resp.json(), indent=2, ensure_ascii=False))
        except Exception:
            print(resp.text)
        
        if resp.status_code == 200:
            print(f"✅ {img_type} 提交成功")
        else:
            print(f"⚠️ {img_type} 提交失败")