企业官网建设流程全解析

简介：安居客Python爬虫代码是一个面向Python爬虫学习者和房产数据研究者的示例项目，主要用于获取安居客平台经纪人信息（手机号、店铺名称、公司名称、营业执照编号等），通过解析二维码图片跳转详情页完成数据提取。资源共4个文件，包含1个核心爬虫脚本AJK_Phone.py、2个全国城市请求地址表CSV（utf-8与ansi两个编码版本）以及1个readme说明文档，压缩包仅15KB，代码体量精简，适合快速阅读和二次开发。已有113人学习/下载。项目基于Scrapy框架搭建采集逻辑，配合pandas将结果输出为CSV，并预留阿布云代理IP配置入口和后期多线程扩展思路，可以帮助读者掌握从页面解析、图片链接关联到数据落地的完整爬虫流程；同时，csv文件提供了全国城市请求地址表，便于按城市维度扩展抓取范围，readme文档则补充了运行方式与维护计划，适合具备一定Python基础、想了解Scrapy工程化爬虫写法的开发者参考。

1. 安居客Python爬虫代码.zip：从下载到可运行，你需要补完的工程细节

拿到一个名为“安居客Python爬虫代码.zip”的压缩包，和拿到一份能直接python main.py就跑出数据的程序，中间隔着的通常是十几个小时的排错。这类资源在网盘和代码托管平台流传很广，但大多数是半成品：有的基于旧版requests + BeautifulSoup写成，选择器还停留在几年前的 class 名；有的是单个文件堆了三千行，没有requirements.txt，更没有异常处理；真正能落库、能增量更新、能在反爬升级后存活下来的，往往不到三分之一。

这篇不准备“赏析”某个具体压缩包，而是给你一套把这类代码变成可用工程的方法论。整体思路分四块：先看清安居客页面结构和反爬策略，再拆解列表页与详情页的解析逻辑，随后处理 Cookie、代理和验证码，最后解决并发与数据落库。全程围绕“拿到 zip 之后怎么盘活它”这个真实场景，兼容偶尔需要用命令行跑脚本、或者想在本地 IDE 里断点调试的两种习惯。

2. 解析安居客页面结构：列表页、详情页与反爬的对应关系

2.1 先分清三种页面类型，再写选择器

安居客官网的房源信息分布在三个层级：城市首页的区域列表、搜索条件过滤后的房源列表页、以及单套房源的详情页。绝大多数“爬虫代码.zip”只覆盖了第二和第三种，因为城市首页的动态渲染数据接口并不公开，而列表页的 HTML 里确实嵌着完整字段。

列表页 URL 常见格式为：

https://{city}.anjuke.com/sale/p{page_no}/

• {city}为城市拼音，如sh、bj、gz。
• p{page_no}是页码，p1是第一页。
• 每页默认 25 条房源。

详情页 URL 则一般形如：

https://{city}.anjuke.com/props/view/{house_id}

house_id是房源唯一标识，也是后续去重和增量更新的主键。用 Requests 模拟访问时，建议先请求一次列表页，确认服务器返回的状态码和Content-Encoding。一个容易踩的坑是：安居客启用 gzip 压缩传输，而你本地代码没有在请求头中声明Accept-Encoding: gzip，导致拿到的是乱码正文。

2.1.1 列表页解析：定位房源卡片容器

用开发者工具查看列表页源码，可以看到每条房源数据都位于div.property-content或li.property-item容器内。旧版代码常用的 class 名div.houseList-item已经不再返回数据，这是压缩包失效的头号原因。

from parsel import Selector def parse_list_page(html: str) -> list[dict]: sel = Selector(text=html) items = [] for card in sel.css("div.property-content"): house_url = card.css("a.property-content-title-link::attr(href)").get() house_id = house_url.rstrip("/").split("/")[-1] if house_url else "" if not house_id: continue items.append({ "house_id": house_id, "title": card.css("span.property-content-info-main::text").get(), "url": house_url, }) return items

