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 import time class AmazonImageUploader: """亚马逊图片上传类""" def __init__(self): self.client_id = "amzn1.application-oa2-client.9f775d35077a4c7598116e5a89b60222" self.client_secret = "amzn1.oa2-cs.v1.6909777baa82b796555a431cd23709342e8e01434abcad1f8b59cadf048b51c3" self.refresh_token = "Atzr|IwEBICh_fojh7qAfrePzSjOZhQxTnHRZCYOuWNvSAHGaBs1WfkjR4qjC1Dmv0uK3oyq_L-R9SRKK4cYmPwF1MQXQdqKlpzOAGItWeXnNzh-SyhHHISv0_vCt3D6hluVmIEmF2kA7ro_XF-PslD41g3Eq6bv0Qq4y4pQbkDj-9dudDuz7nku8UL-nc4EWgJCrHCJTAs8TK48Yvrfi3E35uv97tYhs6qb8D9d4dR9C7daaC49eoDqP88qyUHThQV1Kqon8xukx4AT2_wLqWNcabpADzuK-5w7D-vvFFtvzfkbmtfzoPeEdHPulG6scc59BPFTyBlGU7a3d1JSFg2ROHIlrljRm" self.aws_access_key = "AKIA24CLQSVTW2K4S6RP" self.aws_secret_key = "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" self.aws_region = "us-east-1" self.seller_id = "A1H50Z0WYYLEGQ" self.marketplace_id = "ATVPDKIKX0DER" self.product_type = "TABLE" 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) class ImageMatcherApp: def __init__(self, root): self.root = root self.root.title("图片匹配与上传工具 - 亚马逊版") self.root.geometry("950x750") # 存储文件路径 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() # 状态栏 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 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 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 get_config_path(self): """获取配置文件路径""" return os.path.join(os.path.dirname(os.path.abspath(__file__)), "amazon_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 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 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 main(): root = tk.Tk() app = ImageMatcherApp(root) root.mainloop() if __name__ == "__main__": main()