|
|
@@ -0,0 +1,467 @@
|
|
|
+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("900x700")
|
|
|
+ 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=15, 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)
|
|
|
+
|
|
|
+ # 信息显示区域
|
|
|
+ info_frame = ttk.LabelFrame(main_frame, text="商品信息", padding="10")
|
|
|
+ info_frame.pack(fill=BOTH, expand=True, pady=(0, 10))
|
|
|
+
|
|
|
+ self.info_text = scrolledtext.ScrolledText(info_frame, height=8, font=("微软雅黑", 10))
|
|
|
+ self.info_text.pack(fill=BOTH, expand=True)
|
|
|
+
|
|
|
+ # 进度条
|
|
|
+ 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 update_info_display(self, product_info):
|
|
|
+ """更新商品信息显示"""
|
|
|
+ self.info_text.delete(1.0, END)
|
|
|
+ info_str = f"""
|
|
|
+【商品名称】{product_info.get('name', '未知')}
|
|
|
+【商品描述】{product_info.get('description', '无')}
|
|
|
+【价格】¥{product_info.get('min_price', '未知')}
|
|
|
+【规格】{product_info.get('spec', '无')}
|
|
|
+【品牌】{product_info.get('brand', '未知')}
|
|
|
+【月销量】{product_info.get('month_saled', '未知')}
|
|
|
+ """
|
|
|
+ self.info_text.insert(1.0, info_str.strip())
|
|
|
+
|
|
|
+ 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:
|
|
|
+ # 从common_info中提取SKU主图
|
|
|
+ if 'data' in data and 'ext' in data['data']:
|
|
|
+ common_info = data['data']['ext'].get('common_info', {})
|
|
|
+
|
|
|
+ # 提取商品名称和描述
|
|
|
+ if 'skus' in common_info:
|
|
|
+ for sku in common_info['skus']:
|
|
|
+ if 'picture' in sku and sku['picture']:
|
|
|
+ add_url(sku['picture'], 'main')
|
|
|
+
|
|
|
+ # 从blocks中提取
|
|
|
+ if 'blocks' in data['data']['ext']:
|
|
|
+ blocks = data['data']['ext']['blocks']
|
|
|
+
|
|
|
+ # 提取主图（头图）
|
|
|
+ for block in blocks.get('banner', []):
|
|
|
+ s_type = block.get('s_type', '')
|
|
|
+
|
|
|
+ # 头图模块
|
|
|
+ if s_type == 'sm_type_detail_pop_head_photo':
|
|
|
+ head_info = block.get('data', {}).get('head_info', {})
|
|
|
+ pictures = head_info.get('pictures', [])
|
|
|
+ for pic in pictures:
|
|
|
+ add_url(pic, 'main')
|
|
|
+
|
|
|
+ # 商品详情图模块
|
|
|
+ elif s_type == 'sm_type_goods_detail_describe_non_food':
|
|
|
+ json_data = block.get('data', {}).get('json_data', {})
|
|
|
+ pic_content = json_data.get('pic_content', {})
|
|
|
+ contents = pic_content.get('contents', [])
|
|
|
+ for content in contents:
|
|
|
+ if content.startswith('http') and ('jpg' in content or 'png' in content or 'jpeg' in content):
|
|
|
+ add_url(content, 'detail')
|
|
|
+
|
|
|
+ # 推荐搭配模块中的图片
|
|
|
+ elif s_type == 'sm_type_goods_detail_match_product':
|
|
|
+ json_data = block.get('data', {}).get('json_data', {})
|
|
|
+ collocate = json_data.get('collocate_recommend', {})
|
|
|
+ spus = collocate.get('spus', [])
|
|
|
+ for spu in spus:
|
|
|
+ if 'picture' in spu and spu['picture']:
|
|
|
+ add_url(spu['picture'], 'detail')
|
|
|
+
|
|
|
+ # 从float模块中提取购物车相关图片
|
|
|
+ for block in blocks.get('float', []):
|
|
|
+ if block.get('s_type') == 'sm_type_cart_info':
|
|
|
+ data_json = block.get('data', {})
|
|
|
+ shopping_cart = data_json.get('shopping_cart', {})
|
|
|
+ if shopping_cart.get('shopping_cart_icon'):
|
|
|
+ add_url(shopping_cart['shopping_cart_icon'], 'detail')
|
|
|
+
|
|
|
+ # 使用正则表达式提取所有图片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
|
|
|
+ seen_urls.add(img_url)
|
|
|
+
|
|
|
+ # 根据URL特征分类
|
|
|
+ if 'head_photo' in img_url or 'main' in img_url.lower() or 'sku' in img_url.lower():
|
|
|
+ urls['main_images'].append(img_url)
|
|
|
+ else:
|
|
|
+ urls['detail_images'].append(img_url)
|
|
|
+
|
|
|
+ # 添加视频
|
|
|
+ for video_url in all_videos:
|
|
|
+ if video_url not in seen_urls:
|
|
|
+ seen_urls.add(video_url)
|
|
|
+ urls['videos'].append(video_url)
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"提取URL时出错: {e}")
|
|
|
+
|
|
|
+ # 限制数量
|
|
|
+ urls['main_images'] = urls['main_images'][:10]
|
|
|
+ urls['detail_images'] = urls['detail_images'][:30]
|
|
|
+ urls['videos'] = urls['videos'][:5]
|
|
|
+
|
|
|
+ return urls
|
|
|
+
|
|
|
+ def extract_product_info(self, data):
|
|
|
+ """提取商品信息"""
|
|
|
+ product_info = {
|
|
|
+ 'name': '未命名商品',
|
|
|
+ 'description': '无',
|
|
|
+ 'min_price': '未知',
|
|
|
+ 'spec': '无',
|
|
|
+ 'brand': '未知',
|
|
|
+ 'month_saled': '未知'
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ if 'data' in data and 'ext' in data['data']:
|
|
|
+ common_info = data['data']['ext'].get('common_info', {})
|
|
|
+
|
|
|
+ product_info['name'] = common_info.get('name', '未命名商品')
|
|
|
+ product_info['description'] = common_info.get('description', '无')
|
|
|
+ product_info['min_price'] = common_info.get('min_price', '未知')
|
|
|
+ product_info['spec'] = common_info.get('sku_label', '无')
|
|
|
+
|
|
|
+ # 提取月销量
|
|
|
+ if 'skus' in common_info and common_info['skus']:
|
|
|
+ sku = common_info['skus'][0]
|
|
|
+ if 'promotion_info' in sku:
|
|
|
+ match = re.search(r'月售(\d+)', sku.get('promotion_info', ''))
|
|
|
+ if match:
|
|
|
+ product_info['month_saled'] = match.group(1)
|
|
|
+
|
|
|
+ # 提取品牌
|
|
|
+ if 'name' in product_info and '农夫山泉' in product_info['name']:
|
|
|
+ product_info['brand'] = '农夫山泉'
|
|
|
+ elif 'name' in product_info and '可口可乐' in product_info['name']:
|
|
|
+ product_info['brand'] = '可口可乐'
|
|
|
+ elif 'name' in product_info and '百事' in product_info['name']:
|
|
|
+ product_info['brand'] = '百事'
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"提取商品信息时出错: {e}")
|
|
|
+
|
|
|
+ return product_info
|
|
|
+
|
|
|
+ def download_file(self, url, filepath, timeout=30):
|
|
|
+ """下载文件"""
|
|
|
+ try:
|
|
|
+ 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': []
|
|
|
+ }
|
|
|
+
|
|
|
+ # 创建子目录
|
|
|
+ main_img_dir = output_path / "主图"
|
|
|
+ detail_img_dir = output_path / "详情图"
|
|
|
+ video_dir = output_path / "视频"
|
|
|
+
|
|
|
+ main_img_dir.mkdir(exist_ok=True)
|
|
|
+ detail_img_dir.mkdir(exist_ok=True)
|
|
|
+ video_dir.mkdir(exist_ok=True)
|
|
|
+
|
|
|
+ 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 = main_img_dir / filename
|
|
|
+
|
|
|
+ progress_callback(f"下载主图 {i}/{len(urls['main_images'])}...",
|
|
|
+ int(results['total'] / total_items * 100))
|
|
|
+
|
|
|
+ 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 = detail_img_dir / filename
|
|
|
+
|
|
|
+ progress_callback(f"下载详情图 {i}/{len(urls['detail_images'])}...",
|
|
|
+ int(results['total'] / total_items * 100))
|
|
|
+
|
|
|
+ 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 = video_dir / filename
|
|
|
+
|
|
|
+ progress_callback(f"下载视频 {i}/{len(urls['videos'])}...",
|
|
|
+ int(results['total'] / total_items * 100))
|
|
|
+
|
|
|
+ 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)
|
|
|
+
|
|
|
+ # 更新信息显示
|
|
|
+ self.root.after(0, self.update_info_display, product_info)
|
|
|
+
|
|
|
+ # 清理产品名作为文件夹名
|
|
|
+ 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)
|
|
|
+
|
|
|
+ total_urls = len(urls['main_images']) + len(urls['detail_images']) + len(urls['videos'])
|
|
|
+
|
|
|
+ if total_urls == 0:
|
|
|
+ self.root.after(0, messagebox.showwarning, "警告", "未找到任何图片或视频资源")
|
|
|
+ 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)
|
|
|
+
|
|
|
+ # 更新进度到100%
|
|
|
+ self.root.after(0, self.progress_var.set, 100)
|
|
|
+
|
|
|
+ # 显示结果
|
|
|
+ if results['downloaded'] > 0:
|
|
|
+ msg = f"✅ 下载完成！\n\n成功下载 {results['downloaded']}/{results['total']} 个文件\n保存路径: {product_path}"
|
|
|
+ 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, "错误", "下载失败，没有成功下载任何文件")
|
|
|
+ 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()
|
PUGE