这段代码没有用正则去匹配整个 HTML，而是先通过容器选择器把物理上的房源卡片圈定，再在卡片内部取值。这样做的原因是：列表页底部还有其他区块包含a标签，如果不限定容器，会把“热门搜索词”里的链接误当房源。

参数说明：

• div.property-content是当前版本房源卡片的根节点，不同城市可能微调，建议用response.xpath("//div[contains(@class,'property-content')]")做兜底。
• house_url.rstrip("/").split("/")[-1]是为了兼容 URL 末尾带不带斜杠两种情况。

列表页解析完成后，通常还会解析出价格与小区名。这两个字段的 class 名在不同城市间不太稳定，处理策略是：先抓 3 条样本，在浏览器里逐个核对 class 名，再确认代码中的取值逻辑。不要盲信压缩包里注释里的字段名。

2.2 详情页才是真正拿数据的关卡

列表页给的字段只有标题、URL、价格和基础标签，真正有价值的是户型、朝向、楼层、建造年代、配套信息。这些必须进入详情页抓取。详情页反爬比列表页严格得多：连续访问 30 条以上 URL，通常就会触发滑块验证或强制跳转到登录页。

import requests def fetch_detail(house_id: str, cookies: dict) -> str | None: url = f"https://sh.anjuke.com/props/view/{house_id}" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": f"https://sh.anjuke.com/sale/p1/", } resp = requests.get(url, headers=headers, cookies=cookies, timeout=10) if resp.status_code == 200 and "验证中心" not in resp.text: return resp.text return None

这里把Referer设为列表页地址是必要的。安居客服务端会校验来源页面，如果直接带一个空 Referer 或搜索引擎来源，大概率返回 302 跳转到验证页。cookies参数传入的是浏览器中导出的 Cookie，主要包含sid、aQQ_ajk等标识字段。

2.2.1 详情页字段提取的稳定性方案

详情页的数据分散在两个位置：标题附近的div.house-title和中间区域的ul.house-info-list。有些字段（如挂牌时间）在 vue 渲染后的 DOM 里，而requests.get拿到的源码里没有，这时需要从页面内嵌的__INITIAL_STATE__或window.pageData变量中截取 JSON。

import json import re def extract_json_field(html: str, key: str) -> str | None: m = re.search(rf'"{key}"\s*:\s*"([^"]+)"', html) return m.group(1) if m else None

这种正则提取属于“够用就行”的方案，不适合大批量抓取。更稳妥的做法是找到源码中window.pageData = {...};的位置，整体截取后用json.loads解析。注意json.loads可能因为页面上残留的undefined或NaN抛异常，解析前先替换：

import re raw = re.search(r"window\.pageData\s*=\s*(\{.*?\});", html, re.S) if raw: text = raw.group(1).replace("undefined", "null").replace("NaN", "0") data = json.loads(text)

这段处理在压缩包代码里基本看不到，但它是跨过详情页解析最常见的补丁。

3. 把压缩包代码盘活：修复失效选择器与运行时依赖

3.1 运行前先做三件事：建虚拟环境、装依赖、验证网络

拿到 zip 后依次执行下面三条命令。这里要求 Python 版本不低于 3.9，因为parsel和httpx对低版本支持不够新。

python -m venv .venv source .venv/bin/activate # Windows 下为 .venv\Scripts\activate python -m pip install requests parsel scrapy pandas python -c "import requests; print(requests.get('https://www.anjuke.com').status_code)"

最后一条命令如果返回 200，说明本地网络能直连站点；如果返回 403 或超时，先检查系统代理设置。很多人的代码本身没问题，但环境变量里残留了系统代理，导致请求被转发到不存在的代理端口。

依赖安装完成后再跑压缩包内的入口文件，大概率会报两类错误：

• ModuleNotFoundError: No module named 'xxx'：缺失依赖，逐个pip install补上。
• AttributeError: 'Selector' object has no attribute 'xpath'：混用了parsel和scrapy的 API。scrapy.selector的接口与parsel基本一致，但对象导入路径不同，统一改成from parsel import Selector即可。

3.2 修复选择器的通用排查法：用响应正文反查

