亚马逊商品图片批量采集系统：多变体SKU图提取与自动分类

引言

很多做跨境的卖家在问：“支持亚马逊商品图片下载的工具”“亚马逊商品主图和变体图批量保存”

亚马逊是跨境卖家最主要的平台。商品包含主图、多角度附图、变体图（颜色/尺寸），手动保存效率极低，且图片质量参差不齐。

本文将完整实现一套亚马逊商品图片批量采集系统，涵盖商品列表获取、主图提取、变体SKU图识别、详情图下载、自动分类等核心功能。一键存图正是基于这套技术实现的，下载的是原图、原尺寸、原格式，无任何压缩、无水印、无MD5篡改。

一、亚马逊平台技术特点分析

1.1 核心难点

1.2 亚马逊图片URL格式

python

# 亚马逊图片URL示例 # 原图格式 https://images-na.ssl-images-amazon.com/images/I/71xxx.jpg # 缩放版本（需要转换） https://images-na.ssl-images-amazon.com/images/I/71xxx._AC_SL1500_.jpg # 原图规则：去除尺寸标识 # ._AC_SL1500_.jpg -> .jpg

1.3 变体SKU图片

亚马逊商品的变体（颜色、尺寸）通常有独立的SKU图片，需要关联属性名称：

html

<!-- 变体容器示例 --> <div class="variation-selector"> <div data-value="红色" data-sku="SKU001"> <img src="red_thumb.jpg"> </div> <div data-value="蓝色" data-sku="SKU002"> <img src="blue_thumb.jpg"> </div> </div>

二、系统整体架构

text

┌─────────────────────────────────────────────────────────────────────────────┐ │ 亚马逊商品图片批量采集系统架构 │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 控制层 │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ │ │ 商品解析 │ │ 变体识别 │ │ 队列管理 │ │ 断点续传 │ │ │ │ │ │ 引擎 │ │ 引擎 │ │ 器 │ │ 器 │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 采集层 │ │ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │ │ │ Chromium浏览器内核 │ │ │ │ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ │ │ │ │ 页面加载 │ │ JS执行 │ │ 懒加载 │ │ Cookie │ │ │ │ │ │ │ │ 控制器 │ │ 引擎 │ │ 触发器 │ │ 管理器 │ │ │ │ │ │ │ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ │ └─────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 处理层 │ │ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ │ │ 主图提取 │ │ 变体图 │ │ 附图提取 │ │ 原图转换 │ │ │ │ │ │ 引擎 │ │ 提取引擎 │ │ 引擎 │ │ 引擎 │ │ │ │ │ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 存储层 │ │ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ │ │ 主图目录 │ │ 变体目录 │ │ 附图目录 │ │ 详情目录 │ │ │ │ │ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────┘

三、亚马逊商品解析引擎

javascript

