从零实现一个轻量级 Python 爬虫框架（300 行代码）

# mini_spider/core.py
import requests
import threading
from queue import Queue
from lxml import etree
from urllib.parse import urljoin
import time
import hashlib


class Scheduler:
    """URL 调度器，带去重"""
    def __init__(self):
        self.queue = Queue()
        self.seen = set()
        self.lock = threading.Lock()

    def add(self, url):
        fp = hashlib.md5(url.encode()).hexdigest()
        with self.lock:
            if fp in self.seen:
                return False
            self.seen.add(fp)
            self.queue.put(url)
            return True

    def get(self):
        return self.queue.get()

    def empty(self):
        return self.queue.empty()


class Downloader:
    """下载器，带重试与限速"""
    def __init__(self, delay=0.5, retries=3):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'MiniSpider/1.0'
        })
        self.delay = delay
        self.retries = retries
        self.last_time = 0

    def fetch(self, url):
        for i in range(self.retries):
            try:
                wait = self.delay - (time.time() - self.last_time)
                if wait > 0:
                    time.sleep(wait)
                self.last_time = time.time()
                resp = self.session.get(url, timeout=10)
                resp.raise_for_status()
                return resp.text
            except Exception as e:
                print(f'[{i+1}/{self.retries}] 失败 {url}: {e}')
        return None

class Spider:
    def __init__(self, start_urls, parse_func, workers=5):
        self.scheduler = Scheduler()
        self.downloader = Downloader()
        self.parse = parse_func
        self.workers = workers
        self.results = []
        for u in start_urls:
            self.scheduler.add(u)

    def worker(self):
        while True:
            if self.scheduler.empty():
                break
            url = self.scheduler.get()
            html = self.downloader.fetch(url)
            if not html:
                continue
            data, new_urls = self.parse(url, html)
            if data:
                self.results.append(data)
            for u in new_urls:
                self.scheduler.add(urljoin(url, u))

    def run(self):
        threads = [threading.Thread(target=self.worker) for _ in range(self.workers)]
        for t in threads: t.start()
        for t in threads: t.join()
        return self.results

def parse_page(url, html):
    tree = etree.HTML(html)
    items = tree.xpath('//div[@class="article"]')
    data = []
    for it in items:
        data.append({
            'title': ''.join(it.xpath('./h2//text()')).strip(),
            'link': ''.join(it.xpath('./h2/a/@href')),
        })
    next_urls = tree.xpath('//a[@class="next"]/@href')
    return data, next_urls


spider = Spider(
    start_urls=['https://example.com/page/1'],
    parse_func=parse_page,
    workers=8
)
results = spider.run()
print(f'共抓取 {len(results)} 条')