我把 12 个 Scrapy 爬虫项目迁到 Crawlo，踩了这些坑

# Scrapy 写法
class MySpider(scrapy.Spider):
    name = "example"

    def start_requests(self):
        urls = ["https://example.com/page/1", "https://example.com/page/2"]
        for url in urls:
            yield scrapy.Request(url, callback=self.parse)

    def parse(self, response):
        for item in response.css(".item"):
            yield {"title": item.css("h2::text").get()}

# Crawlo 写法（第一版，有坑）
class MySpider(CrawloSpider):
    name = "example"

    async def start_requests(self):
        urls = ["https://example.com/page/1", "https://example.com/page/2"]
        for url in urls:
            yield Request(url, callback=self.parse)  # ⚠️ 这里有问题

    async def parse(self, response):
        for item in response.css(".item"):
            yield {"title": item.css("h2::text").get()}

# Crawlo 写法（第一版，有坑）
class MySpider(CrawloSpider):
    name = "example"

    async def start_requests(self):
        urls = ["https://example.com/page/1", "https://example.com/page/2"]
        for url in urls:
            yield Request(url, callback=self.parse)  # ⚠️ 这里有问题

    async def parse(self, response):
        for item in response.css(".item"):
            yield {"title": item.css("h2::text").get()}

# Scrapy 中间件
class CustomHeaderMiddleware:
    def process_request(self, request, spider):
        request.headers["X-Custom"] = "my-value"
        return None  # 返回 None 继续流程

    def process_response(self, request, response, spider):
        return response

# Crawlo 中间件
class CustomHeaderMiddleware:
    async def process_request(self, request, spider):
        request.headers["X-Custom"] = "my-value"
        # 不返回值，或返回 None

    async def process_response(self, request, response, spider):
        return response

# 迁移辅助脚本：反转中间件优先级
def convert_middleware_priority(scrapy_settings: dict) -> dict:
    """将 Scrapy 中间件优先级转换为 Crawlo 优先级"""
    crawlo_settings = {}
    for key, value in scrapy_settings.items():
        if isinstance(value, dict):
            crawlo_settings[key] = {
                k: -v if isinstance(v, int) else v  # 取负值反转顺序
                for k, v in value.items()
            }
        else:
            crawlo_settings[key] = value
    return crawlo_settings

# Scrapy Pipeline
class MySQLPipeline:
    def open_spider(self, spider):
        self.pool = aiomysql.create_pool(host="127.0.0.1", ...)

    def process_item(self, item, spider):
        # 写入逻辑
        return item

    def close_spider(self, spider):
        self.pool.close()

# Crawlo Pipeline（修正版）
class MySQLPipeline:
    async def open_spider(self, spider):
        # 只做不依赖引擎状态的初始化
        self.pool = await aiomysql.create_pool(host="127.0.0.1", ...)

    async def process_item(self, item, spider):
        async with self.pool.acquire() as conn:
            await conn.execute("INSERT INTO ...", ...)
        return item

    async def close_spider(self, spider):
        self.pool.close()
        await self.pool.wait_closed()

# Crawlo 自定义去重过滤器
from crawlo.dupefilter import BaseDupeFilter
from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse

class NormalizedURLDupeFilter(BaseDupeFilter):
    """规范化 URL 后去重，处理参数排序和空值问题"""

    def __init__(self):
        self.seen = set()

    def request_fingerprint(self, request):
        url = request.url
        parsed = urlparse(url)
        # 解析 query 参数，过滤空值，排序
        params = [(k, v) for k, v in parse_qsl(parsed.query) if v]
        params.sort()
        normalized_query = urlencode(params)
        normalized_url = urlunparse((
            parsed.scheme, parsed.netloc, parsed.path,
            parsed.params, normalized_query, ""  # 丢弃 fragment
        ))
        return f"{request.method}:{normalized_url}"

    def is_duplicate(self, request):
        fp = self.request_fingerprint(request)
        if fp in self.seen:
            return True
        self.seen.add(fp)
        return False

# 请求队列迁移脚本
import pickle
import json
import redis

r = redis.Redis(host="127.0.0.1", port=6379, db=0)

def migrate_scrapy_queue_to_crawlo(scrapy_queue: str, crawlo_queue: str):
    """将 scrapy-redis 的 List 队列迁移到 Crawlo 的 Sorted Set"""
    count = 0
    while True:
        # 从 Scrapy 队列右侧弹出
        raw = r.rpop(scrapy_queue)
        if raw is None:
            break

        try:
            # Scrapy 的 Request 是 pickle 序列化的
            scrapy_request = pickle.loads(raw)
            # 转换为 Crawlo 格式的 JSON
            crawlo_request = {
                "url": scrapy_request.url,
                "method": scrapy_request.method,
                "headers": dict(scrapy_request.headers),
                "body": scrapy_request.body.decode() if scrapy_request.body else "",
                "meta": scrapy_request.meta,
                "priority": getattr(scrapy_request, "priority", 0),
            }
            # 写入 Crawlo 的 Sorted Set（score = priority）
            r.zadd(crawlo_queue, {json.dumps(crawlo_request): crawlo_request["priority"]})
            count += 1
        except Exception as e:
            print(f"迁移失败: {e}, URL: {scrapy_request.url if 'scrapy_request' in dir() else 'unknown'}")

    print(f"迁移完成，共 {count} 个请求")

