前言：关于虚拟手机号进行虚拟用户的注册行为，我们之后再讲。这里由于链家的严格反爬及本人的能力有限，这里不得不考虑降级策略。

降级策略

多网站爬虫

在初始的数据采集方案中，首选目标是链家，因为其房源数据规范、真实度高。然而，链家拥有非常严格的反爬机制（如高频IP封禁、动态加载数据、甚至触发滑块验证码等）。为了保证系统的稳定性和数据的持续产出，我们设计了多源备选方案：

• 第一梯队（主目标）：链家。尝试通过常规请求头伪装和合理的请求间隔进行抓取。
• 第二梯队（备选降级）：当链家触发高强度风控（如连续返回403或验证码）时，自动降级切换至安居客或房天下。这些平台虽然数据异构性强，但反爬强度相对较低，且同样包含大量核心房源信息。

数据库兜底

考虑到房源信息是极具时效性的数据（如“在售”状态可能随时变为“已成交”），我们选择轻量级的 SQLite 数据库作为本地兜底存储。

轻量化：SQLite 无需复杂的服务器配置，非常适合单机爬虫的快速存取。

数据库设计

1. 过期自动清理：为了保证数据的“新鲜度”，我们在数据库中引入了过期时间机制。设置 7天 为数据生命周期，超过7天未更新的房源将被视为过期数据并自动删除，确保用户看到的永远是近期的有效房源。

2. 关系模式： 按照初始设计，设置id，house_id,source,title,rent_type,price,area等字段，设置创建时间方便定时删除。

CREATE TABLE IF NOT EXISTS rental_listings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            house_id TEXT NOT NULL,      -- 房源唯一ID
            source TEXT NOT NULL,        -- 来源：链家/安居客
            title TEXT,
            rent_type TEXT,
            orientation TEXT,
            price REAL,
            city TEXT,
            district TEXT,
            area REAL,
            feature TEXT,
            url TEXT,
            created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            UNIQUE(house_id, source)     -- 联合唯一约束：防止同一平台同一房源重复
        );

保障优先or速度优先

责任链模式（保障优先）

这种模式的核心在于“稳”。我们将不同的房产网站（链家、安居客、房天下）封装成一个个独立的处理器（Handler），串联成一条责任链。

执行逻辑：请求首先交给“链家处理器”。如果链家正常返回数据，则流程结束；如果链家触发反爬（抛出异常或返回无效数据），责任链会自动将请求传递给下一个节点“安居客处理器”，以此类推。

优势：极大提高了系统的容错率。即使某个平台彻底封禁了当前IP，爬虫依然能通过备选平台获取到部分数据，保障业务不中断。

这里简单的使用了try-except链，来进行责任链设计。在询问千问后，仿照spring boot 的过滤链，设计了更高级，更解耦，更符合设计模式的代码。

from abc import ABC, abstractmethod

class DataFetcher(ABC):
    """责任链抽象基类"""
    def __init__(self):
        self._next_fetcher = None

    # 依然保留 set_next，但这次交给管理器在内部自动调用
    def set_next(self, fetcher: "DataFetcher") -> "DataFetcher":
        self._next_fetcher = fetcher
        return fetcher

    @abstractmethod
    def fetch(self, address: str, requirements: str):
        pass

    def _fetch_next(self, address: str, requirements: str):
        if self._next_fetcher:
            return self._next_fetcher.fetch(address, requirements)
        return None

# ... (LianjiaFetcher, AnjukeFetcher, DatabaseFetcher 的具体实现保持完全不变) ...
class LianjiaFetcher(DataFetcher):
    def fetch(self, address: str, requirements: str):
        print(f"🔍 [链家节点] 正在尝试爬取 {address}...")
        # 模拟链家失败，交给下一个
        print("⏭️ [链家节点] 未获取到数据，转交...")
        return self._fetch_next(address, requirements)

class AnjukeFetcher(DataFetcher):
    def fetch(self, address: str, requirements: str):
        print(f"🔍 [安居客节点] 正在尝试爬取 {address}...")
        # 模拟安居客成功
        data = {"source": "安居客", "price": "200万"}
        if data: return data
        return self._fetch_next(address, requirements)

class DatabaseFetcher(DataFetcher):
    def fetch(self, address: str, requirements: str):
        print(f"🔍 [数据库节点] 正在尝试兜底查询...")
        return self._fetch_next(address, requirements)