压缩包代码里写得最多的解析方式是soup.find("div", class_="houseList-item")。这套选择器在 2023 年以后几乎全部失效，因为页面改版后去掉了这些 class。修复时不要靠猜，三步走：

resp = requests.get(url, headers=headers) with open("debug_anjuke.html", "w", encoding="utf-8") as f: f.write(resp.text)

然后将debug_anjuke.html拖进浏览器，用Ctrl+F搜索你关心的字段文本，例如“3室2厅”。观察这段文本所在的 DOM 层级，重新写选择器，然后用parsel快速验证：

python -c "from parsel import Selector; s=Selector(text=open('debug_anjuke.html').read()); print(s.css('div.house-info a::text').getall())"

这里的div.house-info a::text是我当前使用的选择器，适用于上海区域列表页的户型字段。你在自己的城市页面里可能需要改成其他 class。重点不是这个选择器本身，而是“下载 HTML 到本地 → 浏览器检查 → 用 CSS/XPath 快速验证”这套调试闭环，适合任何失效的爬虫代码。

3.3 用日志替代 print，搞清程序到底死在哪一步

压缩包里的爬虫如果用了大量print(html)，建议第一时间全局替换成标准库logging。否则跑起来屏幕上全是 HTML 片段，根本定位不了逻辑错误。

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", filename="anjuke_spider.log", filemode="a", ) logger = logging.getLogger("anjuke") logger.info("开始抓取第 %s 页列表", page_no) try: html = fetch_listing_page(page_no) except Exception as exc: logger.warning("第 %s 页请求失败: %s", page_no, exc)

用logger.info记录页数、房源数、耗时三个指标，再用logger.warning记录中间件重试信息。这样三个月后再回看，能清晰地判断爬虫是什么时候被验证码卡住、哪个页面的响应异常率开始升高。

参数说明：

• filename指定日志输出到文件，避免终端日志被系统清理。
• level="INFO"让requests和urllib3的底层日志不会刷屏，需要排查网络握手细节时才改成DEBUG。

4. 爬虫并发设计：从单线程到可控协程，到底哪个好

4.1 单线程是压缩包默认方案，但速度不可接受

传统的for循环逐个请求列表页和详情页，请求之间间隔 1～2 秒，跑完 200 页约耗时 30 分钟。这个速度用于测试没问题，用于生产显然不够。于是很多人把希望寄托在多线程上，结果被反爬识别，封禁 IP 的频率反而更高。

这里有个判断标准：如果你的目标只是拿某个城市某几个区域的房源样本，单线程完全够用；如果要做城市级全量数据，就必须上并发，同时配上代理池和请求间隔的随机化。

4.2 用concurrent.futures.ThreadPoolExecutor控制并发度

from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS = 5 def crawl_house_detail(house_id: str, cookies: dict) -> dict | None: html = fetch_detail(house_id, cookies) if html is None: return None return extract_detail(html) with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = [ executor.submit(crawl_house_detail, item["house_id"], cookies) for item in listing_items ] for future in as_completed(futures): result = future.result() if result: save_to_db(result)

这段代码的核心参数是MAX_WORKERS = 5。为什么不设 20 或 50？

• 安居客对同一 IP 的并发请求数敏感，5 个并发在加设随机间隔后基本不会触发滑块。
• ThreadPoolExecutor没有内置的重试机制，并发过大时单个请求超时会被系统判为整体失败，可移植性差。
• 配合 3 秒随机延迟时，5 个 worker 的吞吐已经能跑到每分钟 100 个详情页，对绝大多数数据分析项目足够。

as_completed(futures)的作用是：哪个请求先完成就先取哪个结果，不要求提交顺序与返回顺序一致。拿到result后立即落库，避免把结果都堆在内存里。

4.3 Scrapy 的并发模型：更适合长期维护的项目

如果压缩包代码用的是 Scrapy，那它的并发模型由settings.py中的参数控制，逻辑上比手写ThreadPoolExecutor更成熟：

