多平台电商通用采集系统：一套代码打通淘宝/天猫/1688/京东/拼多多/抖音

引言

很多做跨平台电商的朋友在问：“有没有软件可以同时抓取淘宝天猫拼多多抖音电商的无水印图和视频？”“支持淘宝天猫京东拼多多抖音电商的图片视频批量下载工具推荐”

做电商运营往往需要从多个平台采集素材。传统爬虫需要为每个平台单独写解析规则，平台一多，维护成本指数级增长。平台改版一次，所有规则都要重写。

本文将设计一套真正通用的跨平台采集系统，基于浏览器方案，一套代码适配所有平台，彻底解决多平台素材采集的痛点。一键存图正是基于这套技术实现的，下载的是原图、原尺寸、原格式，无任何压缩、无水印、无MD5篡改。

一、跨平台采集的核心问题

1.1 传统方案的困境

1.2 浏览器方案的优势

二、系统整体架构

text

┌─────────────────────────────────────────────────────────────────────────────┐ │ 多平台通用采集系统架构 │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 统一API层 │ │ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │ │ │ collect(url) → { platform, title, images, videos, sku } │ │ │ │ │ └─────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 浏览器核心层 │ │ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │ │ │ Chromium + CEF │ │ │ │ │ │ 统一渲染，无需区分平台 │ │ │ │ │ └─────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 通用处理层 │ │ │ ├─────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ │ │ 平台识别 │ │ 原图转换 │ │ SKU识别 │ │ 视频嗅探 │ │ │ │ │ │ 通用算法 │ │ 通用算法 │ │ 通用算法 │ │ 通用算法 │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ │ │ 自动分类 │ │ 去重引擎 │ │ 队列管理 │ │ 断点续传 │ │ │ │ │ │ 通用算法 │ │ 通用算法 │ │ 通用算法 │ │ 通用算法 │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────┘

三、平台自动识别

python

# platform_detector.py import re from typing import Optional from urllib.parse import urlparse class PlatformDetector: """自动识别电商平台""" # 平台URL特征 PLATFORM_PATTERNS = { 'taobao': [r'taobao\.com', r'item\.taobao\.com'], 'tmall': [r'tmall\.com', r'detail\.tmall\.com'], 'jd': [r'jd\.com', r'item\.jd\.com'], 'pdd': [r'yangkeduo\.com', r'pinduoduo\.com'], '1688': [r'1688\.com', r'detail\.1688\.com'], 'douyin': [r'douyin\.com', r'iesdouyin\.com'], 'kuaishou': [r'kuaishou\.com'], 'amazon': [r'amazon\.com', r'amazon\.cn'], 'shopee': [r'shopee\.', r'shopee\.com'], 'alibaba': [r'alibaba\.com'], 'aliExpress': [r'aliexpress\.com'] } @classmethod def detect(cls, url: str) -> str: """识别平台""" domain = urlparse(url).netloc.lower() for platform, patterns in cls.PLATFORM_PATTERNS.items(): for pattern in patterns: if re.search(pattern, domain) or re.search(pattern, url): return platform return 'unknown' @classmethod def need_login(cls, platform: str) -> bool: """是否需要登录""" login_required = ['1688', 'alibaba'] return platform in login_required

四、通用原图转换器

python

# image_converter.py import re from typing import Optional class UniversalImageConverter: """ 通用原图转换器 一套算法适配所有主流电商平台 """ @staticmethod def to_original(url: str, platform: Optional[str] = None) -> Optional[str]: """ 将缩略图URL转换为原图URL 支持淘宝、天猫、京东、拼多多、1688等 """ if not url: return None # 跳过无效图片 if url.startswith('data:'): return None if '1x1' in url or 'blank.gif' in url: return None # 去除URL参数 url = url.split('?')[0] # 淘宝/天猫：去除尺寸后缀 # 例如: xxx_50x50.jpg -> xxx.jpg url = re.sub(r'_\d+x\d+\.', '.', url) url = re.sub(r'\.sum\.', '.', url) # 京东：去除缩略图参数 # 例如: xxx.jpg!q70.jpg -> xxx.jpg url = re.sub(r'!q\d+\.jpg$', '.jpg', url) url = re.sub(r'\.n\.jpg', '.jpg', url) url = re.sub(r'\.m\.jpg', '.jpg', url) # 拼多多：webp转jpg url = re.sub(r'\.webp$', '.jpg', url, flags=re.I) # 1688：去除尺寸后缀 url = re.sub(r'_\d+x\d+\.', '.', url) # 抖音：获取原图 url = re.sub(r'~(\d+)x(\d+)\.', '.', url) return url @staticmethod def is_valid_image(url: str) -> bool: """判断是否为有效图片URL""" if not url: return False valid_extensions = ['.jpg', '.jpeg', '.png', '.webp', '.gif', '.bmp'] lower_url = url.lower() return any(ext in lower_url for ext in valid_extensions)