// amazon_product_extractor.js (function() { 'use strict'; /** * 亚马逊商品解析器 * 支持主图、变体图、附图、详情图提取 */ class AmazonProductExtractor { constructor() { this.result = { title: '', mainImages: [], variantImages: [], // 变体图片（颜色/尺寸） altImages: [], // 多角度附图 detailImages: [] // 详情描述图 }; this.seenUrls = new Set(); } /** * 等待页面完全加载 */ async waitForPageReady() { while (document.readyState !== 'complete') { await this.sleep(200); } // 等待图片容器加载 await this.waitForImageContainer(); await this.sleep(1000); } async waitForImageContainer() { let maxWait = 30; while (maxWait-- > 0) { const imgBlock = document.querySelector('#imgTagWrapperId, .imgTagWrapper, .image-block'); if (imgBlock && imgBlock.querySelector('img')) { return; } await this.sleep(500); } } sleep(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } /** * 触发懒加载 */ async triggerLazyLoad() { window.scrollTo(0, document.body.scrollHeight); await this.sleep(500); const step = document.body.scrollHeight / 5; for (let i = 1; i <= 5; i++) { window.scrollTo(0, i * step); await this.sleep(300); } window.scrollTo(0, 0); await this.sleep(300); } /** * 获取亚马逊原图URL */ getOriginalUrl(url) { if (!url) return null; if (url.startsWith('data:')) return null; if (url.includes('1x1') || url.includes('blank')) return null; // 去除亚马逊尺寸参数 // ._AC_SL1500_.jpg -> .jpg // ._SX679_.jpg -> .jpg url = url.replace(/\._[A-Z]+_\d+_\./g, '.'); url = url.replace(/\._[A-Z]+_\d+\./g, '.'); url = url.split('?')[0]; return url; } /** * 提取标题 */ extractTitle() { const selectors = [ '#productTitle', '.a-size-large', '.product-title-word-break', 'h1' ]; for (const selector of selectors) { const el = document.querySelector(selector); if (el && el.textContent) { let title = el.textContent.trim(); if (title.length > 10) return title; } } return document.title || '亚马逊商品'; } /** * 提取主图 */ extractMainImages() { const images = []; // 方法1: 从主图容器提取 const mainImageContainer = document.querySelector('#imgTagWrapperId, .imgTagWrapper, #main-image-container'); if (mainImageContainer) { const mainImg = mainImageContainer.querySelector('img'); if (mainImg) { let url = mainImg.src || mainImg.getAttribute('data-old-hires'); if (url) { const originalUrl = this.getOriginalUrl(url); if (originalUrl && !this.seenUrls.has(originalUrl)) { this.seenUrls.add(originalUrl); images.push(originalUrl); } } } } // 方法2: 从缩略图列表提取 const thumbSelectors = [ '#altImages img', '.a-carousel img', '.image-thumbnail img', '[class*="thumbnail"] img' ]; for (const selector of thumbSelectors) { const thumbs = document.querySelectorAll(selector); for (const thumb of thumbs) { let url = thumb.src || thumb.getAttribute('data-src'); if (url) { // 替换缩略图URL为主图URL url = url.replace(/\._US\d+_\./g, '.'); url = url.replace(/\._AC_\d+_\./g, '.'); const originalUrl = this.getOriginalUrl(url); if (originalUrl && !this.seenUrls.has(originalUrl)) { this.seenUrls.add(originalUrl); images.push(originalUrl); } } } if (images.length > 0) break; } return images; } /** * 提取变体图片（颜色/尺寸SKU图） */ extractVariantImages() { const variants = []; // 变体选择器 const variantSelectors = [ '#variation_color_name', '#variation_size_name', '.variation-selector', '[data-csa-c-item-type="variation"]', '.a-section .a-row .a-dropdown-container' ]; let variantContainer = null; for (const selector of variantSelectors) { variantContainer = document.querySelector(selector); if (variantContainer) break; } if (variantContainer) { // 提取所有变体选项 const options = variantContainer.querySelectorAll('.a-button, .swatchElement, [data-value]'); options.forEach(option => { // 提取变体名称 let name = ''; const nameSelectors = [ '.a-button-text', '.swatchTitle', '[data-csa-c-content-id]', 'span' ]; for (const selector of nameSelectors) { const nameEl = option.querySelector(selector); if (nameEl) { name = nameEl.textContent?.trim(); if (name) break; } } if (!name) { name = option.getAttribute('data-value') || option.getAttribute('title') || '变体'; } // 提取变体图片 const img = option.querySelector('img'); if (img) { let url = img.src || img.getAttribute('data-src'); if (url) { const originalUrl = this.getOriginalUrl(url); if (originalUrl && !this.seenUrls.has(originalUrl)) { this.seenUrls.add(originalUrl); variants.push({ url: originalUrl, name: name }); } } } }); } return variants; } /** * 提取多角度附图 */ extractAltImages() { const images = []; const altSelectors = [ '#altImages img', '.a-carousel img', '.image-thumbnail img', '[class*="alt-image"] img', '[class*="thumbnail"] img' ]; for (const selector of altSelectors) { const imgs = document.querySelectorAll(selector); for (const img of imgs) { let url = img.src || img.getAttribute('data-src'); if (url) { const originalUrl = this.getOriginalUrl(url); if (originalUrl && !this.seenUrls.has(originalUrl)) { this.seenUrls.add(originalUrl); images.push(originalUrl); } } } if (images.length > 3) break; } return images; } /** * 提取详情描述图 */ extractDetailImages() { const images = []; const detailSelectors = [ '#productDescription img', '.aplus-v2 img', '.aplus img', '.product-description img', '[class*="description"] img' ]; let detailContainer = null; for (const selector of detailSelectors) { detailContainer = document.querySelector(selector); if (detailContainer) break; } if (detailContainer) { const imgs = detailContainer.querySelectorAll('img'); for (const img of imgs) { let url = img.src || img.getAttribute('data-src'); if (url) { const originalUrl = this.getOriginalUrl(url); if (originalUrl && !this.seenUrls.has(originalUrl)) { this.seenUrls.add(originalUrl); images.push(originalUrl); } } } } return images; } /** * 主入口 */ async extract() { await this.waitForPageReady(); await this.triggerLazyLoad(); this.result.title = this.extractTitle(); this.result.mainImages = this.extractMainImages(); this.result.variantImages = this.extractVariantImages(); this.result.altImages = this.extractAltImages(); this.result.detailImages = this.extractDetailImages(); return this.result; } } const extractor = new AmazonProductExtractor(); return extractor.extract(); })();