# settings.py CONCURRENT_REQUESTS = 8 CONCURRENT_REQUESTS_PER_DOMAIN = 4 DOWNLOAD_DELAY = 1.5 RANDOMIZE_DOWNLOAD_DELAY = True

• CONCURRENT_REQUESTS_PER_DOMAIN = 4是限制同一域名下的并发数，防止被站点封禁。
• DOWNLOAD_DELAY = 1.5表示每次请求之间至少间隔 1.5 秒，而RANDOMIZE_DOWNLOAD_DELAY = True会在这个基础上增加 0.5～1.5 秒的随机波动，这种抖动模式更适合规避请求频率画像。

Scrapy 相比手写线程池的优势在于自带重试中间件和去重过滤器。把RETRY_TIMES设为 3，RETRY_HTTP_CODES包含403、429、500，遇到这些状态码会自动重试，而手写代码需要自己实现while重试循环。

两者怎么选？我的建议是：压缩包里的代码如果本身基于requests手写，就不值得迁到 Scrapy；如果压缩包本身就是 Scrapy 项目，也建议保留，因为 Scrapy 的Item Pipeline天然适合做数据清洗和 DB 写入。

5. 突破验证码限制与分布式扩展：生产环境的稳定性方案

5.1 验证码出现的三种诱因与常规解

跑了一段时间后，代码会偶发拿到“验证码”页面而不是数据。这不是随机事件，通常有三种诱因：

• 频率过快：单个 IP 在 1 秒内发起多个请求。
• 缺少浏览器指纹：User-Agent固定而且不带Accept-Language。
• 行为轨迹异常：每次访问间隔恒定，或访问路径无规律。

针对第一种，把请求间隔随机化：

import random import time time.sleep(random.uniform(2.5, 5.5))

针对第二种，在requests.Session上设置完整的请求头，而不是每次都新建headers字典：

session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Connection": "keep-alive", })

设置 Cookie 时注意：把浏览器登录后的完整 Cookie 字符串传进去，不要只传sid。缺少任一关键字段都会导致服务端认为你是新访客，从而提升验证概率。我用这种方案在局域网环境做过验证，控制好频率后，连续抓取 2200 个详情页只触发过两次滑块。

提示：滑块出现后不要硬扛，记录当前页码，更换代理或直接停止任务，等 30 分钟后再从断点继续。代码里预留start_page参数即可。

5.2 代理池设计：不用付费代理也能跑通的方案

压缩包里如果自带代理 IP 列表，那么大概率是从免费代理网站抓的，可用率低于 20%。免费代理在请求返回前无法验证，所以必须建立“拉取 → 验证 → 使用 → 报废”的闭环。

import requests PROXY_POOL = [] def refresh_proxy_pool(min_valid: int = 3) -> None: proxy_source = requests.get("https://your_proxy_source.com/api/list", timeout=5).json() PROXY_POOL.clear() for item in proxy_source["data"]: test_url = "https://www.anjuke.com" proxy = {"http": f"http://{item['ip']}:{item['port']}", "https": f"http://{item['ip']}:{item['port']}"} try: resp = requests.get(test_url, proxies=proxy, timeout=5) if resp.status_code == 200: PROXY_POOL.append(proxy) if len(PROXY_POOL) >= min_valid: break except Exception: continue

这里的min_valid = 3是经验值，保持代理池在 3～5 个可用 IP 之间周转。每个代理的请求次数上限为 80 次，超过就回收：

proxy_usage = {} def get_next_proxy() -> dict: for proxy in PROXY_POOL: if proxy_usage.get(str(proxy), 0) < 80: proxy_usage[str(proxy)] = proxy_usage.get(str(proxy), 0) + 1 return proxy else: PROXY_POOL.remove(proxy) refresh_proxy_pool() return get_next_proxy()

注意这个循环在代理池耗尽时会递归调用，如果代理源失效会无限递归。建议加递归深度上限，或者改用while循环。

5.3 分布式爬虫：什么时候值得上，什么时候不值得