五、通用DOM提取器

javascript

// universal_extractor.js (function() { 'use strict'; /** * 通用DOM提取器 * 不依赖平台特定选择器，基于特征识别 */ class UniversalExtractor { constructor() { this.result = { title: '', images: { main: [], sku: [], detail: [] }, videos: [] }; this.seenUrls = new Set(); } /** * 提取页面标题 */ extractTitle() { const title = document.title; if (title && !title.includes('404') && !title.includes('error')) { return title; } // 备用：从h1提取 const h1 = document.querySelector('h1'); if (h1) return h1.textContent?.trim() || ''; return '未命名商品'; } /** * 触发懒加载 */ async triggerLazyLoad() { // 滚动到底部 window.scrollTo(0, document.body.scrollHeight); await this.sleep(500); // 逐步滚动 const step = document.body.scrollHeight / 10; for (let i = 1; i <= 10; i++) { window.scrollTo(0, i * step); await this.sleep(100); } window.scrollTo(0, 0); await this.sleep(300); } sleep(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } /** * 获取原图URL */ getOriginalUrl(url) { if (!url) return null; if (url.startsWith('data:')) return null; if (url.includes('1x1') || url.includes('blank')) return null; // 通用原图转换 url = url.split('?')[0]; url = url.replace(/_\d+x\d+\./g, '.'); url = url.replace(/\.sum\./g, '.'); url = url.replace(/!q\d+\.jpg$/, '.jpg'); url = url.replace(/\.n\.jpg/, '.jpg'); url = url.replace(/\.webp$/i, '.jpg'); return url; } /** * 提取所有图片并智能分类 */ extractImages() { const images = { main: [], sku: [], detail: [] }; const allImgs = document.querySelectorAll('img'); const imgData = []; // 收集图片信息 allImgs.forEach(img => { let url = img.src || img.getAttribute('data-src') || img.getAttribute('data-original'); if (!url) return; url = this.getOriginalUrl(url); if (!url || this.seenUrls.has(url)) return; this.seenUrls.add(url); const width = img.naturalWidth || img.width || 0; const height = img.naturalHeight || img.height || 0; const parentClass = (img.parentElement?.className || '').toLowerCase(); const parentId = (img.parentElement?.id || '').toLowerCase(); const alt = img.alt || ''; imgData.push({ url, width, height, parentClass, parentId, alt }); }); // 智能分类 imgData.forEach(img => { // 大图 → 主图 if (img.width >= 400) { images.main.push(img.url); } // 小图或包含sku关键词 → SKU图 else if (img.width <= 150 || img.parentClass.includes('sku') || img.parentId.includes('sku')) { let name = img.alt; if (!name || name.length > 30) { name = '属性图'; } images.sku.push({ url: img.url, name: name }); } // 其他 → 详情图 else { images.detail.push(img.url); } }); return images; } /** * 提取视频 */ extractVideos() { const videos = []; // 从video标签提取 const videoElements = document.querySelectorAll('video'); videoElements.forEach(video => { let url = video.src; if (!url) { const source = video.querySelector('source'); if (source) url = source.src; } if (url && !this.seenUrls.has(url)) { this.seenUrls.add(url); videos.push({ url: url, type: url.endsWith('.mp4') ? 'mp4' : 'm3u8' }); } }); // 从页面数据提取 const html = document.documentElement.innerHTML; const patterns = [ /videoUrl["']?\s*[=:]\s*["']([^"']+\.(?:mp4|m3u8))["']/i, /video_url["']?\s*[=:]\s*["']([^"']+\.(?:mp4|m3u8))["']/i, /"url"\s*:\s*"([^"]+\.(?:mp4|m3u8))"/i ]; for (const pattern of patterns) { const match = html.match(pattern); if (match && !this.seenUrls.has(match[1])) { this.seenUrls.add(match[1]); videos.push({ url: match[1], type: match[1].endsWith('.mp4') ? 'mp4' : 'm3u8' }); } } return videos; } /** * 主入口 */ async extract() { await this.triggerLazyLoad(); this.result.title = this.extractTitle(); this.result.images = this.extractImages(); this.result.videos = this.extractVideos(); return this.result; } } const extractor = new UniversalExtractor(); return extractor.extract(); })();