四、批量采集调度器

python

# amazon_batch_collector.py import os import json import time import threading from queue import Queue from typing import List, Dict, Optional from dataclasses import dataclass, asdict from datetime import datetime @dataclass class AmazonProductData: """亚马逊商品数据结构""" url: str asin: str title: str main_images: List[str] variant_images: List[Dict] alt_images: List[str] detail_images: List[str] success: bool = True error: str = None timestamp: str = None def __post_init__(self): if not self.timestamp: self.timestamp = datetime.now().isoformat() class AmazonBatchCollector: """亚马逊批量采集调度器""" def __init__(self, output_dir: str = './downloads/amazon', max_concurrent: int = 1): self.output_dir = output_dir self.max_concurrent = max_concurrent self.queue = Queue() self.results = [] self.lock = threading.Lock() self.completed_asins = set() self.state_file = "amazon_batch_state.json" self._load_state() def _load_state(self): if os.path.exists(self.state_file): try: with open(self.state_file, 'r') as f: data = json.load(f) self.completed_asins = set(data.get('completed_asins', [])) print(f"📁 加载断点: 已完成{len(self.completed_asins)}个商品") except Exception as e: print(f"加载状态失败: {e}") def _save_state(self): with self.lock: with open(self.state_file, 'w') as f: json.dump({ 'completed_asins': list(self.completed_asins), 'last_update': datetime.now().isoformat() }, f, indent=2) def add_products(self, products: List[Dict]): """添加商品到队列""" for product in products: asin = product.get('asin') url = product.get('url') if asin and asin not in self.completed_asins: self.queue.put({'asin': asin, 'url': url}) print(f"📋 队列中有{self.queue.qsize()}个待处理商品") def collect_all(self, collector_func) -> List[AmazonProductData]: print(f"🚀 开始批量采集，并发数: {self.max_concurrent}") threads = [] for _ in range(self.max_concurrent): t = threading.Thread(target=self._worker, args=(collector_func,)) t.start() threads.append(t) for t in threads: t.join() self._save_results() return self.results def _worker(self, collector_func): while not self.queue.empty(): try: task = self.queue.get(timeout=1) print(f"📦 采集商品: {task['asin']}") result = collector_func(task['url']) result.asin = task['asin'] with self.lock: self.results.append(result) if result.success: self.completed_asins.add(task['asin']) self._save_state() success_count = sum(1 for r in self.results if r.success) print(f"📊 进度: {len(self.results)} 个商品, 成功: {success_count}") time.sleep(3) except Exception as e: print(f"❌ 异常: {e}") def _save_results(self): results_file = f"amazon_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(results_file, 'w', encoding='utf-8') as f: json.dump([asdict(r) for r in self.results], f, ensure_ascii=False, indent=2) print(f"📁 结果保存: {results_file}") def generate_report(self) -> Dict: if not self.results: return {} total = len(self.results) success = sum(1 for r in self.results if r.success) total_main = sum(len(r.main_images) for r in self.results) total_variant = sum(len(r.variant_images) for r in self.results) total_alt = sum(len(r.alt_images) for r in self.results) total_detail = sum(len(r.detail_images) for r in self.results) return { 'total_products': total, 'success': success, 'failed': total - success, 'success_rate': f"{success/total*100:.1f}%", 'total_main_images': total_main, 'total_variant_images': total_variant, 'total_alt_images': total_alt, 'total_detail_images': total_detail }

