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): return hashlib.sha256(data.encode("utf-8")).hexdigest() def hmac_sha256(self, key, msg): return hmac.new(key, msg.encode("utf-8"), 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.encode("utf-8")).hexdigest() 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(6, 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) self.amazon_template_path = tk.StringVar() ttk.Entry(tab, textvariable=self.amazon_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_amazon_template).grid(row=1, column=2, pady=5, padx=5) # SKU输入（手动输入，不是从表格读取） ttk.Label(tab, text="SKU (手动输入):").grid(row=2, 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=2, column=1, sticky=tk.W, pady=5, padx=5) # 主图列选择 ttk.Label(tab, text="主图列名:").grid(row=3, column=0, sticky=tk.W, pady=5) self.main_image_column = tk.StringVar(value="产品主图") ttk.Entry(tab, textvariable=self.main_image_column, width=30).grid(row=3, column=1, sticky=tk.W, pady=5, padx=5) # 附图列选择 ttk.Label(tab, text="附图列名(逗号分隔):").grid(row=4, column=0, sticky=tk.W, pady=5) self.other_image_columns = tk.StringVar(value="上架图片1,上架图片2,上架图片3,上架图片4,上架图片5,上架图片6,上架图片7,上架图片8") ttk.Entry(tab, textvariable=self.other_image_columns, width=60).grid(row=4, column=1, sticky=(tk.W, tk.E), pady=5, padx=5) # 按钮框架 button_frame = ttk.Frame(tab) button_frame.grid(row=5, column=0, columnspan=3, pady=10) ttk.Button(button_frame, text="保存配置", command=self.save_amazon_config, width=12).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="加载配置", command=self.load_amazon_config, width=12).pack(side=tk.LEFT, padx=5) 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=6, 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=7, 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(8, 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) # 模板文件选择 ttk.Label(tab, text="模板文件:").grid(row=1, column=0, sticky=tk.W, pady=5) self.detail_template_path = tk.StringVar() ttk.Entry(tab, textvariable=self.detail_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_detail_template).grid(row=1, column=2, pady=5, padx=5) # ASIN输入 ttk.Label(tab, text="ASIN (手动输入):").grid(row=2, 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=2, column=1, sticky=tk.W, pady=5, padx=5) # 图片URL列选择（详情图） ttk.Label(tab, text="详情图列名(逗号分隔):").grid(row=3, column=0, sticky=tk.W, pady=5) self.detail_image_columns = tk.StringVar(value="详情图URL1,详情图URL2,详情图URL3") ttk.Entry(tab, textvariable=self.detail_image_columns, 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.image_width = tk.StringVar(value="970") ttk.Entry(tab, textvariable=self.image_width, width=15).grid(row=4, column=1, sticky=tk.W, pady=5, padx=5) ttk.Label(tab, text="图片高度(像素):").grid(row=5, column=0, sticky=tk.W, pady=5) self.image_height = tk.StringVar(value="600") ttk.Entry(tab, textvariable=self.image_height, width=15).grid(row=5, column=1, sticky=tk.W, pady=5, padx=5) # 标题和描述设置 ttk.Label(tab, text="模块标题:").grid(row=6, column=0, sticky=tk.W, pady=5) self.module_headline = tk.StringVar(value="Premium Quality Product") ttk.Entry(tab, textvariable=self.module_headline, width=50).grid(row=6, column=1, sticky=tk.W, pady=5, padx=5) ttk.Label(tab, text="模块描述:").grid(row=7, column=0, sticky=tk.W, pady=5) self.module_body = tk.StringVar(value="High-quality product designed for your needs.") ttk.Entry(tab, textvariable=self.module_body, width=50).grid(row=7, column=1, sticky=tk.W, pady=5, padx=5) # 按钮框架 button_frame = ttk.Frame(tab) button_frame.grid(row=8, column=0, columnspan=3, pady=10) ttk.Button(button_frame, text="保存配置", command=self.save_detail_config, width=12).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="加载配置", command=self.load_detail_config, width=12).pack(side=tk.LEFT, padx=5) 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=9, 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=10, 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_template(self): filename = filedialog.askopenfilename( title="选择模板文件", filetypes=[("Excel文件", "*.xlsx *.xls"), ("所有文件", "*.*")] ) if filename: self.amazon_template_path.set(filename) self.amazon_log(f"已选择模板文件: {filename}") def select_detail_template(self): filename = filedialog.askopenfilename( title="选择模板文件", filetypes=[("Excel文件", "*.xlsx *.xls"), ("所有文件", "*.*")] ) if filename: self.detail_template_path.set(filename) self.detail_log(f"已选择模板文件: {filename}") 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_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 save_amazon_config(self): """保存亚马逊配置""" config = { "sku": self.sku_input.get(), "main_image_column": self.main_image_column.get(), "other_image_columns": self.other_image_columns.get(), "template_path": self.amazon_template_path.get() } try: with open(self.get_config_path(), 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) self.amazon_log("✅ 配置已保存") except Exception as e: self.amazon_log(f"❌ 保存配置失败: {str(e)}") def load_amazon_config(self): """加载亚马逊配置""" try: config_path = self.get_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")) self.main_image_column.set(config.get("main_image_column", "产品主图")) self.other_image_columns.set(config.get("other_image_columns", "上架图片1,上架图片2,上架图片3,上架图片4,上架图片5,上架图片6,上架图片7,上架图片8")) if config.get("template_path"): self.amazon_template_path.set(config.get("template_path")) self.amazon_log("✅ 配置已加载") else: self.amazon_log("⚠️ 未找到配置文件，使用默认配置") except Exception as e: self.amazon_log(f"❌ 加载配置失败: {str(e)}") def save_detail_config(self): """保存详情图配置""" config = { "asin": self.asin_input.get(), "detail_image_columns": self.detail_image_columns.get(), "template_path": self.detail_template_path.get(), "image_width": self.image_width.get(), "image_height": self.image_height.get(), "module_headline": self.module_headline.get(), "module_body": self.module_body.get() } try: with open(self.get_detail_config_path(), 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) self.detail_log("✅ 配置已保存") except Exception as e: self.detail_log(f"❌ 保存配置失败: {str(e)}") def load_detail_config(self): """加载详情图配置""" 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_image_columns.set(config.get("detail_image_columns", "详情图URL1,详情图URL2,详情图URL3")) self.image_width.set(config.get("image_width", "970")) self.image_height.set(config.get("image_height", "600")) self.module_headline.set(config.get("module_headline", "Premium Quality Product")) self.module_body.set(config.get("module_body", "High-quality product designed for your needs.")) if config.get("template_path"): self.detail_template_path.set(config.get("template_path")) self.detail_log("✅ 配置已加载") else: self.detail_log("⚠️ 未找到配置文件，使用默认配置") except Exception as e: self.detail_log(f"❌ 加载配置失败: {str(e)}") def load_config(self): """启动时自动加载所有配置""" # 加载亚马逊上传配置 try: config_path = self.get_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")) self.main_image_column.set(config.get("main_image_column", "产品主图")) self.other_image_columns.set(config.get("other_image_columns", "上架图片1,上架图片2,上架图片3,上架图片4,上架图片5,上架图片6,上架图片7,上架图片8")) if config.get("template_path"): self.amazon_template_path.set(config.get("template_path")) 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_image_columns.set(config.get("detail_image_columns", "详情图URL1,详情图URL2,详情图URL3")) self.image_width.set(config.get("image_width", "970")) self.image_height.set(config.get("image_height", "600")) self.module_headline.set(config.get("module_headline", "Premium Quality Product")) self.module_body.set(config.get("module_body", "High-quality product designed for your needs.")) if config.get("template_path"): self.detail_template_path.set(config.get("template_path")) 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, 7): 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: 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.amazon_template_path.get(): messagebox.showerror("错误", "请先选择模板文件！") return if not self.sku_input.get().strip(): messagebox.showerror("错误", "请先输入SKU！") return thread = threading.Thread(target=self.process_amazon_upload, daemon=True) thread.start() def process_amazon_upload(self): """执行亚马逊图片上传""" try: self.amazon_progress_var.set("处理中...") self.amazon_log("=" * 50) self.amazon_log("开始亚马逊图片上传...") template_path = self.amazon_template_path.get() sku = self.sku_input.get().strip() main_img_col_name = self.main_image_column.get() other_cols_str = self.other_image_columns.get() other_col_names = [col.strip() for col in other_cols_str.split(',') if col.strip()] self.amazon_log(f"读取模板文件: {template_path}") self.amazon_log(f"使用的SKU: {sku}") # 读取Excel df = pd.read_excel(template_path) self.amazon_log(f"找到 {len(df)} 行数据") # 检查必要的列 if main_img_col_name not in df.columns: self.amazon_log(f"❌ 未找到主图列: {main_img_col_name}") self.amazon_progress_var.set("上传失败") return # 检查附图列 existing_other_cols = [col for col in other_col_names if col in df.columns] if not existing_other_cols: self.amazon_log("⚠️ 未找到任何附图列") self.amazon_log(f"主图列: {main_img_col_name}") self.amazon_log(f"附图列: {existing_other_cols}") # 获取Access Token self.amazon_log("🔑 获取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 total_rows = len(df) for idx, row in df.iterrows(): self.amazon_log(f"\n处理第 {idx+1}/{total_rows} 行") # 处理主图 main_img_url = str(row[main_img_col_name]).strip() if main_img_url and main_img_url != 'nan' and main_img_url != 'None': if main_img_url.startswith('http'): try: resp = self.amazon_uploader.patch_image(token, "main", main_img_url, sku) if resp.status_code == 200: self.amazon_log(f" ✅ 主图上传成功") success_count += 1 else: self.amazon_log(f" ❌ 主图上传失败: {resp.status_code}") fail_count += 1 # 打印错误详情 try: error_data = resp.json() self.amazon_log(f" 错误详情: {json.dumps(error_data, ensure_ascii=False)[:200]}") except: pass except Exception as e: self.amazon_log(f" ❌ 主图上传异常: {str(e)}") fail_count += 1 else: self.amazon_log(f" ⚠️ 主图URL无效: {main_img_url[:50]}...") else: self.amazon_log(f" ⚠️ 主图为空，跳过") # 处理附图 for i, col_name in enumerate(existing_other_cols, start=1): img_url = str(row[col_name]).strip() if img_url and img_url != 'nan' and img_url != 'None': if img_url.startswith('http'): try: image_type = f"other_{i}" resp = self.amazon_uploader.patch_image(token, image_type, img_url, sku) if resp.status_code == 200: self.amazon_log(f" ✅ 附图{i}上传成功") success_count += 1 else: self.amazon_log(f" ❌ 附图{i}上传失败: {resp.status_code}") fail_count += 1 except Exception as e: self.amazon_log(f" ❌ 附图{i}上传异常: {str(e)}") fail_count += 1 else: self.amazon_log(f" ⚠️ 附图{i} URL无效: {img_url[:50]}...") else: self.amazon_log(f" ⚠️ 附图{i}为空，跳过") # 避免请求过快 time.sleep(0.5) self.amazon_log(f"\n✅ 处理完成！成功: {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.detail_template_path.get(): messagebox.showerror("错误", "请先选择模板文件！") return if not self.asin_input.get().strip(): messagebox.showerror("错误", "请先输入ASIN！") return thread = threading.Thread(target=self.process_detail_upload, daemon=True) thread.start() def process_detail_upload(self): """执行亚马逊详情图上传 (A+ Content)""" try: self.detail_progress_var.set("处理中...") self.detail_log("=" * 50) self.detail_log("开始亚马逊详情图上传 (A+ Content)...") template_path = self.detail_template_path.get() asin = self.asin_input.get().strip() detail_cols_str = self.detail_image_columns.get() detail_col_names = [col.strip() for col in detail_cols_str.split(',') if col.strip()] # 获取图片尺寸 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 module_headline = self.module_headline.get().strip() module_body = self.module_body.get().strip() self.detail_log(f"读取模板文件: {template_path}") self.detail_log(f"使用的ASIN: {asin}") self.detail_log(f"图片尺寸: {target_width}x{target_height}") # 读取Excel df = pd.read_excel(template_path) self.detail_log(f"找到 {len(df)} 行数据") # 检查详情图列 existing_detail_cols = [col for col in detail_col_names if col in df.columns] if not existing_detail_cols: self.detail_log("❌ 未找到任何详情图列") self.detail_progress_var.set("上传失败") return self.detail_log(f"详情图列: {existing_detail_cols}") # 尝试导入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("🔑 获取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 total_rows = min(len(df), 1) # 每次只处理一行（A+ Content每次只能绑定一个ASIN） # 只处理第一行 row = df.iloc[0] # 收集所有图片URL image_urls = [] for col_name in existing_detail_cols: img_url = str(row[col_name]).strip() if img_url and img_url != 'nan' and img_url != 'None' and img_url.startswith('http'): image_urls.append(img_url) if not image_urls: self.detail_log("❌ 未找到有效的图片URL") self.detail_progress_var.set("上传失败") return self.detail_log(f"找到 {len(image_urls)} 张详情图") # 处理每张图片 upload_ids = [] for i, img_url in enumerate(image_urls, start=1): self.detail_log(f"\n处理第 {i} 张图片: {img_url[:60]}...") try: # 下载图片 img_resp = requests.get(img_url, timeout=30) img_resp.raise_for_status() image_binary = img_resp.content # 调整尺寸 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 img_url.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 img_url.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("\n📝 构建A+ Content文档...") # 构建模块列表 - 每张图片一个模块 content_module_list = [] for i, upload_info in enumerate(upload_ids): module = { "contentModuleType": "STANDARD_HEADER_IMAGE_TEXT", "standardHeaderImageText": { "headline": { "value": module_headline if i == 0 else f"{module_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": module_body 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✅ 处理完成！成功: {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()