从 Scrapy 迁移到 Crawlee Python：保留字段契约，用双跑结果决定切换

export BOOKS_PAGE_1_URL='<paste-the-actual-books-page-1-url>'
export BOOKS_PAGE_2_URL='<paste-the-actual-books-page-2-url>'

python -m venv .venv
source .venv/bin/activate
python -m pip install "Scrapy==2.13.4" "crawlee[parsel]==1.11.1"

import os
from decimal import Decimal
 
START_URLS = [
    os.environ["BOOKS_PAGE_1_URL"],
    os.environ["BOOKS_PAGE_2_URL"],
]
 
 
def normalize(selector, source_url):
    title = selector.css(".product_main h1::text").get()
    price = selector.css(".product_main .price_color::text").get()
    upc = selector.xpath(
        "//table//tr[th[normalize-space()='UPC']]/td/text()"
    ).get()
    if not title or not price or not upc:
        raise ValueError("missing_required_product_field")
    title, price, upc = title.strip(), price.strip(), upc.strip()
    if not title or not upc or not price.startswith("£"):
        raise ValueError("invalid_product_field")
    cents = Decimal(price[1:]) * 100
    if not cents.is_finite() or cents < 0 or cents != cents.to_integral_value():
        raise ValueError("invalid_gbp_amount")
    return {
        "upc": upc,
        "title": title,
        "price_minor": int(cents),
        "currency": "GBP",
        "price_display": price,
        "source_url": source_url,
    }

import scrapy
 
from contract import START_URLS, normalize
 
 
class ProductsSpider(scrapy.Spider):
    name = "migration_baseline"
    allowed_domains = ["books.toscrape.com"]
    start_urls = START_URLS
    custom_settings = {
        "CONCURRENT_REQUESTS": 2,
        "DOWNLOAD_DELAY": 0.5,
        "RETRY_TIMES": 2,
        "FEED_EXPORT_ENCODING": "utf-8",
    }
 
    def parse(self, response):
        links = response.css("article.product_pod h3 a::attr(href)").getall()
        if not links:
            raise ValueError("list_has_no_product_links")
        for href in links:
            yield response.follow(href, callback=self.parse_detail)
 
    def parse_detail(self, response):
        yield normalize(response, response.url)

scrapy runspider old_spider.py -O old-products.jsonl 2> old-run.log

import asyncio
import json
from pathlib import Path
from urllib.parse import urljoin
 
from crawlee import ConcurrencySettings
from crawlee.crawlers import BasicCrawlingContext, ParselCrawler, ParselCrawlingContext
 
from contract import START_URLS, normalize
 
 
def save_jsonl(path, rows):
    Path(path).write_text(
        "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
        encoding="utf-8",
    )
 
 
async def main():
    products = {}
    failures = []
    list_urls = set()
    discovered = set()
    crawler = ParselCrawler(
        max_requests_per_crawl=60,
        max_request_retries=2,
        concurrency_settings=ConcurrencySettings(
            min_concurrency=1, max_concurrency=2
        ),
    )
 
    @crawler.router.default_handler
    async def listing(context: ParselCrawlingContext):
        if context.request.url not in START_URLS:
            raise ValueError("unexpected_list_url")
        links = context.selector.css(
            "article.product_pod h3 a::attr(href)"
        ).getall()
        if not links:
            raise ValueError("list_has_no_product_links")
        discovered.update(urljoin(context.request.url, href) for href in links)
        await context.enqueue_links(
            selector="article.product_pod h3 a", label="DETAIL"
        )
        list_urls.add(context.request.url)
 
    @crawler.router.handler("DETAIL")
    async def detail(context: ParselCrawlingContext):
        record = normalize(context.selector, context.request.url)
        previous = products.get(record["upc"])
        if previous is not None and previous != record:
            raise ValueError("conflicting_duplicate_upc")
        products[record["upc"]] = record
 
    @crawler.failed_request_handler
    async def failed(context: BasicCrawlingContext, error: Exception):
        failures.append({
            "url": context.request.url,
            "label": context.request.label,
            "error_type": type(error).__name__,
            "error": str(error),
        })
 
    try:
        await crawler.run(START_URLS)
    finally:
        save_jsonl("new-products.jsonl", [
            products[key] for key in sorted(products)
        ])
        Path("new-errors.json").write_text(
            json.dumps(failures, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
 
    actual = {row["source_url"] for row in products.values()}
    if failures or list_urls != set(START_URLS) or actual != discovered:
        raise RuntimeError("incomplete_migration_batch_check_outputs")
 
 
if __name__ == "__main__":
    asyncio.run(main())

CRAWLEE_STORAGE_DIR=./storage-migration-01 python new_crawler.py

import json
from pathlib import Path
 
 
def load(path):
    rows = [
        json.loads(line)
        for line in Path(path).read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    result = {}
    for row in rows:
        key = row["upc"]
        if key in result:
            raise ValueError("duplicate_upc_in_" + path + ":" + key)
        if type(row["price_minor"]) is not int or row["currency"] != "GBP":
            raise ValueError("schema_mismatch_in_" + path)
        result[key] = row
    return result
 
 
old = load("old-products.jsonl")
new = load("new-products.jsonl")
missing = sorted(old.keys() - new.keys())
extra = sorted(new.keys() - old.keys())
changed = {
    key: {
        field: {"old": old[key].get(field), "new": new[key].get(field)}
        for field in sorted(old[key].keys() | new[key].keys())
        if old[key].get(field) != new[key].get(field)
    }
    for key in sorted(old.keys() & new.keys())
    if old[key] != new[key]
}
report = {
    "old_count": len(old), "new_count": len(new),
    "missing": missing, "extra": extra, "changed": changed,
}
Path("comparison.json").write_text(
    json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
)
print(json.dumps(report, ensure_ascii=False, indent=2))
if len(old) != 40 or len(new) != 40 or missing or extra or changed:
    raise SystemExit("migration_comparison_failed")