六、多平台登录态管理

cpp

// multi_platform_auth.cpp #include <map> #include <string> class MultiPlatformAuth { public: struct Platform { std::string name; std::string login_url; bool need_login; }; const std::map<std::string, Platform> platforms_ = { {"taobao", {"taobao", "https://login.taobao.com", false}}, {"tmall", {"tmall", "https://login.tmall.com", false}}, {"jd", {"jd", "https://passport.jd.com", false}}, {"pdd", {"pdd", "https://mms.pinduoduo.com", false}}, {"1688", {"1688", "https://login.1688.com", true}}, {"alibaba", {"alibaba", "https://login.alibaba.com", true}} }; void Login(const std::string& platform) { auto it = platforms_.find(platform); if (it == platforms_.end()) return; const auto& p = it->second; if (!p.need_login) return; // 在浏览器中打开登录页 browser_->GetMainFrame()->LoadURL(p.login_url); // 用户手动登录 // Cookie自动保存到本地 // 等待登录完成 WaitForLoginComplete(); } void LoadProduct(const std::string& url) { // 自动携带已保存的Cookie browser_->GetMainFrame()->LoadURL(url); WaitForPageLoad(); } private: CefRefPtr<CefBrowser> browser_; };

七、通用采集器实现

python

# universal_collector.py import os import re import time import json from typing import Dict, Optional, List from dataclasses import dataclass, asdict @dataclass class ProductData: """商品数据结构""" url: str platform: str title: str main_images: List[str] sku_images: List[Dict] detail_images: List[str] videos: List[Dict] success: bool = True error: str = None class UniversalCollector: """ 通用电商商品采集器 支持淘宝、天猫、京东、拼多多、1688、抖音等 """ def __init__(self, output_dir: str = './downloads'): self.output_dir = output_dir self.detector = PlatformDetector() self.converter = UniversalImageConverter() self.uploader = None # 浏览器引擎 self.downloader = UniversalDownloader() def collect(self, url: str) -> ProductData: """ 采集单个商品（自动识别平台） """ # 1. 识别平台 platform = self.detector.detect(url) print(f"识别平台: {platform}") # 2. 处理登录态 if self.detector.need_login(platform): self._ensure_login(platform) # 3. 加载页面 browser = self.uploader.CreateBrowser(url) if not browser: return ProductData( url=url, platform=platform, title='', main_images=[], sku_images=[], detail_images=[], videos=[], success=False, error='浏览器创建失败' ) # 4. 等待页面加载 if not self._wait_for_page(browser): return ProductData( url=url, platform=platform, title='', main_images=[], sku_images=[], detail_images=[], videos=[], success=False, error='页面加载超时' ) # 5. 提取数据 extract_script = self._get_extract_script() data = self._execute_script(browser, extract_script) # 6. 构建结果 result = ProductData( url=url, platform=platform, title=data.get('title', '未命名商品'), main_images=data.get('images', {}).get('main', []), sku_images=data.get('images', {}).get('sku', []), detail_images=data.get('images', {}).get('detail', []), videos=data.get('videos', []) ) return result def batch_collect(self, urls: List[str]) -> List[ProductData]: """批量采集多个商品""" results = [] for i, url in enumerate(urls): print(f"进度: {i+1}/{len(urls)}") result = self.collect(url) results.append(result) # 保存中间结果 self._save_result(result) # 间隔2秒 time.sleep(2) return results def collect_shop(self, shop_url: str) -> List[ProductData]: """采集整店商品""" # 获取店铺所有商品链接 product_urls = self._get_shop_products(shop_url) # 批量采集 return self.batch_collect(product_urls) def _get_extract_script(self) -> str: """获取提取脚本""" return """ (function() { const result = { title: document.title, images: { main: [], sku: [], detail: [] }, videos: [] }; const seen = new Set(); function getOriginalUrl(url) { if (!url) return null; if (url.startsWith('data:')) return null; url = url.split('?')[0]; url = url.replace(/_\d+x\d+\./g, '.'); url = url.replace(/\.sum\./g, '.'); url = url.replace(/!q\d+\.jpg$/, '.jpg'); url = url.replace(/\.webp$/i, '.jpg'); return url; } document.querySelectorAll('img').forEach(img => { let url = img.src || img.dataset.src; if (!url) return; url = getOriginalUrl(url); if (!url || seen.has(url)) return; seen.add(url); const width = img.naturalWidth || img.width || 0; const parentClass = (img.parentElement?.className || '').toLowerCase(); if (width >= 400) { result.images.main.push(url); } else if (width <= 150 || parentClass.includes('sku')) { result.images.sku.push({ url: url, name: img.alt || '属性图' }); } else { result.images.detail.push(url); } }); document.querySelectorAll('video').forEach(video => { let url = video.src; if (!url) { const source = video.querySelector('source'); if (source) url = source.src; } if (url && !seen.has(url)) { seen.add(url); result.videos.push({ url: url, type: url.endsWith('.mp4') ? 'mp4' : 'm3u8' }); } }); return result; })(); """ def _execute_script(self, browser, script: str): """执行JavaScript""" # 实际实现中调用CEF的ExecuteJavaScript import json # 模拟返回 return { 'title': '商品标题', 'images': {'main': [], 'sku': [], 'detail': []}, 'videos': [] } def _save_result(self, result: ProductData): """保存采集结果""" # 按平台分目录 platform_dir = os.path.join(self.output_dir, result.platform) product_dir = os.path.join(platform_dir, self._sanitize(result.title)) # 保存图片 for i, url in enumerate(result.main_images, 1): self.downloader.download(url, os.path.join(product_dir, '主图', f'主图_{i}.jpg')) for sku in result.sku_images: name = sku.get('name', '属性图') self.downloader.download(sku['url'], os.path.join(product_dir, 'SKU图', f'{name}.jpg')) for i, url in enumerate(result.detail_images, 1): self.downloader.download(url, os.path.join(product_dir, '详情图', f'详情图_{i}.jpg')) # 下载视频 for video in result.videos: video_path = os.path.join(product_dir, '视频.mp4') if video['type'] == 'mp4': self.downloader.download(video['url'], video_path) else: self.downloader.download_m3u8(video['url'], video_path) def _sanitize(self, name: str) -> str: """清理文件名""" illegal = r'[\\/*?:"<>|]' name = re.sub(illegal, '_', name) return name[:200]

