import tkinter as tk from tkinter import ttk, filedialog, messagebox, scrolledtext import pandas as pd import os import threading import requests from openpyxl import load_workbook from openpyxl.utils import get_column_letter import re import json import datetime import hashlib import hmac from urllib.parse import quote, urlparse, parse_qsl import time import base64 from io import BytesIO class AmazonImageUploader: """亚马逊图片上传类""" def __init__(self, config=None): # 默认配置 defaults = { "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", "marketplace_id": "ATVPDKIKX0DER", "product_type": "TABLE", } if config: defaults.update(config) self.client_id = defaults["client_id"] self.client_secret = defaults["client_secret"] self.refresh_token = defaults["refresh_token"] self.aws_access_key = defaults["aws_access_key"] self.aws_secret_key = defaults["aws_secret_key"] self.aws_region = defaults["aws_region"] self.seller_id = defaults["seller_id"] self.marketplace_id = defaults["marketplace_id"] self.product_type = defaults["product_type"] def update_config(self, config): """更新配置""" if "client_id" in config: self.client_id = config["client_id"] if "client_secret" in config: self.client_secret = config["client_secret"] if "refresh_token" in config: self.refresh_token = config["refresh_token"] if "aws_access_key" in config: self.aws_access_key = config["aws_access_key"] if "aws_secret_key" in config: self.aws_secret_key = config["aws_secret_key"] if "aws_region" in config: self.aws_region = config["aws_region"] if "seller_id" in config: self.seller_id = config["seller_id"] if "marketplace_id" in config: self.marketplace_id = config["marketplace_id"] if "product_type" in config: self.product_type = config["product_type"] def sha256_hash(self, data): """兼容 str 和 bytes""" if isinstance(data, str): data = data.encode("utf-8") return hashlib.sha256(data).hexdigest() def hmac_sha256(self, key, msg): """返回 bytes 类型的 digest，兼容 str 和 bytes 输入""" if isinstance(key, str): key = key.encode("utf-8") if isinstance(msg, str): msg = msg.encode("utf-8") return hmac.new(key, msg, hashlib.sha256).digest() def get_signature_key(self, secret_key, date_stamp, region, service): k_date = self.hmac_sha256(("AWS4" + secret_key).encode("utf-8"), date_stamp) k_region = self.hmac_sha256(k_date, region) k_service = self.hmac_sha256(k_region, service) return self.hmac_sha256(k_service, "aws4_request") def get_access_token(self): url = "https://api.amazon.com/auth/o2/token" resp = requests.post(url, data={ "grant_type": "refresh_token", "refresh_token": self.refresh_token, "client_id": self.client_id, "client_secret": self.client_secret, }) resp.raise_for_status() return resp.json()["access_token"] def spapi_request(self, 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 = self.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}/{self.aws_region}/{service}/aws4_request" string_to_sign = ( f"{algorithm}\n{amz_date}\n{credential_scope}\n" f"{self.sha256_hash(canonical_request)}" ) signing_key = self.get_signature_key(self.aws_secret_key, date_stamp, self.aws_region, service) signature = self.hmac_sha256(signing_key, string_to_sign).hex() auth_header = ( f"{algorithm} Credential={self.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(self, access_token, image_type, image_url, sku): """ image_type 可选值： "main" → 主图 "other_1" ~ "other_8" → 副图1~8 "swatch" → 变体图 """ encoded_sku = quote(sku, safe="") path = f"/listings/2021-08-01/items/{self.seller_id}/{encoded_sku}" query = f"marketplaceIds={self.marketplace_id}" 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": self.product_type, "patches": [ { "op": "replace", "path": attr_path, "value": [ { "marketplace_id": self.marketplace_id, "media_location": image_url } ] } ] } return self.spapi_request("PATCH", path, query, body=body, access_token=access_token) # ==================== A+ Content API 方法 ==================== def create_upload_destination(self, access_token, content_md5, content_type="image/png"): """创建上传目标""" resource = "aplus/2020-11-01/contentDocuments" path = f"/uploads/2020-11-01/uploadDestinations/{quote(resource, safe='/')}" query = f"marketplaceIds={self.marketplace_id}&contentMD5={quote(content_md5, safe='')}&contentType={quote(content_type, safe='')}" resp = self.spapi_request("POST", path, query, access_token=access_token) if resp.status_code not in [200, 201]: raise Exception(f"创建上传目标失败: {resp.text}") return resp.json().get("payload", {}) def upload_to_s3(self, upload_url, image_binary, content_type="image/png"): """上传图片到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, content_type)} resp = requests.post(base_url, data=query_params, files=files, timeout=60) if resp.status_code not in [200, 201, 204]: raise Exception(f"S3上传失败: {resp.status_code}") return True def create_content_document(self, access_token, content_doc): """创建内容文档""" resp = self.spapi_request( "POST", "/aplus/2020-11-01/contentDocuments", f"marketplaceId={self.marketplace_id}", body={"contentDocument": content_doc}, access_token=access_token ) if resp.status_code not in [200, 201]: raise Exception(f"创建内容文档失败: {resp.text}") return resp.json().get("contentReferenceKey") def bind_asin_to_content(self, access_token, content_key, asin): """绑定ASIN到内容""" body = {"asinSet": [asin]} resp = self.spapi_request( "POST", f"/aplus/2020-11-01/contentDocuments/{content_key}/asins", f"marketplaceId={self.marketplace_id}", body=body, access_token=access_token ) if resp.status_code not in [200, 201]: raise Exception(f"绑定ASIN失败: {resp.text}") return True def submit_for_approval(self, access_token, content_key): """提交审核""" resp = self.spapi_request( "POST", f"/aplus/2020-11-01/contentDocuments/{content_key}/approvalSubmissions", f"marketplaceId={self.marketplace_id}", access_token=access_token ) if resp.status_code not in [200, 201]: raise Exception(f"提交审核失败: {resp.text}") return True class ImageMatcherApp: def __init__(self, root): self.root = root self.root.title("图片匹配与上传工具 - 亚马逊版") self.root.geometry("950x800") # 存储文件路径 self.template_path = tk.StringVar() self.image_dir = tk.StringVar() # 固定上传配置 self.upload_url = "https://cos.port.run/uploadFile" self.upload_group = "soundSynthesis" self.upload_user = "1" # 亚马逊上传器 self.amazon_uploader = AmazonImageUploader() # 创建UI self.create_widgets() # 加载保存的配置 self.load_config() def create_widgets(self): # 主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 配置网格权重 self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame.columnconfigure(0, weight=1) main_frame.rowconfigure(1, weight=1) # 创建Notebook（选项卡） self.notebook = ttk.Notebook(main_frame) self.notebook.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), pady=10) # 选项卡1: 图片匹配与上传 self.tab_matching = ttk.Frame(self.notebook) self.notebook.add(self.tab_matching, text="图片匹配上传") self.create_matching_tab() # 选项卡2: 亚马逊主图更换 self.tab_amazon = ttk.Frame(self.notebook) self.notebook.add(self.tab_amazon, text="亚马逊主图更换") self.create_amazon_tab() # 选项卡3: 亚马逊详情图更换 (A+ Content) self.tab_detail = ttk.Frame(self.notebook) self.notebook.add(self.tab_detail, text="亚马逊详情图更换") self.create_detail_tab() # 选项卡4: 设置 self.tab_settings = ttk.Frame(self.notebook) self.notebook.add(self.tab_settings, text="设置") self.create_settings_tab() # 状态栏 self.status_var = tk.StringVar() self.status_var.set("就绪") status_bar = ttk.Label(main_frame, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W) status_bar.grid(row=2, column=0, sticky=(tk.W, tk.E), pady=5) def create_matching_tab(self): """创建图片匹配上传选项卡""" tab = self.tab_matching tab.columnconfigure(1, weight=1) tab.rowconfigure(4, weight=1) # 标题 title_label = ttk.Label(tab, text="图片匹配与上传", font=("Arial", 14, "bold")) title_label.grid(row=0, column=0, columnspan=3, pady=10) # 模板文件选择 ttk.Label(tab, text="模板文件:").grid(row=1, column=0, sticky=tk.W, pady=5) ttk.Entry(tab, textvariable=self.template_path, width=60).grid(row=1, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) ttk.Button(tab, text="选择模板", command=self.select_template).grid(row=1, column=2, pady=5, padx=5) # 图片目录选择 ttk.Label(tab, text="图片目录:").grid(row=2, column=0, sticky=tk.W, pady=5) ttk.Entry(tab, textvariable=self.image_dir, width=60).grid(row=2, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) ttk.Button(tab, text="选择目录", command=self.select_image_dir).grid(row=2, column=2, pady=5, padx=5) # 按钮框架 button_frame = ttk.Frame(tab) button_frame.grid(row=3, column=0, columnspan=3, pady=10) ttk.Button(button_frame, text="开始处理", command=self.start_matching, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="清空日志", command=self.clear_log, width=15).pack(side=tk.LEFT, padx=5) # 日志框 log_frame = ttk.LabelFrame(tab, text="日志", padding="5") log_frame.grid(row=4, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S), pady=10) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.log_text = scrolledtext.ScrolledText(log_frame, height=15, width=80, wrap=tk.WORD) self.log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 存储生成的文件路径 self.generated_file_path = None def create_amazon_tab(self): """创建亚马逊主图更换选项卡""" tab = self.tab_amazon tab.columnconfigure(1, weight=1) tab.rowconfigure(5, weight=1) # 标题 title_label = ttk.Label(tab, text="亚马逊主图/附图更换", font=("Arial", 14, "bold")) title_label.grid(row=0, column=0, columnspan=3, pady=10) # SKU输入 ttk.Label(tab, text="SKU:").grid(row=1, column=0, sticky=tk.W, pady=5) self.sku_input = tk.StringVar(value="bar grey 03") ttk.Entry(tab, textvariable=self.sku_input, width=40).grid(row=1, column=1, sticky=tk.W, pady=5, padx=5) # 图片文件夹选择 ttk.Label(tab, text="图片文件夹:").grid(row=2, column=0, sticky=tk.W, pady=5) self.amazon_image_dir = tk.StringVar() ttk.Entry(tab, textvariable=self.amazon_image_dir, width=60).grid(row=2, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) ttk.Button(tab, text="选择文件夹", command=self.select_amazon_image_dir).grid(row=2, column=2, pady=5, padx=5) # 说明标签 note_label = ttk.Label( tab, text="说明：文件夹中的图片将按文件名排序，第1张作为主图，第2-9张作为附图1-8\n" "支持格式：.jpg / .jpeg / .png", foreground="gray" ) note_label.grid(row=3, column=0, columnspan=3, sticky=tk.W, pady=5, padx=5) # 按钮框架 button_frame = ttk.Frame(tab) button_frame.grid(row=4, column=0, columnspan=3, pady=10) ttk.Button(button_frame, text="开始上传", command=self.start_amazon_upload, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="清空日志", command=self.clear_amazon_log, width=15).pack(side=tk.LEFT, padx=5) # 日志框 log_frame = ttk.LabelFrame(tab, text="上传日志", padding="5") log_frame.grid(row=5, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S), pady=10) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.amazon_log_text = scrolledtext.ScrolledText(log_frame, height=15, width=80, wrap=tk.WORD) self.amazon_log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 进度标签 self.amazon_progress_var = tk.StringVar(value="就绪") progress_label = ttk.Label(tab, textvariable=self.amazon_progress_var, relief=tk.SUNKEN, anchor=tk.W) progress_label.grid(row=6, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=5) def create_detail_tab(self): """创建亚马逊详情图更换选项卡 (A+ Content)""" tab = self.tab_detail tab.columnconfigure(1, weight=1) tab.rowconfigure(9, weight=1) # 标题 title_label = ttk.Label(tab, text="亚马逊详情图更换 (A+ Content)", font=("Arial", 14, "bold")) title_label.grid(row=0, column=0, columnspan=3, pady=10) # ASIN输入 ttk.Label(tab, text="ASIN:").grid(row=1, column=0, sticky=tk.W, pady=5) self.asin_input = tk.StringVar(value="B0FR1WRK4M") ttk.Entry(tab, textvariable=self.asin_input, width=40).grid(row=1, column=1, sticky=tk.W, pady=5, padx=5) # 标题输入 ttk.Label(tab, text="标题:").grid(row=2, column=0, sticky=tk.W, pady=5) self.detail_headline = tk.StringVar(value="Premium Quality Product") ttk.Entry(tab, textvariable=self.detail_headline, width=60).grid(row=2, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) # 描述输入 ttk.Label(tab, text="描述:").grid(row=3, column=0, sticky=tk.W, pady=5) self.detail_body = tk.StringVar(value="High-quality product designed for your needs.") ttk.Entry(tab, textvariable=self.detail_body, width=60).grid(row=3, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) # 图片文件夹选择 ttk.Label(tab, text="图片文件夹:").grid(row=4, column=0, sticky=tk.W, pady=5) self.detail_image_dir = tk.StringVar() ttk.Entry(tab, textvariable=self.detail_image_dir, width=60).grid(row=4, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) ttk.Button(tab, text="选择文件夹", command=self.select_detail_image_dir).grid(row=4, column=2, pady=5, padx=5) # 图片尺寸设置 size_frame = ttk.Frame(tab) size_frame.grid(row=5, column=0, columnspan=3, sticky=tk.W, pady=5, padx=5) ttk.Label(size_frame, text="图片宽度(像素):").pack(side=tk.LEFT) self.image_width = tk.StringVar(value="970") ttk.Entry(size_frame, textvariable=self.image_width, width=10).pack(side=tk.LEFT, padx=5) ttk.Label(size_frame, text="图片高度(像素):").pack(side=tk.LEFT, padx=(20, 0)) self.image_height = tk.StringVar(value="600") ttk.Entry(size_frame, textvariable=self.image_height, width=10).pack(side=tk.LEFT, padx=5) # 说明标签 note_label = ttk.Label( tab, text="说明：文件夹中的所有图片将按文件名顺序上传，每张图片作为一个 A+ 模块\n" "支持格式：.jpg / .jpeg / .png", foreground="gray" ) note_label.grid(row=6, column=0, columnspan=3, sticky=tk.W, pady=5, padx=5) # 按钮框架 button_frame = ttk.Frame(tab) button_frame.grid(row=7, column=0, columnspan=3, pady=10) ttk.Button(button_frame, text="开始上传", command=self.start_detail_upload, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="清空日志", command=self.clear_detail_log, width=15).pack(side=tk.LEFT, padx=5) # 日志框 log_frame = ttk.LabelFrame(tab, text="上传日志", padding="5") log_frame.grid(row=8, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S), pady=10) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.detail_log_text = scrolledtext.ScrolledText(log_frame, height=15, width=80, wrap=tk.WORD) self.detail_log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 进度标签 self.detail_progress_var = tk.StringVar(value="就绪") progress_label = ttk.Label(tab, textvariable=self.detail_progress_var, relief=tk.SUNKEN, anchor=tk.W) progress_label.grid(row=9, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=5) def create_settings_tab(self): """创建设置选项卡""" tab = self.tab_settings tab.columnconfigure(0, weight=1) tab.rowconfigure(1, weight=1) # 标题 title_label = ttk.Label(tab, text="亚马逊 API 设置", font=("Arial", 14, "bold")) title_label.grid(row=0, column=0, pady=10, sticky=tk.W, padx=10) # 创建一个可滚动的Frame来容纳所有设置项 canvas = tk.Canvas(tab) scrollbar = ttk.Scrollbar(tab, orient="vertical", command=canvas.yview) scrollable_frame = ttk.Frame(canvas) scrollable_frame.bind( "", lambda e: canvas.configure(scrollregion=canvas.bbox("all")) ) canvas.create_window((0, 0), window=scrollable_frame, anchor="nw") canvas.configure(yscrollcommand=scrollbar.set) canvas.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), padx=10) scrollbar.grid(row=1, column=1, sticky=(tk.N, tk.S)) scrollable_frame.columnconfigure(1, weight=1) # 设置项 row = 0 # Client ID ttk.Label(scrollable_frame, text="Client ID:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_client_id = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_client_id, width=80).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # Client Secret ttk.Label(scrollable_frame, text="Client Secret:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_client_secret = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_client_secret, width=80, show="*").grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # Refresh Token ttk.Label(scrollable_frame, text="Refresh Token:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_refresh_token = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_refresh_token, width=80).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # AWS Access Key ttk.Label(scrollable_frame, text="AWS Access Key:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_aws_access_key = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_aws_access_key, width=80).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # AWS Secret Key ttk.Label(scrollable_frame, text="AWS Secret Key:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_aws_secret_key = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_aws_secret_key, width=80, show="*").grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # AWS Region ttk.Label(scrollable_frame, text="AWS Region:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_aws_region = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_aws_region, width=30).grid(row=row, column=1, sticky=tk.W, pady=5, padx=5) row += 1 # Seller ID ttk.Label(scrollable_frame, text="Seller ID:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_seller_id = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_seller_id, width=80).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # Marketplace ID ttk.Label(scrollable_frame, text="Marketplace ID:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_marketplace_id = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_marketplace_id, width=80).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) row += 1 # Product Type ttk.Label(scrollable_frame, text="Product Type:").grid(row=row, column=0, sticky=tk.W, pady=5, padx=5) self.setting_product_type = tk.StringVar() ttk.Entry(scrollable_frame, textvariable=self.setting_product_type, width=30).grid(row=row, column=1, sticky=tk.W, pady=5, padx=5) row += 1 # 分隔线 ttk.Separator(scrollable_frame, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=15) row += 1 # 按钮框架 button_frame = ttk.Frame(scrollable_frame) button_frame.grid(row=row, column=0, columnspan=2, pady=15) ttk.Button(button_frame, text="保存设置", command=self.save_settings, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="重置为默认值", command=self.reset_settings, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="测试连接", command=self.test_connection, width=15).pack(side=tk.LEFT, padx=5) row += 1 # 设置日志框 log_frame = ttk.LabelFrame(scrollable_frame, text="设置日志", padding="5") log_frame.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=10, padx=5) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.settings_log_text = scrolledtext.ScrolledText(log_frame, height=8, width=80, wrap=tk.WORD) self.settings_log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 初始化设置值 self.init_settings_values() def init_settings_values(self): """初始化设置值""" self.setting_client_id.set(self.amazon_uploader.client_id) self.setting_client_secret.set(self.amazon_uploader.client_secret) self.setting_refresh_token.set(self.amazon_uploader.refresh_token) self.setting_aws_access_key.set(self.amazon_uploader.aws_access_key) self.setting_aws_secret_key.set(self.amazon_uploader.aws_secret_key) self.setting_aws_region.set(self.amazon_uploader.aws_region) self.setting_seller_id.set(self.amazon_uploader.seller_id) self.setting_marketplace_id.set(self.amazon_uploader.marketplace_id) self.setting_product_type.set(self.amazon_uploader.product_type) def settings_log(self, message): """在设置选项卡日志框中添加消息""" self.settings_log_text.insert(tk.END, f"{message}\n") self.settings_log_text.see(tk.END) self.root.update_idletasks() def get_settings_config_path(self): """获取设置配置文件路径""" return os.path.join(os.path.dirname(os.path.abspath(__file__)), "amazon_settings.json") def save_settings(self): """保存设置""" config = { "client_id": self.setting_client_id.get(), "client_secret": self.setting_client_secret.get(), "refresh_token": self.setting_refresh_token.get(), "aws_access_key": self.setting_aws_access_key.get(), "aws_secret_key": self.setting_aws_secret_key.get(), "aws_region": self.setting_aws_region.get(), "seller_id": self.setting_seller_id.get(), "marketplace_id": self.setting_marketplace_id.get(), "product_type": self.setting_product_type.get(), } try: with open(self.get_settings_config_path(), 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) # 更新亚马逊上传器配置 self.amazon_uploader.update_config(config) self.settings_log("✅ 设置已保存并应用") messagebox.showinfo("成功", "设置已保存！") except Exception as e: self.settings_log(f"❌ 保存设置失败: {str(e)}") messagebox.showerror("错误", f"保存设置失败: {str(e)}") def reset_settings(self): """重置为默认值""" if not messagebox.askyesno("确认", "确定要重置为默认值吗？"): return defaults = { "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", "marketplace_id": "ATVPDKIKX0DER", "product_type": "TABLE", } self.setting_client_id.set(defaults["client_id"]) self.setting_client_secret.set(defaults["client_secret"]) self.setting_refresh_token.set(defaults["refresh_token"]) self.setting_aws_access_key.set(defaults["aws_access_key"]) self.setting_aws_secret_key.set(defaults["aws_secret_key"]) self.setting_aws_region.set(defaults["aws_region"]) self.setting_seller_id.set(defaults["seller_id"]) self.setting_marketplace_id.set(defaults["marketplace_id"]) self.setting_product_type.set(defaults["product_type"]) self.settings_log("✅ 已重置为默认值") def test_connection(self): """测试连接""" self.settings_log("🔍 测试连接...") # 先保存设置 self.save_settings() try: # 测试获取Access Token self.settings_log(" 🔑 获取Access Token...") token = self.amazon_uploader.get_access_token() self.settings_log(f" ✅ Token获取成功 (长度: {len(token)})") # 测试SP-API连接（获取marketplace参与信息） self.settings_log(" 🌐 测试SP-API连接...") resp = self.amazon_uploader.spapi_request( "GET", "/sellers/v1/marketplaceParticipations", "", access_token=token ) if resp.status_code == 200: self.settings_log(" ✅ SP-API连接成功") data = resp.json() self.settings_log(f" 参与市场数量: {len(data.get('payload', []))}") else: self.settings_log(f" ⚠️ SP-API返回状态码: {resp.status_code}") self.settings_log(f" 响应: {resp.text[:200]}") self.settings_log("✅ 连接测试完成") except Exception as e: self.settings_log(f"❌ 连接测试失败: {str(e)}") messagebox.showerror("连接失败", f"连接测试失败:\n{str(e)}") def load_settings(self): """加载设置""" try: config_path = self.get_settings_config_path() if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) self.setting_client_id.set(config.get("client_id", self.amazon_uploader.client_id)) self.setting_client_secret.set(config.get("client_secret", self.amazon_uploader.client_secret)) self.setting_refresh_token.set(config.get("refresh_token", self.amazon_uploader.refresh_token)) self.setting_aws_access_key.set(config.get("aws_access_key", self.amazon_uploader.aws_access_key)) self.setting_aws_secret_key.set(config.get("aws_secret_key", self.amazon_uploader.aws_secret_key)) self.setting_aws_region.set(config.get("aws_region", self.amazon_uploader.aws_region)) self.setting_seller_id.set(config.get("seller_id", self.amazon_uploader.seller_id)) self.setting_marketplace_id.set(config.get("marketplace_id", self.amazon_uploader.marketplace_id)) self.setting_product_type.set(config.get("product_type", self.amazon_uploader.product_type)) # 更新上传器配置 self.amazon_uploader.update_config(config) except Exception: pass # ==================== 文件/目录选择 ==================== def select_template(self): filename = filedialog.askopenfilename( title="选择模板文件", filetypes=[("Excel文件", "*.xlsx *.xls"), ("所有文件", "*.*")] ) if filename: self.template_path.set(filename) self.log(f"已选择模板文件: {filename}") def select_image_dir(self): directory = filedialog.askdirectory(title="选择图片目录") if directory: self.image_dir.set(directory) self.log(f"已选择图片目录: {directory}") def select_amazon_image_dir(self): directory = filedialog.askdirectory(title="选择图片文件夹") if directory: self.amazon_image_dir.set(directory) self.amazon_log(f"已选择图片文件夹: {directory}") def select_detail_image_dir(self): directory = filedialog.askdirectory(title="选择图片文件夹") if directory: self.detail_image_dir.set(directory) self.detail_log(f"已选择图片文件夹: {directory}") # ==================== 日志方法 ==================== def log(self, message): """在匹配选项卡日志框中添加消息""" self.log_text.insert(tk.END, f"{message}\n") self.log_text.see(tk.END) self.root.update_idletasks() def clear_log(self): """清空匹配选项卡日志""" self.log_text.delete(1.0, tk.END) def amazon_log(self, message): """在亚马逊选项卡日志框中添加消息""" self.amazon_log_text.insert(tk.END, f"{message}\n") self.amazon_log_text.see(tk.END) self.root.update_idletasks() def clear_amazon_log(self): """清空亚马逊选项卡日志""" self.amazon_log_text.delete(1.0, tk.END) def detail_log(self, message): """在详情图选项卡日志框中添加消息""" self.detail_log_text.insert(tk.END, f"{message}\n") self.detail_log_text.see(tk.END) self.root.update_idletasks() def clear_detail_log(self): """清空详情图选项卡日志""" self.detail_log_text.delete(1.0, tk.END) # ==================== 配置路径 ==================== def get_amazon_config_path(self): """获取主图配置路径""" return os.path.join(os.path.dirname(os.path.abspath(__file__)), "amazon_config.json") def get_detail_config_path(self): """获取详情图配置路径""" return os.path.join(os.path.dirname(os.path.abspath(__file__)), "amazon_detail_config.json") # ==================== 配置文件 ==================== def load_config(self): """启动时自动加载配置""" # 加载主图配置 try: config_path = self.get_amazon_config_path() if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) self.sku_input.set(config.get("sku", "bar grey 03")) if config.get("image_dir"): self.amazon_image_dir.set(config["image_dir"]) except Exception: pass # 加载详情图配置 try: config_path = self.get_detail_config_path() if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) self.asin_input.set(config.get("asin", "B0FR1WRK4M")) self.detail_headline.set(config.get("headline", "Premium Quality Product")) self.detail_body.set(config.get("body", "High-quality product designed for your needs.")) self.image_width.set(config.get("image_width", "970")) self.image_height.set(config.get("image_height", "600")) if config.get("image_dir"): self.detail_image_dir.set(config["image_dir"]) except Exception: pass # 加载设置配置 self.load_settings() # ==================== 图片匹配上传 ==================== def start_matching(self): """开始匹配处理""" if not self.template_path.get(): messagebox.showerror("错误", "请先选择模板文件！") return if not self.image_dir.get(): messagebox.showerror("错误", "请先选择图片目录！") return thread = threading.Thread(target=self.process_matching, daemon=True) thread.start() def process_matching(self): """执行匹配处理""" try: self.status_var.set("处理中...") self.log("=" * 50) self.log("开始处理...") template_path = self.template_path.get() image_dir = self.image_dir.get() self.log(f"读取模板文件: {template_path}") # 使用openpyxl加载工作簿，保留所有样式 wb = load_workbook(template_path) ws = wb.active # 获取表头（第二行） headers = [] for col in range(1, ws.max_column + 1): cell_value = ws.cell(row=2, column=col).value headers.append(str(cell_value) if cell_value else '') self.log(f"找到 {len(headers)} 列") # 查找各列的列号 col_a_index = 1 # A列是第1列 product_main_col = None image_cols = {} for idx, header in enumerate(headers, start=1): if header == '产品主图': product_main_col = idx elif '上架图片' in header: match = re.search(r'(\d+)', header) if match: num = int(match.group(1)) image_cols[num] = idx if product_main_col is None: self.log("警告: 未找到'产品主图'列") else: self.log(f"找到产品主图列: 第{product_main_col}列") self.log(f"找到上架图片列: {len(image_cols)} 列") # 处理每一行（从第三行开始） processed_count = 0 for row_idx in range(3, ws.max_row + 1): a_cell = ws.cell(row=row_idx, column=col_a_index) a_value = str(a_cell.value).strip() if a_cell.value else '' if not a_value or a_value == 'nan' or a_value == 'None': continue self.log(f"\n处理第 {row_idx} 行, A列值: {a_value}") # 处理产品主图 (图片编号1) if product_main_col is not None: img_url = self.find_and_upload_image(a_value, 1, image_dir) if img_url: ws.cell(row=row_idx, column=product_main_col, value=img_url) self.log(f" 产品主图上传成功: {img_url[:50]}...") else: self.log(f" 产品主图: 未找到或上传失败") # 处理上架图片1-6 (图片编号2-7) for i in range(1, 20): if i in image_cols: img_url = self.find_and_upload_image(a_value, i+1, image_dir) if img_url: ws.cell(row=row_idx, column=image_cols[i], value=img_url) self.log(f" 上架图片{i}上传成功: {img_url[:50]}...") # else: # self.log(f" 上架图片{i}: 未找到或上传失败") processed_count += 1 # 保存结果（保留所有样式） output_path = self.generate_output_path(template_path) wb.save(output_path) self.generated_file_path = output_path self.log(f"\n处理完成！共处理 {processed_count} 行数据") self.log(f"结果已保存到: {output_path}") self.status_var.set(f"处理完成！共处理 {processed_count} 行") messagebox.showinfo("完成", f"处理完成！\n共处理 {processed_count} 行数据\n结果保存在:\n{output_path}") except Exception as e: error_msg = f"处理出错: {str(e)}" self.log(f"错误: {error_msg}") self.status_var.set("处理出错") messagebox.showerror("错误", error_msg) import traceback self.log(traceback.format_exc()) def generate_output_path(self, template_path): """生成输出文件路径""" dir_path = os.path.dirname(template_path) base_name = os.path.basename(template_path) name_without_ext = os.path.splitext(base_name)[0] new_filename = f"{name_without_ext}_结果.xlsx" return os.path.join(dir_path, new_filename) def find_and_upload_image(self, a_value, num, image_dir): """查找图片并上传，返回URL""" extensions = ['.jpg', '.jpeg', '.png'] local_file = None for ext in extensions: filename = f"{a_value}_{num}{ext}" filepath = os.path.join(image_dir, filename) if os.path.exists(filepath): local_file = filepath break if not local_file: self.log(f"找不到目录:{filepath}") return None try: full_url = f"{self.upload_url}?group={self.upload_group}&user={self.upload_user}" content_type = 'image/jpeg' if local_file.lower().endswith('.png'): content_type = 'image/png' with open(local_file, 'rb') as f: files = {'file': (os.path.basename(local_file), f, content_type)} response = requests.post(full_url, files=files, timeout=30) if response.status_code == 200: result = response.json() if result.get('err') == 0: return result.get('url') else: self.log(f" 上传失败: {result.get('msg', '未知错误')}") return None else: self.log(f" 上传请求失败，状态码: {response.status_code}") return None except Exception as e: self.log(f" 上传出错: {str(e)}") return None # ==================== 亚马逊主图更换 ==================== def start_amazon_upload(self): """开始亚马逊主图上传""" if not self.sku_input.get().strip(): messagebox.showerror("错误", "请先输入SKU！") return if not self.amazon_image_dir.get(): messagebox.showerror("错误", "请先选择图片文件夹！") return # 保存配置 self.save_amazon_config() thread = threading.Thread(target=self.process_amazon_upload, daemon=True) thread.start() def save_amazon_config(self): """保存主图配置""" config = { "sku": self.sku_input.get(), "image_dir": self.amazon_image_dir.get(), } try: with open(self.get_amazon_config_path(), 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) except Exception: pass def get_image_files(self, folder): """获取文件夹中的图片文件，按文件名排序""" extensions = ['.jpg', '.jpeg', '.png'] files = [] if not os.path.isdir(folder): return files for f in os.listdir(folder): ext = os.path.splitext(f)[1].lower() if ext in extensions: files.append(os.path.join(folder, f)) # 按文件名排序 files.sort(key=lambda x: os.path.basename(x).lower()) return files def upload_image_to_cos(self, local_file): """上传图片到图片服务器，返回URL""" try: full_url = f"{self.upload_url}?group={self.upload_group}&user={self.upload_user}" content_type = 'image/jpeg' if local_file.lower().endswith('.png'): content_type = 'image/png' with open(local_file, 'rb') as f: files = {'file': (os.path.basename(local_file), f, content_type)} response = requests.post(full_url, files=files, timeout=30) if response.status_code == 200: result = response.json() if result.get('err') == 0: return result.get('url') else: raise Exception(result.get('msg', '未知错误')) else: raise Exception(f"HTTP {response.status_code}") except Exception as e: raise Exception(f"上传失败: {str(e)}") def process_amazon_upload(self): """执行亚马逊主图上传""" try: self.amazon_progress_var.set("处理中...") self.amazon_log("=" * 50) self.amazon_log("开始亚马逊主图/附图上传...") sku = self.sku_input.get().strip() image_dir = self.amazon_image_dir.get() self.amazon_log(f"SKU: {sku}") self.amazon_log(f"图片文件夹: {image_dir}") # 获取图片文件 image_files = self.get_image_files(image_dir) if not image_files: self.amazon_log("❌ 未找到任何图片文件") self.amazon_progress_var.set("上传失败") return self.amazon_log(f"找到 {len(image_files)} 张图片") for i, f in enumerate(image_files, start=1): self.amazon_log(f" {i}. {os.path.basename(f)}") # 限制最多9张（1主图+8附图） if len(image_files) > 9: self.amazon_log(f"⚠️ 图片数量超过9张，只使用前9张") image_files = image_files[:9] # 获取Access Token self.amazon_log("\n🔑 获取Access Token...") try: token = self.amazon_uploader.get_access_token() self.amazon_log("✅ Token获取成功") except Exception as e: self.amazon_log(f"❌ 获取Token失败: {str(e)}") self.amazon_progress_var.set("上传失败") return # 处理每张图片 success_count = 0 fail_count = 0 for i, local_file in enumerate(image_files): filename = os.path.basename(local_file) # 确定图片类型 if i == 0: image_type = "main" type_name = "主图" else: image_type = f"other_{i}" type_name = f"附图{i}" self.amazon_log(f"\n处理 {type_name}: {filename}") # 上传到图片服务器获取URL try: self.amazon_log(f" 📤 上传到图片服务器...") img_url = self.upload_image_to_cos(local_file) self.amazon_log(f" ✅ 图片服务器上传成功: {img_url[:60]}...") except Exception as e: self.amazon_log(f" ❌ 图片服务器上传失败: {str(e)}") fail_count += 1 continue # 调用亚马逊API更换图片 try: self.amazon_log(f" 🔄 调用亚马逊API更换{type_name}...") resp = self.amazon_uploader.patch_image(token, image_type, img_url, sku) if resp.status_code == 200: self.amazon_log(f" ✅ {type_name}更换成功") success_count += 1 else: self.amazon_log(f" ❌ {type_name}更换失败: HTTP {resp.status_code}") try: error_data = resp.json() self.amazon_log(f" 错误详情: {json.dumps(error_data, ensure_ascii=False)[:300]}") except: self.amazon_log(f" 响应: {resp.text[:300]}") fail_count += 1 except Exception as e: self.amazon_log(f" ❌ {type_name}更换异常: {str(e)}") fail_count += 1 # 避免请求过快 time.sleep(0.5) self.amazon_log(f"\n{'=' * 50}") self.amazon_log(f"✅ 处理完成！成功: {success_count}, 失败: {fail_count}") self.amazon_progress_var.set(f"上传完成！成功: {success_count}, 失败: {fail_count}") messagebox.showinfo("完成", f"亚马逊图片上传完成！\n成功: {success_count}\n失败: {fail_count}") except Exception as e: error_msg = f"处理出错: {str(e)}" self.amazon_log(f"❌ {error_msg}") self.amazon_progress_var.set("处理出错") messagebox.showerror("错误", error_msg) import traceback self.amazon_log(traceback.format_exc()) # ==================== 亚马逊详情图更换 ==================== def start_detail_upload(self): """开始亚马逊详情图上传 (A+ Content)""" if not self.asin_input.get().strip(): messagebox.showerror("错误", "请先输入ASIN！") return if not self.detail_image_dir.get(): messagebox.showerror("错误", "请先选择图片文件夹！") return # 保存配置 self.save_detail_config() thread = threading.Thread(target=self.process_detail_upload, daemon=True) thread.start() def save_detail_config(self): """保存详情图配置""" config = { "asin": self.asin_input.get(), "headline": self.detail_headline.get(), "body": self.detail_body.get(), "image_width": self.image_width.get(), "image_height": self.image_height.get(), "image_dir": self.detail_image_dir.get(), } try: with open(self.get_detail_config_path(), 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) except Exception: pass def process_detail_upload(self): """执行亚马逊详情图上传 (A+ Content)""" try: self.detail_progress_var.set("处理中...") self.detail_log("=" * 50) self.detail_log("开始亚马逊详情图上传 (A+ Content)...") asin = self.asin_input.get().strip() headline = self.detail_headline.get().strip() body_text = self.detail_body.get().strip() image_dir = self.detail_image_dir.get() # 获取图片尺寸 try: target_width = int(self.image_width.get()) target_height = int(self.image_height.get()) except ValueError: self.detail_log("❌ 图片尺寸必须为数字") self.detail_progress_var.set("上传失败") return self.detail_log(f"ASIN: {asin}") self.detail_log(f"标题: {headline}") self.detail_log(f"描述: {body_text}") self.detail_log(f"图片尺寸: {target_width}x{target_height}") self.detail_log(f"图片文件夹: {image_dir}") # 获取图片文件 image_files = self.get_image_files(image_dir) if not image_files: self.detail_log("❌ 未找到任何图片文件") self.detail_progress_var.set("上传失败") return self.detail_log(f"\n找到 {len(image_files)} 张图片") for i, f in enumerate(image_files, start=1): self.detail_log(f" {i}. {os.path.basename(f)}") # 尝试导入PIL try: from PIL import Image from io import BytesIO HAS_PIL = True except ImportError: HAS_PIL = False self.detail_log("⚠️ 未安装Pillow，将使用原始图片尺寸") # 获取Access Token self.detail_log("\n🔑 获取Access Token...") try: token = self.amazon_uploader.get_access_token() self.detail_log("✅ Token获取成功") except Exception as e: self.detail_log(f"❌ 获取Token失败: {str(e)}") self.detail_progress_var.set("上传失败") return # 处理每张图片 success_count = 0 fail_count = 0 upload_ids = [] for i, local_file in enumerate(image_files, start=1): filename = os.path.basename(local_file) self.detail_log(f"\n处理第 {i}/{len(image_files)} 张图片: {filename}") try: # 读取图片 with open(local_file, 'rb') as f: image_binary = f.read() # 调整尺寸 if HAS_PIL: try: img = Image.open(BytesIO(image_binary)) original_size = img.size # 调整尺寸 img_resized = img.resize((target_width, target_height), Image.LANCZOS) buffer = BytesIO() # 判断原图格式 if filename.lower().endswith('.png'): img_resized.save(buffer, format="PNG") else: img_resized.save(buffer, format="JPEG") image_binary = buffer.getvalue() self.detail_log(f" 图片尺寸调整: {original_size} -> {target_width}x{target_height}") except Exception as e: self.detail_log(f" 图片调整失败: {str(e)}，使用原图") # 计算MD5 md5_hash = hashlib.md5(image_binary).digest() content_md5 = base64.b64encode(md5_hash).decode('utf-8') # 确定Content-Type content_type = "image/jpeg" if filename.lower().endswith('.png'): content_type = "image/png" self.detail_log(f" 图片大小: {len(image_binary)} bytes") # 创建上传目标 upload_data = self.amazon_uploader.create_upload_destination(token, content_md5, content_type) upload_id = upload_data.get("uploadDestinationId") upload_url = upload_data.get("url") if not upload_id or not upload_url: self.detail_log(f" ❌ 创建上传目标失败") fail_count += 1 continue self.detail_log(f" ✅ 上传目标创建成功") # 上传到S3 self.amazon_uploader.upload_to_s3(upload_url, image_binary, content_type) self.detail_log(f" ✅ S3上传成功") upload_ids.append({ "upload_id": upload_id, "width": target_width, "height": target_height, "alt_text": f"Product detail image {i}" }) success_count += 1 time.sleep(0.5) except Exception as e: self.detail_log(f" ❌ 处理失败: {str(e)}") fail_count += 1 if not upload_ids: self.detail_log("❌ 没有成功上传的图片") self.detail_progress_var.set("上传失败") return # 构建A+ Content内容文档 self.detail_log(f"\n📝 构建A+ Content文档 ({len(upload_ids)} 个模块)...") # 构建模块列表 - 每张图片一个模块 content_module_list = [] for i, upload_info in enumerate(upload_ids): module = { "contentModuleType": "STANDARD_HEADER_IMAGE_TEXT", "standardHeaderImageText": { "headline": { "value": headline if i == 0 else f"{headline} - {i+1}", "decoratorSet": [] }, "block": { "image": { "uploadDestinationId": upload_info["upload_id"], "imageCropSpecification": { "size": { "width": {"value": upload_info["width"], "units": "pixels"}, "height": {"value": upload_info["height"], "units": "pixels"} }, "offset": { "x": {"value": 0, "units": "pixels"}, "y": {"value": 0, "units": "pixels"} } }, "altText": upload_info["alt_text"] }, "headline": { "value": f"Image {i+1}", "decoratorSet": [] }, "body": { "textList": [ { "value": body_text if i == 0 else f"Product detail view {i+1}", "decoratorSet": [] } ] } } } } content_module_list.append(module) # 构建content文档 content_doc = { "name": f"Product Images for {asin}", "contentType": "EBC", "locale": "en-US", "contentModuleList": content_module_list } # 创建内容文档 self.detail_log("📤 创建A+ Content文档...") try: content_key = self.amazon_uploader.create_content_document(token, content_doc) self.detail_log(f"✅ Content文档创建成功，Content Key: {content_key}") except Exception as e: self.detail_log(f"❌ 创建Content文档失败: {str(e)}") self.detail_progress_var.set("上传失败") return # 绑定ASIN self.detail_log(f"🔗 绑定ASIN: {asin}...") try: self.amazon_uploader.bind_asin_to_content(token, content_key, asin) self.detail_log(f"✅ ASIN绑定成功") except Exception as e: self.detail_log(f"❌ 绑定ASIN失败: {str(e)}") self.detail_progress_var.set("上传失败") return # 提交审核 self.detail_log("📋 提交审核...") try: self.amazon_uploader.submit_for_approval(token, content_key) self.detail_log(f"✅ 审核提交成功") self.detail_log(f"⚠️ 审核通常需要 1-7 个工作日") except Exception as e: self.detail_log(f"❌ 提交审核失败: {str(e)}") self.detail_progress_var.set("上传失败") return self.detail_log(f"\n{'=' * 50}") self.detail_log(f"✅ 处理完成！成功: {success_count}, 失败: {fail_count}") self.detail_log(f" Content Key: {content_key}") self.detail_log(f" ASIN: {asin}") self.detail_progress_var.set(f"上传完成！成功: {success_count}, 失败: {fail_count}") messagebox.showinfo("完成", f"亚马逊详情图上传完成！\n成功: {success_count}\n失败: {fail_count}\n\nContent Key: {content_key}\nASIN: {asin}\n\n审核通常需要 1-7 个工作日") except Exception as e: error_msg = f"处理出错: {str(e)}" self.detail_log(f"❌ {error_msg}") self.detail_progress_var.set("处理出错") messagebox.showerror("错误", error_msg) import traceback self.detail_log(traceback.format_exc()) def main(): root = tk.Tk() app = ImageMatcherApp(root) root.mainloop() if __name__ == "__main__": main()