简介安居客Python爬虫代码是一个面向Python爬虫学习者和房产数据研究者的示例项目主要用于获取安居客平台经纪人信息手机号、店铺名称、公司名称、营业执照编号等通过解析二维码图片跳转详情页完成数据提取。资源共4个文件包含1个核心爬虫脚本AJK_Phone.py、2个全国城市请求地址表CSVutf-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(texthtml) 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 fhttps://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/xhtmlxml, Accept-Language: zh-CN,zh;q0.9, Referer: fhttps://sh.anjuke.com/sale/p1/, } resp requests.get(url, headersheaders, cookiescookies, timeout10) 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(rwindow\.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, headersheaders) with open(debug_anjuke.html, w, encodingutf-8) as f: f.write(resp.text)然后将debug_anjuke.html拖进浏览器用CtrlF搜索你关心的字段文本例如“3室2厅”。观察这段文本所在的 DOM 层级重新写选择器然后用parsel快速验证python -c from parsel import Selector; sSelector(textopen(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( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, filenameanjuke_spider.log, filemodea, ) 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指定日志输出到文件避免终端日志被系统清理。levelINFO让requests和urllib3的底层日志不会刷屏需要排查网络握手细节时才改成DEBUG。4. 爬虫并发设计从单线程到可控协程到底哪个好4.1 单线程是压缩包默认方案但速度不可接受传统的for循环逐个请求列表页和详情页请求之间间隔 12 秒跑完 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_workersMAX_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 TrueCONCURRENT_REQUESTS_PER_DOMAIN 4是限制同一域名下的并发数防止被站点封禁。DOWNLOAD_DELAY 1.5表示每次请求之间至少间隔 1.5 秒而RANDOMIZE_DOWNLOAD_DELAY True会在这个基础上增加 0.51.5 秒的随机波动这种抖动模式更适合规避请求频率画像。Scrapy 相比手写线程池的优势在于自带重试中间件和去重过滤器。把RETRY_TIMES设为 3RETRY_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;q0.9,en;q0.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, timeout5).json() PROXY_POOL.clear() for item in proxy_source[data]: test_url https://www.anjuke.com proxy {http: fhttp://{item[ip]}:{item[port]}, https: fhttp://{item[ip]}:{item[port]}} try: resp requests.get(test_url, proxiesproxy, timeout5) if resp.status_code 200: PROXY_POOL.append(proxy) if len(PROXY_POOL) min_valid: break except Exception: continue这里的min_valid 3是经验值保持代理池在 35 个可用 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/0SCHEDULER 替换为 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; connsqlite3.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的数据这类异常值通常是车位或储藏室会把均价曲线拉出明显噪点。本文还有配套的精品资源点击获取