五、文件存储管理器

python

# amazon_storage.py import os import re import json import requests from typing import Dict class AmazonStorage: """亚马逊商品文件存储管理器""" def __init__(self, base_dir: str = './downloads/amazon'): self.base_dir = base_dir os.makedirs(self.base_dir, exist_ok=True) def _sanitize(self, name: str) -> str: illegal = r'[\\/*?:"<>|]' name = re.sub(illegal, '_', name) return name[:200].strip() def save_product(self, product_data: AmazonProductData) -> Dict: safe_title = self._sanitize(product_data.title) product_dir = os.path.join(self.base_dir, f"{product_data.asin}_{safe_title}") subdirs = ['主图', '变体图', '附图', '详情图'] for subdir in subdirs: os.makedirs(os.path.join(product_dir, subdir), exist_ok=True) stats = {'main': 0, 'variant': 0, 'alt': 0, 'detail': 0} # 主图 for i, url in enumerate(product_data.main_images, 1): path = os.path.join(product_dir, '主图', f'主图_{i}.jpg') if self._download(url, path): stats['main'] += 1 # 变体图 for variant in product_data.variant_images: name = self._sanitize(variant.get('name', '变体')) path = os.path.join(product_dir, '变体图', f'{name}.jpg') if self._download(variant['url'], path): stats['variant'] += 1 # 附图 for i, url in enumerate(product_data.alt_images, 1): path = os.path.join(product_dir, '附图', f'附图_{i}.jpg') if self._download(url, path): stats['alt'] += 1 # 详情图 for i, url in enumerate(product_data.detail_images, 1): path = os.path.join(product_dir, '详情图', f'详情图_{i}.jpg') if self._download(url, path): stats['detail'] += 1 # 保存信息 info_path = os.path.join(product_dir, '商品信息.json') with open(info_path, 'w', encoding='utf-8') as f: json.dump({ 'asin': product_data.asin, 'title': product_data.title, 'url': product_data.url, 'stats': stats }, f, ensure_ascii=False, indent=2) return stats def _download(self, url: str, path: str, retry: int = 3) -> bool: for attempt in range(retry): try: headers = {'User-Agent': 'Mozilla/5.0'} resp = requests.get(url, headers=headers, timeout=30) if resp.status_code == 200: with open(path, 'wb') as f: f.write(resp.content) return True except: if attempt < retry - 1: time.sleep(1) return False

六、保存目录结构

text

downloads/amazon/ ├── B0XXXXXX1_2024秋冬新款女装/ │ ├── 主图/ │ │ └── 主图_1.jpg │ ├── 变体图/ │ │ ├── 红色.jpg │ │ ├── 蓝色.jpg │ │ └── 黑色.jpg │ ├── 附图/ │ │ ├── 附图_1.jpg │ │ ├── 附图_2.jpg │ │ └── 附图_3.jpg │ ├── 详情图/ │ │ ├── 详情图_1.jpg │ │ └── 详情图_2.jpg │ └── 商品信息.json └── B0XXXXXX2_2025新款运动鞋/ └── ...

七、实测数据

八、总结

核心要点：

• 基于Chromium浏览器内核，下载的是亚马逊的原图、原尺寸、原格式

• 变体图自动按颜色/尺寸分类命名

• 支持断点续传

结论：如果你需要一款稳定、自动分类、支持全平台的电商图片下载工具，一键存图是目前最省心的选择。