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"

MARKETPLACE_ID = "ATVPDKIKX0DER"
ASIN = "B0FR1WRK4M"
IMAGE_URL = "https://cunchu.site/temp/8c7a90e1-a5a1-41af-a119-41b1ca74b40d.png"

TARGET_WIDTH = 970
TARGET_HEIGHT = 600

# ==================== 签名工具 ====================
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():
    resp = requests.post("https://api.amazon.com/auth/o2/token", 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)

# ==================== 主程序 ====================
if __name__ == "__main__":
    print("=" * 60)
    print("A+ Content 完整流程 - marketplaceId 修正版")
    print("=" * 60)
    
    # Step 0
    print("\n🔑 Step 0: 获取 Access Token...")
    token = get_access_token()
    print("   ✅ Token 获取成功")
    
    # Step 1
    print("\n📥 Step 1: 下载图片并调整尺寸...")
    img_resp = requests.get(IMAGE_URL, timeout=30)
    img_resp.raise_for_status()
    
    try:
        from PIL import Image
        img = Image.open(BytesIO(img_resp.content))
        print(f"   原图尺寸: {img.size}")
        img_resized = img.resize((TARGET_WIDTH, TARGET_HEIGHT), Image.LANCZOS)
        buffer = BytesIO()
        img_resized.save(buffer, format="PNG")
        image_binary = buffer.getvalue()
        print(f"   ✅ 调整后尺寸: {TARGET_WIDTH}x{TARGET_HEIGHT}")
    except ImportError:
        print("   ❌ 需要安装 Pillow: pip install Pillow")
        exit(1)
    
    md5_hash = hashlib.md5(image_binary).digest()
    content_md5 = base64.b64encode(md5_hash).decode('utf-8')
    print(f"   图片大小: {len(image_binary)} bytes")
    
    # Step 2: Uploads API 用 marketplaceIds（复数）✅
    print("\n📝 Step 2: 创建上传目标...")
    resource = "aplus/2020-11-01/contentDocuments"
    path = f"/uploads/2020-11-01/uploadDestinations/{quote(resource, safe='/')}"
    query = f"marketplaceIds={MARKETPLACE_ID}&contentMD5={quote(content_md5, safe='')}&contentType=image%2Fpng"
    
    resp = spapi_request("POST", path, query, access_token=token)
    print(f"   状态码: {resp.status_code}")
    
    if resp.status_code not in [200, 201]:
        print(f"   ❌ 失败: {resp.text}")
        exit(1)
    
    payload = resp.json().get("payload", {})
    upload_id = payload.get("uploadDestinationId")
    upload_url = payload.get("url")
    print(f"   ✅ uploadDestinationId: {upload_id}")
    
    # Step 3
    print("\n⬆️ Step 3: 上传图片到 S3...")
    parsed = urlparse(upload_url)
    base_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
    query_params = dict(parse_qsl(parsed.query))
    files = {"File": ("image.png", image_binary, "image/png")}
    s3_resp = requests.post(base_url, data=query_params, files=files, timeout=60)
    print(f"   S3 状态码: {s3_resp.status_code}")
    
    if s3_resp.status_code not in [200, 201, 204]:
        print("   ❌ S3 上传失败")
        exit(1)
    print("   ✅ S3 上传成功")
    
    # 构建 contentDocument
    content_doc = {
        "name": "Product Image",
        "contentType": "EMC",
        "locale": "en-US",
        "contentModuleList": [
            {
                "contentModuleType": "STANDARD_HEADER_IMAGE_TEXT",
                "standardHeaderImageText": {
                    "headline": {
                        "value": "Premium Quality Bar Table",
                        "decoratorSet": []
                    },
                    "block": {
                        "image": {
                            "uploadDestinationId": upload_id,
                            "imageCropSpecification": {
                                "size": {
                                    "width": {"value": TARGET_WIDTH, "units": "pixels"},
                                    "height": {"value": TARGET_HEIGHT, "units": "pixels"}
                                },
                                "offset": {
                                    "x": {"value": 0, "units": "pixels"},
                                    "y": {"value": 0, "units": "pixels"}
                                }
                            },
                            "altText": "Product detail image"
                        },
                        "headline": {
                            "value": "Elegant Design for Modern Living",
                            "decoratorSet": []
                        },
                        "body": {
                            "textList": [
                                {
                                    "value": "Space-saving bar-height table comfortably seats 2-6 people. Adjustable height, durable engineered wood construction, and smart storage shelves make it perfect for any home.",
                                    "decoratorSet": []
                                }
                            ]
                        }
                    }
                }
            }
        ]
    }
    
    # Step 4: 验证 - A+ API 用 marketplaceId（单数）❌ 之前错了
    print("\n📝 Step 4: 验证内容文档...")
    validate_body = {"contentDocument": content_doc}
    
    # 关键修正：A+ Content API 用 marketplaceId（单数）
    resp = spapi_request("POST", "/aplus/2020-11-01/contentAsinValidations", f"marketplaceId={MARKETPLACE_ID}", body=validate_body, access_token=token)
    print(f"   验证状态码: {resp.status_code}")
    print(f"   验证响应: {resp.text[:800]}")
    
    # Step 5: 创建文档 - A+ API 用 marketplaceId（单数）❌ 之前错了
    print("\n📝 Step 5: 创建 Content Document...")
    
    # 关键修正：A+ Content API 用 marketplaceId（单数）
    resp = spapi_request("POST", "/aplus/2020-11-01/contentDocuments", f"marketplaceId={MARKETPLACE_ID}", body={"contentDocument": content_doc}, access_token=token)
    print(f"   状态码: {resp.status_code}")
    print(f"   响应: {resp.text}")
    
    if resp.status_code not in [200, 201]:
        print("\n   ❌ 创建失败")
        exit(1)
    
    doc_data = resp.json()
    content_ref = doc_data.get("contentReference", {})
    content_key = content_ref.get("key")
    
    if not content_key:
        print("   ❌ 响应中缺少 contentReference.key")
        exit(1)
    
    print(f"   ✅ 创建成功，Content Key: {content_key}")
    
    # Step 6: 绑定 ASIN - A+ API 用 marketplaceId（单数）❌ 之前错了
    print("\n🔗 Step 6: 绑定 ASIN...")
    body = {"asinSet": [ASIN]}
    
    # 关键修正：A+ Content API 用 marketplaceId（单数）
    resp = spapi_request("POST", f"/aplus/2020-11-01/contentDocuments/{content_key}/asins", f"marketplaceId={MARKETPLACE_ID}", body=body, access_token=token)
    print(f"   状态码: {resp.status_code}")
    print(f"   响应: {resp.text[:500]}")
    
    if resp.status_code not in [200, 201]:
        print("   ❌ 绑定失败")
        exit(1)
    print("   ✅ ASIN 绑定成功")
    
    # Step 7: 提交审核 - A+ API 用 marketplaceId（单数）❌ 之前错了
    print("\n📋 Step 7: 提交审核...")
    
    # 关键修正：A+ Content API 用 marketplaceId（单数）
    resp = spapi_request("POST", f"/aplus/2020-11-01/contentDocuments/{content_key}/approvalSubmissions", f"marketplaceId={MARKETPLACE_ID}", access_token=token)
    print(f"   状态码: {resp.status_code}")
    print(f"   响应: {resp.text[:500]}")
    
    if resp.status_code not in [200, 201]:
        print("   ❌ 提交失败")
        exit(1)
    print("   ✅ 审核提交成功")
    
    print("\n" + "=" * 60)
    print("🎉 全部步骤执行成功！")
    print(f"   Content Key: {content_key}")
    print(f"   ASIN: {ASIN}")
    print("=" * 60)
    print("\n⚠️ 请保存 Content Key，用于后续查询状态")
    print("   审核通常需要 1-7 个工作日")