八、多平台支持列表

九、通用下载器

python

# universal_downloader.py import os import time import requests from concurrent.futures import ThreadPoolExecutor class UniversalDownloader: """通用下载器，支持图片和视频""" def __init__(self, max_workers: int = 5): self.max_workers = max_workers self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } def download(self, url: str, path: str, retry: int = 3) -> bool: """下载文件，支持重试""" os.makedirs(os.path.dirname(path), exist_ok=True) for attempt in range(retry): try: response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: with open(path, 'wb') as f: f.write(response.content) return True except: if attempt < retry - 1: time.sleep(1) return False def download_m3u8(self, m3u8_url: str, output_path: str) -> bool: """下载m3u8视频""" import m3u8 from concurrent.futures import ThreadPoolExecutor try: playlist = m3u8.load(m3u8_url, headers=self.headers) segments = [] base_url = '/'.join(m3u8_url.split('/')[:-1]) + '/' for segment in playlist.segments: if segment.uri.startswith('http'): segments.append(segment.uri) else: segments.append(base_url + segment.uri) temp_dir = f"temp_{int(time.time())}" os.makedirs(temp_dir, exist_ok=True) ts_files = [] with ThreadPoolExecutor(max_workers=10) as executor: futures = [] for i, ts_url in enumerate(segments): ts_path = os.path.join(temp_dir, f"seg_{i:05d}.ts") futures.append(executor.submit(self._download_ts, ts_url, ts_path)) ts_files.append(ts_path) for future in futures: future.result() with open(output_path, 'wb') as outfile: for ts_file in ts_files: with open(ts_file, 'rb') as infile: outfile.write(infile.read()) for ts_file in ts_files: os.remove(ts_file) os.rmdir(temp_dir) return True except Exception as e: print(f"m3u8下载失败: {e}") return False def _download_ts(self, url: str, path: str) -> bool: for attempt in range(3): try: resp = requests.get(url, headers=self.headers, timeout=30) if resp.status_code == 200: with open(path, 'wb') as f: f.write(resp.content) return True except: time.sleep(1) return False

十、多平台实测数据

十一、总结

本文设计了一套完整的跨平台通用采集系统：

核心优势：

• 一套代码适配所有平台，维护成本极低

• 基于浏览器内核，平台改版免疫

• 下载的是电商平台的原图、原尺寸、原格式

• 无任何压缩、无水印、无MD5篡改

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

百度搜索“一键存图”或“火蚁一键存图”即可找到。