import json import os import re import requests from pathlib import Path from urllib.parse import urlparse from tkinter import * from tkinter import ttk, filedialog, messagebox, scrolledtext import threading from datetime import datetime class ProductDownloaderGUI: def __init__(self, root): self.root = root self.root.title("商品资源下载器") self.root.geometry("900x650") self.root.resizable(True, True) # 设置样式 style = ttk.Style() style.theme_use('clam') # 创建主框架 main_frame = ttk.Frame(root, padding="10") main_frame.pack(fill=BOTH, expand=True) # 输入区域 input_frame = ttk.LabelFrame(main_frame, text="商品数据输入", padding="10") input_frame.pack(fill=BOTH, expand=True, pady=(0, 10)) # 输入框 self.input_text = scrolledtext.ScrolledText(input_frame, height=18, font=("Consolas", 10)) self.input_text.pack(fill=BOTH, expand=True) # 输出目录选择 output_frame = ttk.LabelFrame(main_frame, text="输出设置", padding="10") output_frame.pack(fill=X, pady=(0, 10)) dir_select_frame = ttk.Frame(output_frame) dir_select_frame.pack(fill=X) self.output_dir = StringVar() self.output_entry = ttk.Entry(dir_select_frame, textvariable=self.output_dir) self.output_entry.pack(side=LEFT, fill=X, expand=True, padx=(0, 5)) browse_btn = ttk.Button(dir_select_frame, text="浏览...", command=self.browse_output_dir) browse_btn.pack(side=RIGHT) # 进度条 self.progress_var = DoubleVar() self.progress_bar = ttk.Progressbar(main_frame, variable=self.progress_var, maximum=100) self.progress_bar.pack(fill=X, pady=(0, 10)) # 状态标签 self.status_label = ttk.Label(main_frame, text="就绪") self.status_label.pack(pady=(0, 10)) # 按钮区域 button_frame = ttk.Frame(main_frame) button_frame.pack(fill=X) # 启动按钮 self.start_btn = ttk.Button(button_frame, text="▶ 启动下载", command=self.start_download, width=15) self.start_btn.pack(side=LEFT, padx=5) clear_btn = ttk.Button(button_frame, text="清空输入", command=self.clear_input) clear_btn.pack(side=LEFT, padx=5) # 退出按钮 exit_btn = ttk.Button(button_frame, text="退出", command=self.root.quit) exit_btn.pack(side=RIGHT, padx=5) # 设置下载会话 self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) def browse_output_dir(self): """选择输出目录""" directory = filedialog.askdirectory() if directory: self.output_dir.set(directory) def clear_input(self): """清空输入框""" self.input_text.delete(1.0, END) def extract_urls(self, data): """从JSON数据中提取图片和视频URL - 适配多种数据结构""" urls = { 'main_images': [], # 主图 'detail_images': [], # 详情图 'videos': [] # 视频 } seen_urls = set() def add_url(url, url_type): """添加URL，避免重复""" if url and url not in seen_urls: seen_urls.add(url) if url_type == 'main': urls['main_images'].append(url) elif url_type == 'detail': urls['detail_images'].append(url) elif url_type == 'video': urls['videos'].append(url) try: # 获取data部分 data_content = data.get('data', {}) # 1. 提取主图 - pictures字段 pictures = data_content.get('pictures', []) for pic in pictures: add_url(pic, 'main') # 2. 提取主图 - opt_pictures字段 opt_pictures = data_content.get('opt_pictures', []) for pic in opt_pictures: add_url(pic, 'main') # 3. 从SKU中提取主图 skus = data_content.get('skus', []) for sku in skus: sku_pic = sku.get('picture') if sku_pic: add_url(sku_pic, 'main') # 4. 提取详情图 - pic_content.contents pic_content = data_content.get('pic_content', {}) contents = pic_content.get('contents', []) for content in contents: if content and isinstance(content, str): add_url(content, 'detail') # 5. 提取视频 video_infos = data_content.get('product_video_infos', []) for video_info in video_infos: video_url = video_info.get('url') if video_url: add_url(video_url, 'video') # 视频封面图作为主图 main_picture = video_info.get('mainPicture') if main_picture: add_url(main_picture, 'main') # 6. 使用正则表达式补充提取所有图片URL data_str = json.dumps(data, ensure_ascii=False) # 匹配图片URL image_pattern = r'https?://[^\s"\']+\.(?:jpg|jpeg|png|gif|webp)(?:\?[^\s"\']*)?' all_images = re.findall(image_pattern, data_str, re.IGNORECASE) # 视频匹配 video_pattern = r'https?://[^\s"\']+\.(?:mp4|mov|avi|flv|webm)(?:\?[^\s"\']*)?' all_videos = re.findall(video_pattern, data_str, re.IGNORECASE) # 分类正则匹配到的图片 for img_url in all_images: if img_url in seen_urls: continue # 根据URL特征分类 if 'head_photo' in img_url or 'main' in img_url.lower() or 'sku' in img_url.lower(): add_url(img_url, 'main') # 添加正则匹配到的视频 for video_url in all_videos: if video_url not in seen_urls: add_url(video_url, 'video') except Exception as e: print(f"提取URL时出错: {e}") # 限制数量 urls['main_images'] = urls['main_images'][:15] urls['detail_images'] = urls['detail_images'][:40] urls['videos'] = urls['videos'][:5] return urls def extract_product_info(self, data): """提取商品简要信息""" product_info = { 'id': '未知', 'name': '未命名商品', 'min_price': '未知', 'origin_price': '未知', 'month_saled': '未知', 'month_saled_content': '未知', 'sku_label': '无', 'brand': '未知', 'sku_count': 0, 'image_count': 0, 'video_count': 0, 'sku_list': [], 'standard_params': [], 'delivery_info': '', 'activity_infos': [] } try: data_content = data.get('data', {}) # 基础信息 product_info['id'] = data_content.get('id', '未知') product_info['name'] = data_content.get('name', '未命名商品') product_info['min_price'] = data_content.get('min_price', '未知') product_info['month_saled_content'] = data_content.get('month_saled_content', '未知') # 提取划线价（从unify_price中获取） unify_price = data_content.get('unify_price', {}) product_info['origin_price'] = unify_price.get('underlined_price', '未知') if product_info['origin_price'] == '未知': # 如果unify_price中没有，从skus中获取 skus = data_content.get('skus', []) if skus: product_info['origin_price'] = skus[0].get('origin_price', '未知') # SKU信息 skus = data_content.get('skus', []) product_info['sku_count'] = len(skus) product_info['sku_label'] = data_content.get('sku_label', '规格') # 提取每个SKU的详细信息 for sku in skus: sku_info = { 'spec': sku.get('spec', ''), 'price': sku.get('price', ''), 'origin_price': sku.get('origin_price', ''), 'stock': sku.get('stock', ''), 'upccode': sku.get('upccode', ''), 'picture': sku.get('picture', '') } product_info['sku_list'].append(sku_info) # 提取标准参数 standard_list = data_content.get('standard_productinfo_list', []) for item in standard_list: product_info['standard_params'].append({ 'fieldName': item.get('fieldName', ''), 'value': item.get('value', '') }) # 提取品牌（从标准参数中获取） for param in product_info['standard_params']: if param['fieldName'] == '品牌': product_info['brand'] = param['value'] break if product_info['brand'] == '未知': # 如果标准参数中没有，尝试从名称中提取 name = product_info['name'] brand_keywords = ['农夫山泉', '可口可乐', '百事', '康师傅', '统一', '伊利', '蒙牛', '华为', '小米', '苹果'] for brand in brand_keywords: if brand in name: product_info['brand'] = brand break # 图片数量 pictures = data_content.get('pictures', []) product_info['image_count'] = len(pictures) # 视频数量 video_infos = data_content.get('product_video_infos', []) product_info['video_count'] = len(video_infos) # 配送信息 delivery_info = data_content.get('delivery_info', {}) if delivery_info: product_info['delivery_info'] = delivery_info.get('delivery_content', '') # 活动信息 activity_infos = data_content.get('activity_infos', []) for act in activity_infos: product_info['activity_infos'].append(act.get('activity_text', '')) except Exception as e: print(f"提取商品信息时出错: {e}") return product_info def save_product_info(self, product_info, output_path, urls): """保存商品简要信息到data.txt""" try: txt_path = output_path / 'data.txt' with open(txt_path, 'w', encoding='utf-8') as f: # 商品名称 f.write(f"商品: {product_info.get('name', '未命名商品')}\n") # 价格信息 min_price = product_info.get('min_price', '未知') origin_price = product_info.get('origin_price', '未知') if min_price != '未知' and origin_price != '未知': f.write(f"价格: {min_price}元起 (划线价 {origin_price}元)\n") # 计算折扣 try: discount = int((1 - float(min_price) / float(origin_price)) * 100) if discount > 0: f.write(f"折扣: {discount}% OFF\n") except: pass elif min_price != '未知': f.write(f"价格: {min_price}元起\n") else: f.write(f"价格: 未知\n") # 销量 month_saled = product_info.get('month_saled_content', '未知') f.write(f"销量: {month_saled}\n") f.write("\n") # SKU信息 sku_list = product_info.get('sku_list', []) sku_label = product_info.get('sku_label', '规格') if sku_list: f.write(f"可选{sku_label} (共{len(sku_list)}种):\n") for i, sku in enumerate(sku_list, 1): spec = sku.get('spec', '') price = sku.get('price', '') stock = sku.get('stock', '') upccode = sku.get('upccode', '') f.write(f" {i}. {spec} - {price}元") if stock and stock != '未知' and stock != '': f.write(f" (库存{stock})") if upccode: f.write(f" 代码:{upccode}") f.write("\n") else: f.write(f"无{sku_label}信息\n") f.write("\n") # 参数信息 standard_params = product_info.get('standard_params', []) if standard_params: params_str = " | ".join([f"{p['fieldName']}:{p['value']}" for p in standard_params]) f.write(f"参数: {params_str}\n") f.write("\n") # 配送信息 delivery_info = product_info.get('delivery_info', '') if delivery_info: f.write(f"配送: {delivery_info}\n") # 活动信息 activity_infos = product_info.get('activity_infos', []) for act in activity_infos: if act: f.write(f"优惠: {act}\n") return True except Exception as e: print(f"保存data.txt失败: {e}") return False def download_file(self, url, filepath, timeout=30): """下载文件""" try: print(url) print(filepath) response = self.session.get(url, timeout=timeout, stream=True) if response.status_code == 200: with open(filepath, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) return True else: return False except Exception as e: print(f"下载失败 {url}: {e}") return False def sanitize_filename(self, name): """清理文件名中的非法字符""" name = re.sub(r'[<>:"/\\|?*]', '_', name) if len(name) > 100: name = name[:100] return name.strip() def download_resources(self, product_name, output_path, urls, progress_callback): """下载所有资源""" results = { 'success': True, 'downloaded': 0, 'total': 0, 'errors': [] } total_items = len(urls['main_images']) + len(urls['detail_images']) + len(urls['videos']) # 下载主图 for i, url in enumerate(urls['main_images'], 1): results['total'] += 1 path = urlparse(url).path ext = os.path.splitext(path)[1] or '.jpg' if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: ext = '.jpg' filename = f"zhutu_{i}{ext}" filepath = output_path / filename progress_callback(f"下载主图 {i}/{len(urls['main_images'])}...", int(results['total'] / total_items * 100) if total_items > 0 else 0) if self.download_file(url, filepath): results['downloaded'] += 1 else: results['errors'].append(f"主图{i}: {url}") # 下载详情图 for i, url in enumerate(urls['detail_images'], 1): results['total'] += 1 path = urlparse(url).path ext = os.path.splitext(path)[1] or '.jpg' if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: ext = '.jpg' filename = f"xiangqing_{i}{ext}" filepath = output_path / filename progress_callback(f"下载详情图 {i}/{len(urls['detail_images'])}...", int(results['total'] / total_items * 100) if total_items > 0 else 0) if self.download_file(url, filepath): results['downloaded'] += 1 else: results['errors'].append(f"详情图{i}: {url}") # 下载视频 for i, url in enumerate(urls['videos'], 1): results['total'] += 1 path = urlparse(url).path ext = os.path.splitext(path)[1] or '.mp4' if ext not in ['.mp4', '.mov', '.avi', '.webm']: ext = '.mp4' filename = f"shipin_{i}{ext}" filepath = output_path / filename progress_callback(f"下载视频 {i}/{len(urls['videos'])}...", int(results['total'] / total_items * 100) if total_items > 0 else 0) if self.download_file(url, filepath): results['downloaded'] += 1 else: results['errors'].append(f"视频{i}: {url}") return results def start_download(self): """启动下载任务""" input_data = self.input_text.get(1.0, END).strip() if not input_data: messagebox.showerror("错误", "请输入商品数据") return output_dir = self.output_dir.get() if not output_dir: messagebox.showerror("错误", "请选择输出目录") return self.start_btn.config(state=DISABLED, text="⏳ 下载中...") self.progress_var.set(0) thread = threading.Thread(target=self.download_task, args=(input_data, output_dir)) thread.daemon = True thread.start() def download_task(self, input_data, output_dir): """下载任务""" try: # 解析JSON data = json.loads(input_data) # 提取商品信息 product_info = self.extract_product_info(data) # 清理产品名作为文件夹名 safe_name = self.sanitize_filename(product_info['name']) # 创建产品文件夹 product_path = Path(output_dir) / safe_name product_path.mkdir(exist_ok=True) # 提取URL self.update_status_thread("正在分析数据，提取资源链接...") urls = self.extract_urls(data) # 保存商品简要信息到data.txt self.update_status_thread("正在保存商品信息...") if self.save_product_info(product_info, product_path, urls): self.update_status_thread("商品信息已保存") else: self.update_status_thread("保存商品信息失败") total_urls = len(urls['main_images']) + len(urls['detail_images']) + len(urls['videos']) if total_urls == 0: self.root.after(0, messagebox.showwarning, "警告", "未找到任何图片或视频资源\n但data.txt已保存") self.update_status_thread("未找到资源") self.enable_button_thread() return self.update_status_thread(f"发现 {len(urls['main_images'])} 张主图, {len(urls['detail_images'])} 张详情图, {len(urls['videos'])} 个视频") def update_progress(msg, progress): self.update_status_thread(msg) self.root.after(0, self.progress_var.set, progress) results = self.download_resources(safe_name, product_path, urls, update_progress) self.root.after(0, self.progress_var.set, 100) if results['downloaded'] > 0: msg = f"✅ 下载完成！\n\n成功下载 {results['downloaded']}/{results['total']} 个文件\n保存路径: {product_path}\n\n商品信息已保存为 data.txt" if results['errors']: msg += f"\n\n⚠️ 失败 {len(results['errors'])} 个文件" self.root.after(0, messagebox.showinfo, "完成", msg) self.update_status_thread(f"下载完成 - 成功 {results['downloaded']}/{results['total']}") else: self.root.after(0, messagebox.showerror, "错误", "下载失败，没有成功下载任何文件\n但data.txt已保存") self.update_status_thread("下载失败") except json.JSONDecodeError as e: self.root.after(0, messagebox.showerror, "JSON解析错误", f"输入的不是有效的JSON格式:\n{str(e)}") self.update_status_thread("JSON解析失败") except Exception as e: self.root.after(0, messagebox.showerror, "错误", f"处理过程中出现错误:\n{str(e)}") self.update_status_thread("处理失败") finally: self.enable_button_thread() def update_status_thread(self, message): """从线程更新状态""" self.root.after(0, lambda: self.status_label.config(text=message)) def enable_button_thread(self): """从线程启用按钮""" self.root.after(0, lambda: self.start_btn.config(state=NORMAL, text="▶ 启动下载")) def main(): root = Tk() app = ProductDownloaderGUI(root) root.mainloop() if __name__ == "__main__": main()