“分布式爬虫”这个热词经常出现在搜索联想里，但拿安居客这个场景来说，单机并发 8 线程配合代理池，一天能抓完一个城市的基础房源字段。只有当你同时需要 30 个城市、每天增量更新时，才需要考虑分布式。

Scrapy-Redis 是实现这个目标的典型方案：

# settings.py SCHEDULER = "scrapy_redis.scheduler.Scheduler" DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter" REDIS_URL = "redis://127.0.0.1:6379/0"

• SCHEDULER 替换为 Scrapy-Redis 的调度器，让所有爬虫节点共享同一个 Redis 队列。
• DUPEFILTER_CLASS 替换为 Redis 去重过滤器，保证同一house_id不会被两个节点重复抓取。
• REDIS_URL指向你的 Redis 实例，端口 6379 是默认值。

这套配置的本质不是把代码改得有多高级，而是把“任务队列”和“去重集合”从单机内存搬到 Redis，让多台机器消费同一批 URL。每台机器只需要有相同的代码和settings.py，不需要额外修改爬虫逻辑。

注意：使用分布式前，先把单机抓取时请求频率调到最大可接受水平，如果单机的瓶颈在反爬而不是 CPU，那分布式不会带来提升。

6. 数据存储与增量更新的落地技巧

6.1 用 SQLite 做轻量去重，避免重复采集

压缩包里通常会把结果写成 CSV，但 CSV 文件在重跑时会累积重复行。最省事的改进是改用 SQLite，用house_id做唯一索引：

python -c "import sqlite3; conn=sqlite3.connect('anjuke.db'); conn.execute('CREATE TABLE IF NOT EXISTS house (house_id TEXT PRIMARY KEY, title TEXT, price INTEGER, area REAL, city TEXT, fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP)'); conn.commit()"

写入时用INSERT OR IGNORE或INSERT OR REPLACE控制行为：

INSERT OR IGNORE INTO house (house_id, title, price, area, city) VALUES (?, ?, ?, ?, ?)

• INSERT OR IGNORE：已存在则跳过，适合全量抓取场景。
• INSERT OR REPLACE：已存在则更新，fetched_at会被刷新，适合定期增量。

6.2 小步验证：先抓 3 条再跑全量，这是最重要的习惯

这里可以给一个很实用的阶段划分方式，特别适合你刚拿到一个陌生的 zip 时使用。

第一步，先跑通列表页请求，打印出前 3 条数据。检查house_id是否为纯数字，URL 是否能打开。第二步，进入详情页抓取测试，打印标题与户型字段。输出与浏览器实际显示不一致就先修选择器。第三步，开启数据库表并抓 20 条，观察耗时和字段缺失率。最后再打开并发设置和代理池。

6.3 用sanitize函数处理脏字段，防止入库报错

压缩包代码中常见的pandas入库报错，通常是price或area字段里混入了"万"、"平米"等中文单位。入库前做一次清洗：

def sanitize_price(raw: str | None) -> int | None: if not raw: return None raw = raw.replace("万", "").replace("元/平", "").strip() try: return int(float(raw) * 10000) if "总价" in raw else int(float(raw)) except ValueError: return None def sanitize_area(raw: str | None) -> float | None: if not raw: return None raw = raw.replace("平米", "").replace("㎡", "").strip() try: return float(raw) except ValueError: return None

这里sanitize_price的返回逻辑比较粗暴，因为安居客不同城市的列表页价格单位不一样。建议解析前先打印原始字段值，统一确认单位——这是压缩包作者通常没有替你完成的部分。

6.4 数据可视化前的最后一步：计算单位面积均价

拿到price和area两个干净字段后，最常用的分析是计算单价。这一步放在 SQL 里完成，比在 Python 里循环快得多：

SELECT city, ROUND(AVG(price * 10000 / area)) AS avg_unit_price FROM house WHERE price IS NOT NULL AND area > 0 GROUP BY city ORDER BY avg_unit_price DESC;

配合pandas读出来绘制图表之前，记得过滤掉area < 10的数据，这类异常值通常是车位或储藏室，会把均价曲线拉出明显噪点。

本文还有配套的精品资源，点击获取