☰
Python爬虫实战:链家网北京租房信息抓取与TaoToken配置避坑指南
2026/9/26 19:39:22 网站建设 项目流程

1. 链家北京租房抓取的真实场景与核心难点

链家网北京租房信息抓取,是很多做房产数据分析、租房比价工具、区域租金行情看板的朋友绕不开的一关。它页面结构规整、字段丰富,看起来很适合练手,但真正跑起来你会发现三个现实问题:一是列表页和详情页的字段分布在不同层级,正则一改版就崩;二是请求频率稍高就会触发验证或返回空数据;三是抓下来的字段经常出现空值、错位,落库时才发现对不上。

这篇内容面向的是已经会写基础 requests 请求、想把这套流程做稳的读者。我会从请求头构造、分页策略、详情页字段抽取、频率控制,一直讲到数据落盘和字段校验。同时把 TaoToken 的统一 Key 接入配置一起交付,因为很多人在做多模型辅助解析房源文本、或者用模型做字段纠错时,Key 管理混乱、环境变量到处散落,反而拖慢了调试节奏。TaoToken 在这里的角色是统一管理模型调用凭证,让爬虫项目里的辅助解析环节有一个稳定的配置入口,而不是每次换模型就改一遍代码。

需要先说明的是,抓取行为要遵守目标站点的 robots 协议和相关规定,控制频率、只取公开信息、不用于商业转售,这是前提。下面的配置和代码都是围绕“稳定、可校验、可复现”来设计的。

2. TaoToken 前置准备:统一 Key 与 config.toml 骨架

在爬虫项目里引入模型能力,最常见的痛点是 Key 散落在各个脚本、测试环境和正式环境混用。TaoToken 提供统一的 API 入口,把模型调用凭证集中管理。你可以在官网了解整体能力,实际接入时用 API 地址https://taotoken.net/api。

先注册并拿到 Key,入口在控制台的 API Keys 页面。拿到之后不要硬编码进脚本,而是写进config.toml,用环境变量兜底。下面是我实测下来比较稳的骨架:

# config.toml [app] name = "lianjia_bj_zufang" timeout = 15 retry = 3 [request] # 列表页与详情页共用请求头 user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" referer = "https://bj.lianjia.com/zufang/" accept_language = "zh-CN,zh;q=0.9" # 请求间隔(秒),控制频率的核心参数 min_delay = 2.5 max_delay = 5.0 page_start = 1 page_end = 20 [storage] # 落盘方式:csv 或 mysql mode = "csv" csv_path = "./data/lianjia_bj.csv" [taotoken] base_url = "https://taotoken.net/api" api_key = "${TAOTOKEN_API_KEY}" model = "claude-3-5-sonnet" # 用于房源文本字段纠错与补全 enable_field_fix = true

这里api_key用${TAOTOKEN_API_KEY}占位,运行时从环境变量读取,避免把凭证提交到仓库。模型对话相关的调试可以直接在模型对话页面验证返回是否符合预期,确认没问题再写进爬虫的辅助流程。

读取配置的代码:

import os import toml def load_config(path="config.toml"): cfg = toml.load(path) # 环境变量覆盖,优先级最高 env_key = os.getenv("TAOTOKEN_API_KEY") if env_key: cfg["taotoken"]["api_key"] = env_key return cfg CFG = load_config()

这样本地开发用环境变量,CI 或服务器上注入密钥,代码本身不含敏感信息。

3. 可复制配置:请求头、分页与详情页字段抽取

链家列表页的 URL 结构是https://bj.lianjia.com/zufang/{区域}/pg{页码}/,比如朝阳区就是chaoyang。分页从 1 开始,超过实际页数会返回空列表,所以要在解析时判断是否还有房源,及时停止,而不是死循环到 100 页。

请求头里最关键的是User-Agent和Referer,缺了 Referer 有时会拿到不完整页面。下面是一个带重试和随机间隔的请求封装:

import time import random import requests def build_headers(cfg): return { "User-Agent": cfg["request"]["user_agent"], "Referer": cfg["request"]["referer"], "Accept-Language": cfg["request"]["accept_language"], } def fetch(url, cfg, session=None): headers = build_headers(cfg) sess = session or requests.Session() for attempt in range(cfg["app"]["retry"]): try: resp = sess.get(url, headers=headers, timeout=cfg["app"]["timeout"]) if resp.status_code == 200 and "zufang" in resp.text: return resp.text # 状态码异常或页面异常,等待后重试 time.sleep(random.uniform(cfg["request"]["min_delay"], cfg["request"]["max_delay"])) except requests.RequestException as e: print(f"请求异常 {url}: {e}") time.sleep(random.uniform(cfg["request"]["min_delay"], cfg["request"]["max_delay"])) return None

分页循环要带终止判断:

def crawl_list(area, cfg): session = requests.Session() results = [] for page in range(cfg["request"]["page_start"], cfg["request"]["page_end"] + 1): url = f"https://bj.lianjia.com/zufang/{area}/pg{page}/" html = fetch(url, cfg, session) if not html: print(f"第 {page} 页无数据,停止") break items = parse_list(html) if not items: print(f"第 {page} 页解析为空,停止") break results.extend(items) print(f"第 {page} 页抓到 {len(items)} 条") time.sleep(random.uniform(cfg["request"]["min_delay"], cfg["request"]["max_delay"])) return results

列表页解析建议用 lxml 或 BeautifulSoup,比正则稳。详情页字段多,用 CSS 选择器逐项取,取不到就留空,不要因为一个字段缺失整条丢弃:

from bs4 import BeautifulSoup def parse_list(html): soup = BeautifulSoup(html, "lxml") items = [] for node in soup.select("div.content__list--item"): title_node = node.select_one("p.content__list--item--title a") if not title_node: continue items.append({ "title": title_node.get_text(strip=True), "url": "https://bj.lianjia.com" + title_node.get("href", ""), }) return items def parse_detail(html): soup = BeautifulSoup(html, "lxml") def pick(selector): el = soup.select_one(selector) return el.get_text(strip=True) if el else "" return { "price": pick("span.content__aside--title span"), "method": pick("ul.content__aside--info li:nth-child(1)"), "square": pick("ul.content__aside--info li:nth-child(2)"), "orientation": pick("ul.content__aside--info li:nth-child(3)"), }

字段抽取完,落盘前做一次校验,价格必须是数字、面积必须带“㎡”、标题非空,不满足的记录单独存到reject.csv,方便回头排查是页面改版还是解析规则失效。

4. 验证请求与成功结果:频率控制与字段校验动作

配置写完,先做小规模验证,不要一上来就跑全量。把page_end设成 2,跑一遍看输出:

if __name__ == "__main__": cfg = load_config() data = crawl_list("chaoyang", cfg) print(f"共抓取 {len(data)} 条") for row in data[:3]: print(row)

成功时你会看到类似输出:

第 1 页抓到 30 条 第 2 页抓到 30 条 共抓取 60 条 {'title': '整租·朝阳门外·...', 'url': 'https://bj.lianjia.com/zufang/BJ...'}

如果第 1 页就返回 0 条,先检查请求头里的 Referer 和 User-Agent 是否被识别,再确认区域拼音是否正确。频率控制上,min_delay和max_delay不要低于 2 秒,实测低于 1.5 秒连续请求,几页之后就会拿到空页面。字段校验用一个独立函数,落盘前统一过一遍:

import re def validate(row): errors = [] if not row.get("title"): errors.append("title_empty") price = row.get("price", "") if price and not re.match(r"^\d+", price): errors.append("price_invalid") return errors

校验不通过的记录写入reject.csv并记录原因,通过的写入正式数据文件。这样即使页面改版,你也能从 reject 文件里快速定位是哪一类字段先失效。

5. 本篇常见错排查:空数据、字段错位与 Key 失效

跑链家抓取时,下面几个错误出现频率最高,我按现象、原因、处理方式列出来,方便对照。

现象可能原因处理方式
列表页返回空频率过高被限流调大 min_delay,加随机抖动
详情页字段全空选择器随改版失效用浏览器检查元素,更新 CSS 选择器
价格字段混入文字正则匹配范围过宽改用选择器取节点文本再清洗
落库字段错位列表与详情顺序不一致用 url 作为主键关联,不靠下标
TaoToken 调用 401Key 未注入或过期检查环境变量,重新生成 Key
模型返回超时单次文本过长截断输入,或分批调用

关于 TaoToken 的 Key 失效,最常见的坑是把 Key 写死在脚本里,换环境后忘了改。用config.toml加环境变量的方式能规避大部分问题。如果调用报错,先去 API Keys 页面确认 Key 状态,再对照接入文档检查请求格式。模型返回异常时,可以单独在模型对话里复现同样的输入,确认是模型侧还是爬虫侧的问题。

还有一个容易忽略的点:详情页请求也要带 Referer,且 Referer 最好指向列表页,而不是详情页自身,否则部分页面会返回精简版。字段错位问题,根源往往是用列表下标去对应详情结果,一旦某条详情抓取失败,后面全部错位。正确做法是以 url 为键做字典映射,落盘时再按 url 组装。

6. 语义一致 CTA:把配置和 Key 管理固定下来

整套流程跑通之后,真正影响长期稳定的是配置管理和 Key 管理这两件事。把config.toml作为唯一配置入口,请求参数、频率、落盘方式、模型接入全部集中,换区域、换页数、换模型都只改一个文件。TaoToken 的 Key 通过环境变量注入,配合统一 API 入口,多脚本共用一套凭证,不用在每个爬虫里重复维护。

如果你后续要做长期编码或 Agent 化的数据流水线,可以了解 Coding Plan,把模型调用纳入统一的开发计划里。需要验证模型返回是否符合字段纠错预期时,直接在模型对话里试。接入细节和请求格式以接入文档为准,Key 的生成和管理在 API Keys 页面完成。把这些固定下来,链家北京租房信息的抓取和后续分析就能稳定复现,而不是每次跑都靠运气。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询