class FetcherChain:
    """
    爬虫责任链管理器（核心优化点）
    """
    def __init__(self):
        self._fetchers = []  # 用列表来保存所有 add 进来的爬虫节点
        self._head = None    # 链表的头节点

    def add(self, fetcher: DataFetcher):
        """
        类似于 Spring 的 add 方法，直接往里加爬虫即可
        """
        self._fetchers.append(fetcher)
        return self  # 返回 self 是为了支持链式调用，比如 chain.add(A).add(B)

    def _build_chain(self):
        """
        内部方法：在真正执行前，自动把列表里的节点串联起来
        """
        if not self._fetchers:
            return None
        
        # 自动串联：第0个的next指向第1个，第1个指向第2个...
        for i in range(len(self._fetchers) - 1):
            self._fetchers[i].set_next(self._fetchers[i + 1])
        
        # 返回链头
        return self._fetchers[0]

    def fetch(self, address: str, requirements: str):
        """
        对外提供的统一执行入口
        """
        # 每次执行前确保链是串好的（也可以放在 add 里做，看具体需求）
        if not self._head:
            self._head = self._build_chain()
        
        if self._head:
            return self._head.fetch(address, requirements)
        return None





################################################################################
# 在主程序中
if __name__ == "__main__":
    # 1. 创建链管理器
    chain = FetcherChain()

    # 2. 直接 add 各种爬虫和兜底策略（顺序即执行顺序）
    chain.add(LianjiaFetcher()) \
         .add(AnjukeFetcher()) \
         .add(DatabaseFetcher())

    # 3. 一键触发执行
    print("🚀 开始执行爬虫责任链...\n")
    result = chain.fetch()

    print("\n" + "="*30)
    if result:
        print(f"🎉 最终获取到的房源数据：{result}")
    else:
        print("💥 爬取彻底失败！")

竞态模式（速度优先）

这种模式的核心在于“快”。适用于对数据实时性要求极高的场景。

• 执行逻辑：同时向链家和备选网站发起并发请求。谁先成功返回有效数据，就采纳谁的结果，并立即取消或忽略其他平台的请求。
• 优势：在面对网络波动或某个平台响应缓慢时，能以最快速度拿到数据，显著提升整体爬取效率。

这里有一个必要条件，就是我的代码必须是异步代码，但是request库，和selenium是严格同步阻塞的。

速度优先，不得不对代码修改。

request--> aiohttp

aiohttp是request的异步版，修改起来相对简单。

aiohttp 就是 Python 中一款优秀的异步 Web 框架，它能够帮助我们构建高效的异步 Web 应用和异步 HTTP 客户端。

aiohttp 使用 Python 的 async /await 语法来实现异步编程，这使得编写异步代码更加直观和简洁。

这里展示部分代码：

    async def run(self, base_url, city='jn',  district=None, max_pages=1):
        """
        主运行函数
        注意：安居客的分页通常是 URL 直接加 /p2 /p3
        """
        self.city = city
        self.district = district

        print(f"🚀 开始异步爬取: {base_url}")

        # 构建任务列表
        tasks = []
        for page in range(1, max_pages + 1):
            # 安居客分页逻辑示例 (请根据实际 URL 结构调整)
            if page == 1:
                page_url = base_url
            else:
                # 这里是一个通用的拼接逻辑，可能需要根据你的 URL 调整
                # 例如: https://.../lixia/p2/
                page_url = f"{base_url}-p{page}/"

            # 创建任务
            task = self.fetch_and_save(page_url)
            tasks.append(task)

        # 并发执行所有任务
        results = await asyncio.gather(*tasks, return_exceptions=True)
        data = []
        for result in results:
            data.extend(result)
        # total_count = 0
        # for result in results:
        #     if isinstance(result, int):
        #         total_count += result
        #     else:
        #         print(f"任务执行出错: {result}")

        # print(f"📊 爬取完成，共获取 {total_count} 条有效数据。")

        # 关闭 Session
        await self.session.close()
        return data

selenium-->playwright

Playwright 的 API 设计非常现代化，很多 Selenium 中繁琐的步骤被高度封装了。

功能场景	Selenium (旧写法)	Playwright (新写法)	改动感受
元素点击	driver.find_element(By.ID, "btn").click()	page.click("#btn")	更简短，选择器直接内嵌
输入文本	elem = driver.find_element(By.NAME, "q") elem.send_keys("hello")	page.fill("input[name='q']", "hello")	一步到位，无需分两步
元素等待	需手动引入 WebDriverWait 和 expected_conditions	自动等待（内置机制，无需写额外代码）	彻底解放，告别不稳定
弹窗处理	alert = driver.switch_to.alert alert.accept()	page.on("dialog", lambda d: d.accept())	事件监听，逻辑更清晰

这里不再展示代码。