# HTML: <div class="title">  Hello World  </div>
# Scrapy
response.css(".title::text").get()  # 返回 "  Hello World  "（保留空格）
response.css(".title::text").getall()  # 返回 ["  Hello World  "]

# Crawlo
response.css(".title::text").get()  # 返回 "Hello World"（自动 strip！）
response.css(".title::text").getall()  # 返回 ["Hello World"]

mport httpx
from crawlo.exceptions import IgnoreRequest


class YiniuProxyPool:
    """亿牛云代理池：封装隧道代理与动态提取两种模式。

    亿牛云（ip3366.cn）的隧道代理按请求自动换出口 IP，
    特别适合高并发下的换源重试场景。

    Args:
        username: 亿牛云账号用户名
        password: 亿牛云授权密码
        tunnel_host: 隧道代理域名，默认 tunnel.ip3366.cn
        tunnel_port: 隧道代理端口，默认 39080
        dynamic_api: 动态短效代理提取接口
    """

    def __init__(
        self,
        username: str,
        password: str,
        tunnel_host: str = "tunnel.ip3366.cn",
        tunnel_port: int = 39080,
        dynamic_api: str = "http://http.tiqu.ip3366.cn:30376/api/getip",
    ) -> None:
        self._auth = f"{username}:{password}"
        self._tunnel = f"http://{self._auth}@{tunnel_host}:{tunnel_port}"
        self._dynamic_api = dynamic_api

    def get_tunnel_proxy(self) -> str:
        """隧道代理入口：亿牛云每次请求自动轮换出口 IP，重试即换源。"""
        return self._tunnel

    async def get_fresh_proxy(self) -> str:
        """动态代理：实时从亿牛云 API 提取一个最新可用 IP。"""
        async with httpx.AsyncClient(timeout=5) as client:
            resp = await client.get(
                self._dynamic_api,
                params={"username": self._auth.split(":")[0],
                        "password": self._auth.split(":")[1],
                        "count": 1, "type": 1},
            )
            ip, port = resp.text.strip().split(":")
            return f"http://{self._auth}@{ip}:{port}"


class SmartRetryMiddleware:
    """自定义重试中间件：重试时切换亿牛云代理。

    隧道代理每次请求自动换出口 IP，因此重试时直接复用隧道入口
    即可实现「换 IP 重试」，无需额外维护 IP 池状态。

    Args:
        proxy_pool: YiniuProxyPool 实例
    """

    RETRY_CODES = {403, 429, 502, 503, 504}
    MAX_RETRY = 3

    def __init__(self, proxy_pool: YiniuProxyPool) -> None:
        self.proxy_pool = proxy_pool

    async def process_response(self, request, response, spider):
        if response.status not in self.RETRY_CODES:
            return response

        retry_count = request.meta.get("retry_count", 0)
        if retry_count >= self.MAX_RETRY:
            spider.logger.warning(f"超过最大重试次数: {request.url}")
            raise IgnoreRequest(request)

        # 构造新的重试请求，并直接换上亿牛云隧道代理（后端自动换 IP）
        retry_request = request.copy()
        retry_request.meta["retry_count"] = retry_count + 1
        retry_request.meta["proxy"] = self.proxy_pool.get_tunnel_proxy()
        retry_request.dont_filter = True  # 避免被去重过滤

        spider.logger.info(
            f"重试 {retry_count + 1}/{self.MAX_RETRY}: {request.url} "
            f"(status={response.status}, 已切换亿牛云代理)"
        )
        return retry_request

class ProxyMiddleware:
    async def process_request(self, request, spider):
        if request.meta.get("need_new_proxy"):
            # 强制获取新代理，不用缓存
            proxy = await self.proxy_pool.get_fresh()
        else:
            proxy = await self.proxy_pool.get()
        request.meta["proxy"] = proxy

# stats 适配层
class ScrapyCompatibleStats:
    """将 Crawlo stats 映射为 Scrapy stats 命名，兼容现有监控"""

    MAPPING = {
        "crawler_response_count": "response_received_count",
        "crawler_item_count": "item_scraped_count",
        "crawler_duration_seconds": "elapsed_time_seconds",
        "crawler_retry_total": "retry/count",
        "crawler_error_count": "log_count/ERROR",
    }

    def __init__(self, crawlo_stats):
        self._stats = crawlo_stats

    def get_stats(self) -> dict:
        raw = self._stats.get_stats()
        mapped = {}
        for key, value in raw.items():
            scrakey = self.MAPPING.get(key, key)
            mapped[scrakey] = int(value) if isinstance(value, float) and scrakey.endswith("_count") else value
        return mapped

# logging 配置
LOGGING = {
    "version": 1,
    "formatters": {
        "scrapy_compatible": {
            "format": "[%(name)s] %(levelname)s: %(message)s",
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "scrapy_compatible",
        }
    },
    "root": {
        "handlers": ["console"],
        "level": "INFO",
    }
}