中间件实现任务去重与精细化分发：设计模式与常见陷阱

import hashlib
import json
import random
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote

# ========= 代理配置（亿牛云示例） =========
PROXY_HOST = "proxy.16yun.com"
PROXY_PORT = "3100"
PROXY_USER = "16YUN"
PROXY_PASS = "16IP"

def get_proxy():
    """生成带认证的代理配置"""
    return {
        "http": f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}",
        "https": f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
    }

# ========= 去重中间件 =========
class DeduplicationMiddleware:
    """任务去重逻辑"""
    def __init__(self):
        self.visited = set()

    def is_duplicate(self, url):
        key = hashlib.md5(url.encode('utf-8')).hexdigest()
        if key in self.visited:
            return True
        self.visited.add(key)
        return False

# ========= 分发中间件 =========
class DistributionMiddleware:
    """根据任务特征选择代理"""
    def select_proxy(self, keyword):
        # 简单策略：根据关键词长度动态分配
        return get_proxy()

# ========= 爬虫核心逻辑 =========
class EbaySpider:
    def __init__(self, keywords):
        self.keywords = keywords
        self.dedup = DeduplicationMiddleware()
        self.dist = DistributionMiddleware()
        self.headers = {
            "User-Agent": random.choice([
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
            ]),
            "Accept-Language": "en-US,en;q=0.9"
        }

    def fetch(self, keyword):
        search_url = f"https://www.ebay.com/sch/i.html?_nkw={quote(keyword)}"
        if self.dedup.is_duplicate(search_url):
            print(f"跳过重复任务：{keyword}")
            return
        
        proxies = self.dist.select_proxy(keyword)
        try:
            response = requests.get(search_url, headers=self.headers, proxies=proxies, timeout=10)
            response.raise_for_status()
            self.parse(response.text, keyword)
        except Exception as e:
            print(f"抓取失败：{keyword} -> {e}")

    def parse(self, html, keyword):
        soup = BeautifulSoup(html, "html.parser")
        items = soup.select(".s-item")

        for item in items:
            title = item.select_one(".s-item__title")
            price = item.select_one(".s-item__price")
            location = item.select_one(".s-item__location")
            time = item.select_one(".s-item__listingDate")
            
            data = {
                "keyword": keyword,
                "title": title.text.strip() if title else None,
                "price": price.text.strip() if price else None,
                "location": location.text.strip() if location else None,
                "time": time.text.strip() if time else None
            }
            print(json.dumps(data, ensure_ascii=False))

    def run(self):
        for kw in self.keywords:
            self.fetch(kw)

# ========= 运行入口 =========
if __name__ == "__main__":
    keywords = ["laptop", "headphones", "watch", "camera"]
    spider = EbaySpider(keywords)
    spider.run()