一、前言
如果说跨境电商是一场没有硝烟的战争,那么数据就是最锋利的武器。 在 eBay 这样的国际电商平台上,热销商品的背后,往往意味着:
- 潮流趋势的风向标 🎯 —— 哪些品类在快速崛起?
- 利润空间的显微镜 💰 —— 市场价格区间在哪里,怎么避免低价内卷?
- 选品决策的导航仪 📊 —— 该备哪些货、投多少广告,才不至于踩坑?
可现实是,很多朋友刚写好一个简单爬虫,没跑几页就遭遇 403 禁止访问 或者被弹出“你是机器人吗?”的拦截提示。这背后的核心问题就是:IP 和访问模式太单一,很容易被目标网站识别。
具体如何实施呢?很多小白依然瞎子抓虾一头雾水。别着急,今天我们就从 0 到 1 带你一起来最终实现:自动批量采集某关键词下的 eBay 热销商品,并保存到 Excel 中。
二、目标与思路
2.1 采集目标
- 商品标题
- 商品价格
- 成交量
- 商品链接
- 商品图片
2.2 实现思路
- 分析搜索页面 URL 参数,找出关键词与翻页逻辑。
- 用 Python + requests 模拟请求,获取页面 HTML。
- 用 XPath 定位商品卡片结构,提取需要的字段。
- 接入海外代理ip,实现 海外节点访问。
- 增加项目成功率的一些小方法:代理池轮换、请求重试、随机 UA、延时限速。
- 结果写入 Excel,方便后续做选品分析。
三、环境准备
- Python:3.10
- 编辑器:PyCharm / VS Code
- 第三方库:
pip install requests lxml pandas
四、接入海外代理IP
在数据采集场景中,海外代理ip的作用可以说是非常必要的。
- 避免同一 IP 短时间请求过多,触发封禁。
- 根据目标市场选择合适的地区节点(例如美国、英国、德国)。
- 提升请求的成功率和稳定性。
获取API 提取链接,例如:
https://overseas.proxy.qg.net/get?key=yourkey&num=1&area=&isp=&format=txt&seq=\r\n&distinct=false
五、eBay 页面分析
在 eBay 搜索“iphone”,URL 如下:
https://www.ebay.com/sch/i.html?_nkw=iphone&_pgn=1
通过对比翻页:
_nkw→ 关键词_pgn→ 页码
这意味着只要改 _pgn=2,3,4…,就能翻页抓取。
六、实战采集
废话不多说,直接上:
import os, time, random, logging, requests, pandas as pd from lxml import etree from typing import List, Dict, Optional # ========================= # 可配置区 # ========================= KEYWORD = "iphone" SITE = "www.ebay.com" PAGES = 5 PROXY_API = "https://overseas.proxy.qg.net/get?key=yourkey&num=1&area=&isp=&format=txt&seq=\r\n&distinct=false"#使用青果网络海外代理IP API OUTPUT_XLSX = f"{KEYWORD}_ebay_hot.xlsx" # ========================= # 日志 # ========================= logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") logger = logging.getLogger("ebay-hot") # ========================= # 工具:UA 池 # ========================= UA_POOL = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_5) AppleWebKit/605.1.15 Version/16.5 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/123.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0", ] def rand_headers(): return {"User-Agent": random.choice(UA_POOL)} # ========================= # 代理池 # ========================= def get_proxy() -> Optional[Dict[str, str]]: try: r = requests.get(PROXY_API, timeout=8) ip = r.text.strip().splitlines()[0].strip() return {"http": f"http://{ip}", "https": f"http://{ip}"} except Exception as e: logger.warning(f"获取代理失败:{e}") return None # ========================= # 请求封装(重试+回退) # ========================= def fetch_html(url: str, retries=4) -> Optional[str]: for attempt in range(1, retries+1): proxies = get_proxy() headers = rand_headers() try: logger.info(f"请求 {url} | 尝试 {attempt}") r = requests.get(url, headers=headers, proxies=proxies, timeout=12) r.raise_for_status() return r.text except Exception as e: backoff = 2 ** attempt + random.random() logger.warning(f"失败:{e} | {backoff:.1f}s 后重试") time.sleep(backoff) return None # ========================= # 解析函数 # ========================= def parse(html: str, page: int) -> List[Dict]: tree = etree.HTML(html) items = tree.xpath('//li[contains(@class,"s-item")]') results = [] for it in items: title = _first(it.xpath('.//h3/text()')) price = _first(it.xpath('.//span[@class="s-item__price"]/text()')) sold = _first(it.xpath('.//span[contains(text(),"sold")]/text()')) link = _first(it.xpath('.//a[@class="s-item__link"]/@href')) img = _first(it.xpath('.//img[contains(@class,"s-item__image-img")]/@src')) if title and link: results.append({ "页码": page, "标题": title, "价格": price, "销量": sold, "链接": link, "图片": img }) return results def _first(lst): return lst[0].strip() if lst else None # ========================= # 主流程 # ========================= def run(): all_data = [] for p in range(1, PAGES+1): url = f"https://{SITE}/sch/i.html?_nkw={KEYWORD}&_pgn={p}" html = fetch_html(url) if html: rows = parse(html, p) logger.info(f"第 {p} 页采集 {len(rows)} 条") all_data.extend(rows) time.sleep(random.uniform(2, 5)) if all_data: df = pd.DataFrame(all_data) df.drop_duplicates(subset=["链接"], inplace=True) df.to_excel(OUTPUT_XLSX, index=False) logger.info(f"导出完成:{os.path.abspath(OUTPUT_XLSX)}") if __name__ == "__main__": run()