pythonTest/test2.py

import requests
import json
import datetime
import hashlib
import hmac
import uuid
import base64
from urllib.parse import quote, urlparse, parse_qsl
from io import BytesIO

# ==================== 配置（全部填好，直接运行） ====================
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 = "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"
AWS_REGION = "us-east-1"

SELLER_ID = "A1H50Z0WYYLEGQ"
SKU = "bar table pink"
MARKETPLACE_ID = "ATVPDKIKX0DER"
ASIN = "B0FR1WRK4M"

CONTENT_KEY = f"amzn1.aplus.content.v1.{uuid.uuid4()}"
IMAGE_URL = "https://cunchu.site/temp/8c7a90e1-a5a1-41af-a119-41b1ca74b40d.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"
    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)

# ==================== Step 1: 下载图片并计算 MD5 ====================
def download_and_hash_image(image_url):
    print(f"⬇️ 下载图片: {image_url}")
    resp = requests.get(image_url, timeout=30)
    resp.raise_for_status()
    image_binary = resp.content
    print(f"   图片大小: {len(image_binary)} bytes")
    
    md5_hash = hashlib.md5(image_binary).digest()
    content_md5 = base64.b64encode(md5_hash).decode('utf-8')
    print(f"   Content-MD5: {content_md5}")
    
    width, height = 1940, 1200
    try:
        from PIL import Image
        img = Image.open(BytesIO(image_binary))
        width, height = img.size
        print(f"   图片尺寸: {width}x{height}")
    except ImportError:
        print(f"   未安装 Pillow，使用默认尺寸: {width}x{height}")
    
    return image_binary, content_md5, width, height

# ==================== Step 2: 创建上传目标（官方修正版） ====================
def create_upload_destination(access_token, content_md5, content_type):
    """
    官方修正：resource 固定为 aplus/2020-11-01/contentDocuments
    不需要 contentDocumentId，不需要图片文件名！
    """
    resource = "aplus/2020-11-01/contentDocuments"  # ← 关键修正：固定字符串
    encoded_resource = quote(resource, safe="/")
    
    path = f"/uploads/2020-11-01/uploadDestinations/{encoded_resource}"
    query = f"marketplaceIds={MARKETPLACE_ID}&contentMD5={quote(content_md5, safe='')}&contentType={quote(content_type, safe='')}"
    
    print(f"\n📝 创建上传目标...")
    print(f"   Resource: {resource}")
    print(f"   Content-MD5: {content_md5}")
    print(f"   Content-Type: {content_type}")
    
    resp = spapi_request("POST", path, query, access_token=access_token)
    print(f"📤 状态码: {resp.status_code}")
    print(f"📄 响应: {resp.text}")
    
    if resp.status_code != 200:
        return None, None
    
    data = resp.json()
    upload_destination_id = data.get("uploadDestinationId")
    url = data.get("url")
    
    print(f"✅ uploadDestinationId: {upload_destination_id}")
    print(f"✅ upload URL: {url[:120]}...")
    
    return upload_destination_id, url

# ==================== Step 3: Multipart Form POST 上传图片到 S3 ====================
def upload_image_to_s3(upload_url, image_binary, content_type):
    print(f"\n⬆️ 解析预签名 URL 并上传图片...")
    
    parsed = urlparse(upload_url)
    base_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
    query_params = dict(parse_qsl(parsed.query))
    
    print(f"   Base URL: {base_url}")
    print(f"   Form fields: {list(query_params.keys())}")
    
    files = {
        "File": ("image.png", image_binary, content_type)
    }
    
    resp = requests.post(base_url, data=query_params, files=files, timeout=60)
    print(f"📤 S3 上传状态码: {resp.status_code}")
    
    if resp.text:
        print(f"📄 S3 响应: {resp.text[:500]}")
    
    return resp.status_code in [200, 201, 204]

