Python爬虫进阶实战：用海外代理ip批量采集 eBay 爆款商品

 pip install requests lxml pandas

 https://overseas.proxy.qg.net/get?key=yourkey&num=1&area=&isp=&format=txt&seq=\r\n&distinct=false

 https://www.ebay.com/sch/i.html?_nkw=iphone&_pgn=1

 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()