# ==================== Step 4: 创建 A+ Content 文档 ====================
def create_content_document(access_token, content_key, upload_id, img_width, img_height):
    path = "/aplus/2020-11-01/contentDocuments"
    query = f"marketplaceIds={MARKETPLACE_ID}"
    
    body = {
        "contentDocument": {
            "name": "Product Detail Image",
            "contentReference": {
                "key": content_key,
                "version": "0"
            },
            "contentModuleList": [
                {
                    "moduleType": "STANDARD_SINGLE_IMAGE",
                    "standardSingleImage": {
                        "image": {
                            "uploadDestinationId": upload_id,
                            "imageCropSpecification": {
                                "size": {
                                    "width": {"value": img_width, "units": "pixels"},
                                    "height": {"value": img_height, "units": "pixels"}
                                },
                                "offset": {
                                    "x": {"value": 0, "units": "pixels"},
                                    "y": {"value": 0, "units": "pixels"}
                                }
                            },
                            "altText": "Product detail image"
                        }
                    }
                }
            ]
        }
    }
    
    print(f"\n📝 创建 A+ Content 文档...")
    print(f"   Content Key: {content_key}")
    resp = spapi_request("POST", path, query, body=body, access_token=access_token)
    print(f"📤 状态码: {resp.status_code}")
    print(resp.text)
    return resp

# ==================== Step 5: 绑定 ASIN ====================
def associate_asin(access_token, content_key, asin):
    path = f"/aplus/2020-11-01/contentDocuments/{content_key}/asins"
    query = f"marketplaceIds={MARKETPLACE_ID}"
    body = {"asinSet": [asin]}
    
    print(f"\n🔗 绑定 ASIN: {asin}")
    resp = spapi_request("POST", path, query, body=body, access_token=access_token)
    print(f"📤 状态码: {resp.status_code}")
    print(resp.text)
    return resp

# ==================== Step 6: 提交审核 ====================
def submit_approval(access_token, content_key):
    path = f"/aplus/2020-11-01/contentDocuments/{content_key}/approvalSubmissions"
    query = f"marketplaceIds={MARKETPLACE_ID}"
    
    print(f"\n📋 提交审核...")
    resp = spapi_request("POST", path, query, access_token=access_token)
    print(f"📤 状态码: {resp.status_code}")
    print(resp.text)
    return resp

# ==================== 主程序 ====================
if __name__ == "__main__":
    print("🔑 获取 Access Token...")
    token = get_access_token()
    print("✅ 令牌获取成功")
    
    print(f"\n🆔 Content Key: {CONTENT_KEY}")
    print(f"🖼️ 图片URL: {IMAGE_URL}")
    print(f"🔗 ASIN: {ASIN}")
    
    # Step 1
    print("\n=== Step 1: 下载图片 ===")
    image_binary, content_md5, img_width, img_height = download_and_hash_image(IMAGE_URL)
    
    # Step 2（官方修正版）
    print("\n=== Step 2: 创建上传目标 ===")
    upload_id, upload_url = create_upload_destination(token, content_md5, "image/png")
    
    if not upload_id or not upload_url:
        print("\n⛔ 创建上传目标失败，终止")
        exit(1)
    
    # Step 3
    print("\n=== Step 3: 上传图片到 S3 ===")
    success = upload_image_to_s3(upload_url, image_binary, "image/png")
    if not success:
        print("\n⛔ 图片上传失败，终止")
        exit(1)
    print("✅ 图片上传成功")
    
    # Step 4
    print("\n=== Step 4: 创建内容文档 ===")
    create_content_document(token, CONTENT_KEY, upload_id, img_width, img_height)
    
    # Step 5
    print("\n=== Step 5: 绑定ASIN ===")
    associate_asin(token, CONTENT_KEY, ASIN)
    
    # Step 6
    print("\n=== Step 6: 提交审核 ===")
    submit_approval(token, CONTENT_KEY)
    
    print("\n🎉 全部步骤执行完毕！")
    print("   A+ 内容已提交审核，通常需要 1-7 个工作日")
    print(f"   Content Key: {CONTENT_KEY}")
    print("   请保存此 Key 以便后续查询状态")