简介基于Python Scrapy框架实现的长沙链家二手房信息爬虫设计源码面向Python爬虫学习者、数据分析师及房地产研究人员用于批量采集长沙区域链家挂牌房源解决手动收集二手房信息效率低下的问题。压缩包共22个文件以8个Python源代码文件为核心覆盖爬虫主模块、管道处理、中间件、设置项等关键部分另有7个编译文件、4个XML配置、1个TXT说明文档、1个cfg配置文件及1个iml工程文件整体仅29KB结构清晰紧凑。目前已有465人学习适合想系统掌握Scrapy爬虫工程结构、提升数据采集实战能力的读者。运行该爬虫可自动获取房源标题、价格、地址、户型、面积、建造年代等结构化字段并直接保存为本地CSV方便使用Excel或Pandas完成数据清洗、价格区间统计与区域趋势分析。此外项目提供了完整的配置与说明便于二次开发和功能扩展作为课设、毕设或行业数据分析的参考项目都比较合适。1. 用Scrapy重新理解长沙链家二手房爬虫的设计想跟踪长沙某个片区的二手房挂牌价手动翻链家是最低效的方式可自己写爬虫时又容易踩进同一个坑列表页抓下来了但价格单位和翻页规则没处理好第二天页面一改版整个脚本报废。基于Python Scrapy框架的长沙链家二手房信息爬虫设计源码如果只看“源码”两个字很容易理解成一份能跑的脚本但真正值得研究的是Scrapy各项组件在区域站点采集场景下如何分工。Item负责固定字段Loader负责清洗Spider负责链接生成与解析Middleware负责频率控制Pipeline负责去重落库。这个结构同样可以平移到其他城市、其他分类信息站点。下面按数据模型、Spider、中间件、入库这条主线展开每一步都给可直接复用的代码。2. 从页面结构到数据模型长沙链家二手房Item与字段设计2.1 先拆页面再定义字段写Spider之前我会先做一次页面拆解。打开长沙链家二手房列表页能看到挂牌标题、小区名、商圈、户型、面积、朝向、装修、楼层、总价、单价等字段点进详情页还会出现房屋年限、产权、挂牌时间等额外信息。字段不是越多越好但统计要用到的一定不能漏。以常见的跟盘需求为例Item可以定义成下面这样。import scrapy class LianjiaHouseItem(scrapy.Item): house_id scrapy.Field() # 房源唯一编号来自详情链接 district scrapy.Field() # 行政区如 yueluqu、furongqu biz_circle scrapy.Field() # 商圈如 meixihu、binjiang community scrapy.Field() # 小区名 title scrapy.Field() # 挂牌标题 rooms scrapy.Field() # 户型原始文本如 3室2厅 area scrapy.Field() # 建筑面积float单位平米 orientation scrapy.Field() # 朝向如 南 北 decoration scrapy.Field() # 装修情况如 精装 floor scrapy.Field() # 楼层与总层数如 高楼层/共33层 total_price scrapy.Field() # 总价float单位万元 unit_price scrapy.Field() # 单价int单位元/平米 listing_date scrapy.Field() # 挂牌日期统一成 YYYY-MM-DD crawl_time scrapy.Field() # 本次抓取时间ISO 格式字符串这里有两个设计点要说明。第一是house_id不是页面直接展示的值而是从详情链接里提取的比如https://cs.lianjia.com/ershoufang/108456789.html去掉.html后得到108456789这就是稳定主键。后面的去重、增量更新、历史价格对比都依赖它所以不要把它和其他字段混在一起。第二是crawl_time应该放在Item里生成而不是等Pipeline入库时补时间这样重试下载或者消息队列延迟消费时记录下来的仍是真实抓取时刻。2.2 字段类型与链家页面样貌的映射陷阱字段定义好后还要处理页面文本和存储类型之间的差异。链家页面上显示的是给人看的内容直接存进数据库会带来大量脏数据。常见的映射关系我整理成了表格。字段页面样貌清洗操作建议存储类型total_price298万去掉“万”转floatREALunit_price11356元/平只提取数字INTarea117.6平米 / 117.6㎡去掉单位与空格REALfloor高楼层/共33层保留原文本TEXTlisting_date2024-03-18统一为日期文本TEXTrooms3室2厅保留原文本统计时再拆TEXT最大的坑是价格单位不统一总价以“万”为单位单价以“元/平”为单位清洗函数必须分开写。面积大概率要兼容“平米”和“㎡”两种写法我在实际项目中甚至遇到过“147.6平米”和“建面147.6㎡”混在同一个列表里。楼层字段建议保留原始字符串不要在这里拆成“高楼层”和“33层”两个值因为后续分析时拆分规则可能变化原始信息一旦丢就很难补回来。另外链家房源标签里的“满五”“满二”“唯一”属于状态信息如果要统计税费成本应该单独用tags字段保存。2.3 用ItemLoader统一清洗逻辑清洗逻辑如果散落在Spider的各个回调里同一个字段列表页要写一次详情页又要写一次改版时漏改一个就出问题。Scrapy提供ItemLoader可以把清洗函数集中声明。import re from scrapy.loader import ItemLoader from scrapy.loader.processors import MapCompose def clean_total_price(value): if value is None: return value return float(str(value).replace(万, ).strip()) def clean_unit_price(value): if value is None: return value nums re.findall(r\d, str(value)) return int(nums[0]) if nums else None def clean_area(value): if value is None: return value return float(re.sub(r[^0-9.], , str(value))) class HouseLoader(ItemLoader): default_item_class LianjiaHouseItem total_price_in MapCompose(clean_total_price) unit_price_in MapCompose(clean_unit_price) area_in MapCompose(clean_area)_in后缀表示这是输入处理器字段被写入Item之前逐值执行清洗_out后缀则是在输出最终Item时才执行。我用_in是希望清洗动作在进入Pipeline前完成后面Pipeline只做范围校验和落库。MapCompose会把一个列表中的所有值依次传入清洗函数所以每个清洗函数都要能容忍None和字符串并且重复执行结果一致。这样在Spider里写loader.add_css(total_price, span.totalPrice::text)时取出来就已经是float类型。3. 构建长沙链家二手房Spider翻页规则、列表解析与详情补全3.1 用start_requests参数化长沙六大区长沙链家二手房列表按行政区入口划分比如岳麓区和芙蓉区的列表路径不一样。如果每次都从城市总入口抓再靠字段里判断区域效率低且容易丢数据。更常见的做法是在start_requests里为每个行政区生成初始请求并把区域名通过cb_kwargs传给后续回调。import scrapy class CsLianjiaSpider(scrapy.Spider): name cs_lianjia_spider base_url https://cs.lianjia.com/ershoufang def start_requests(self): districts [yueluqu, furongqu, tianxinqu, kaifuqu, yuhuaqu, wangchengqu] for district in districts: yield scrapy.Request( urlf{self.base_url}/{district}/, callbackself.parse_list, cb_kwargs{district: district}, )用yield而不是直接声明start_urls是因为我想给每个请求附带区域参数。cb_kwargs里的内容会作为关键字参数传给回调函数这样parse_list再生成详情请求时也能一路把district和house_id带到parse_detail最终所有item都能追溯到来源区域。这个参数传递方式在Scrapy 2.x中是推荐的写法比在meta里塞字典更直观也减少漏传参数的问题。3.2 列表页卡片解析与翻页终止条件列表页的卡片结构相对稳定房源卡片通常在ul.sellListContent li.clear下标题链接是.title a。这里可以先用列表页字段组装一部分Item再把详情页请求发给parse_detail补全其他字段。import json from scrapy.http import Request def parse_list(self, response, district): for card in response.css(ul.sellListContent li.clear): detail_url card.css(.title a::attr(href)).get() if not detail_url: continue house_id detail_url.split(/)[-1].replace(.html, ) yield Request( urldetail_url, callbackself.parse_detail, cb_kwargs{district: district, house_id: house_id}, ) page_box response.css(.house-lst-page-box) if not page_box: return page_data json.loads(page_box.attrib.get(page-data, {})) cur_page int(page_data.get(curPage, 1)) total_page int(page_data.get(totalPage, 1)) if cur_page total_page: next_url f{self.base_url}/{district}/pg{cur_page 1}/ yield Request( urlnext_url, callbackself.parse_list, cb_kwargs{district: district}, )这段逻辑里需要注意两点。第一不要直接写死“下一页”按钮的选择器链家列表翻页状态实际放在.house-lst-page-box的page-data属性里从中读取curPage和totalPage更可靠。第二链家列表最多显示100页超过100页后要继续抓必须在区域下再按商圈拆分URL形如/ershoufang/{district}/{bizcircle}/。这也是为什么Item里要单独留biz_circle字段而不是用district替代。3.3 详情页解析补全列表页缺失的字段列表页能拿到价格、面积、户型而房屋年限、权属、挂牌时间这些信息要进详情页。详情页解析可以直接复用前面定义的HouseLoader保持清洗逻辑唯一。def parse_detail(self, response, district, house_id): loader HouseLoader(itemLianjiaHouseItem(), selectorresponse) loader.add_value(house_id, house_id) loader.add_value(district, district) loader.add_css(community, .communityName a::text) loader.add_css(rooms, .room .mainInfo::text) loader.add_css(area, .area .mainInfo::text) loader.add_css(orientation, .type .subInfo::text) loader.add_css(total_price, .total::text) loader.add_css(unit_price, .unitPriceValue::text) loader.add_css(listing_date, .subInfo::text) yield loader.load_item()使用selectorresponse后add_css里面写的是完整文档上下文的选择器因此.room .mainInfo这种组合写法不能省略否则rooms和area会同时命中多个页面元素把文本混在一起。listing_date也一样.subInfo在详情页里有多处必须用更具体的选择器定位。如果遇到抓取后某些字段为空优先用Scrapy Shell逐步验证选择器而不是急着换解析方案。另外提一个边界情况链家部分详情页存在动态加载内容或者页面被包在iframe里这时Scrapy原生请求拿不到完整HTML。常见做法是给Spider挂scrapy-playwright用无头浏览器渲染后再交给同一个解析函数。但浏览器渲染会明显拉低吞吐量性价比不高建议先确认普通请求里是否已经包含目标字段再决定要不要引入动态渲染。4. 防爬与稳定性自定义中间件、频率控制与请求重试4.1 链家常见的反爬触发点与合规应对链家对于频繁访问的应对主要是两类高频请求直接返回验证页或者短时间内数据异常导致状态码变成403/429。应对思路不是去绕验证码而是把请求节奏降到正常用户水平并让请求头更像真实浏览器。另一个原则是抓取范围控制在公开列表和详情页单次任务时间不宜拖得过长跑完就停不要无限循环。具体落地时我通常在settings里做几件基础配置开启AutoThrottle、设置下载延迟、限制单域名并发数再配一组真实UA做轮换。这些配置的推荐值如下。配置项推荐值说明DOWNLOAD_DELAY1.2同一个域名下两次请求的最小间隔单位秒RANDOMIZE_DOWNLOAD_DELAYTrue在下载延迟上叠加随机波动CONCURRENT_REQUESTS_PER_DOMAIN8每个域名最大并发请求数AUTOTHROTTLE_ENABLEDTrue根据服务器响应速度动态调整频率RETRY_HTTP_CODES[403, 429, 500, 502, 503]遇到这些状态码自动重试4.2 自定义DownloaderMiddleware随机UA与请求节流Scrapy默认的UA是Scrapy/x.x非常容易被识别。自定义一个下载中间件每次请求前从配置列表里随机选一个UA写进请求头。import random class RandomUserAgentMiddleware: def __init__(self, user_agents): self.user_agents user_agents classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist(USER_AGENTS)) def process_request(self, request, spider): request.headers.setdefault(bUser-Agent, random.choice(self.user_agents))这个中间件通过from_crawler从settings读取USER_AGENTS列表所以需要在settings里维护一个真实的UA列表比如Chrome和Edge各几个版本。用setdefault而不是直接赋值是为了保留某些请求里已经由上层逻辑设定的UA。真实浏览器UA可以从自己电脑浏览器的开发者工具里复制不需要编造。注意UA只是让请求看起来正常核心还是频率要慢下来UA轮换和下载延迟需要一起用。4.3 多出口IP池的接入方式单个IP抓上万条数据即使延迟调到1.5秒也可能因为总量过大触发限制。比较稳妥的做法是接入多出口IP资源在请求层设置代理。class ProxyPoolMiddleware: def process_request(self, request, spider): proxy spider.proxy_manager.get() if proxy: request.meta[proxy] proxy def process_response(self, request, response, spider): if response.status in (403, 429): spider.proxy_manager.remove(request.meta.get(proxy)) return request.replace(dont_filterTrue) return response这里的proxy_manager需要自己在Spider里实现例如从一组可用IP中轮询分配并在请求失败后把坏IP剔除。IP来源可以是公司出口资源也可以是商业代理服务商提供的合规代理不建议去抓免费代理稳定性差且容易引入风险。如果团队暂时没有代理资源用DOWNLOAD_DELAY加AUTOTHROTTLE控制节奏也能跑只是总耗时更长。4.4 重试策略与验证页特征识别Scrapy自带的RetryMiddleware已经够用没必要自己重新实现完整重试逻辑。在settings里声明即可。RETRY_ENABLED True RETRY_TIMES 3 RETRY_HTTP_CODES [403, 429, 500, 502, 503]但有些反爬页面返回的状态码是200页面内容却是验证页这种状态码层面的重试不会被触发。我的处理方式是写一个中间件识别页面上的异常特征一旦命中就记录并抛出一个特殊异常让该请求在后续重试中降频。class BlockedPageMiddleware: def process_response(self, request, response, spider): if response.status 200 and verify in response.url: spider.crawler.stats.inc_value(blocked/verify_page) return request.replace(dont_filterTrue) return response这段代码只做识别和重试不会去处理验证页面本身。看到这种页面出现频率升高应该停下来检查请求频率而不是继续加大并发。日志里加上blocked/verify_page这个统计项后可以在每次任务结束时快速判断爬虫是否健康。5. 数据入库与增量更新Pipeline让采集结果可追溯5.1 Pipeline的职责边界在Scrapy里ItemLoader已经完成了字段清洗Pipeline就不应该再重复清洗它的核心职责是校验、去重、落库。对于区域房源数据我更倾向于在Pipeline里做两件事丢弃明显无效的数据并把合法数据写入持久化存储。丢弃的数据要在Stats里计数方便任务结束后核对。5.2 用SQLite Pipeline在本地快速落库本地开发阶段用SQLite最方便不需要额外启动数据库服务。Pipeline里维护一张house表以house_id为主键用INSERT OR REPLACE实现覆盖更新。import sqlite3 from scrapy.exceptions import DropItem class SQLitePipeline: def __init__(self, db_path): self.db_path db_path self.conn None classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get(SQLITE_DB_PATH)) def open_spider(self, spider): self.conn sqlite3.connect(self.db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS house ( house_id TEXT PRIMARY KEY, district TEXT, community TEXT, rooms TEXT, area REAL, total_price REAL, unit_price INT, floor TEXT, crawl_time TEXT ) ) def process_item(self, item, spider): if not item.get(total_price) or item[total_price] 0: spider.crawler.stats.inc_value(item/dropped/empty_price) raise DropItem(empty total_price) self.conn.execute( INSERT OR REPLACE INTO house (house_id, district, community, rooms, area, total_price, unit_price, floor, crawl_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( item[house_id], item.get(district), item.get(community), item.get(rooms), item.get(area), item[total_price], item.get(unit_price), item.get(floor), item.get(crawl_time), )) self.conn.commit() return item def close_spider(self, spider): self.conn.close()from_crawler从settings读取数据库路径避免在代码里写死。open_spider在爬虫启动时创建数据库和表。INSERT OR REPLACE的意思是如果house_id已经存在就直接用新数据覆盖旧数据。这对抓取当前挂牌价是合理的但如果要做价格历史分析就不能只保留当前值必须同时记录每次变化。5.3 保存价格历史用审计表替代过度覆盖实际做市场跟踪时真正有价值的不只是最新价格而是价格随时间变化的曲线。常见做法是把当前快照写入house主表同时把每次抓取的结果追加到house_price_history表。CREATE TABLE IF NOT EXISTS house_price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, house_id TEXT NOT NULL, total_price REAL, unit_price INT, crawl_time TEXT, UNIQUE (house_id, total_price, crawl_time) );这张表每次写入的是一条历史记录不删旧数据。后续想看某个小区均价走势直接按house_id查历史表即可。如果某个房源在两次抓取之间价格没有变化通过UNIQUE(house_id, total_price, crawl_time)可以避免重复记录。要提醒的是入库操作本身比解析更耗时Pipeline处理速度要控制在Spider产出速度以上否则会导致内存积压实测中发现SQLite逐条commit在小数据量下没问题但超过5万条后建议改成批量提交每100条commit一次。5.4 扩容到MySQL时的改造点项目进入生产环境后SQLite的并发能力和运维监控都偏弱我会切换到MySQL。迁移成本主要在Pipeline的连接方式以及表结构里的字符集指定。MySQL表结构需要显式指定utf8mb4否则小区名里生僻字可能写入失败。其余字段设计可以保持一致这样Spider不需要改动只是替换Pipeline即可。6. 落盘后的验证与优化用Scrapy扩展核验数据质量6.1 用Scrapy Shell验证选择器正确性改动解析代码后不要直接跑步整个爬虫先用Scrapy Shell验证单页选择器。这是排查解析问题最有效的手段。scrapy shell https://cs.lianjia.com/ershoufang/yueluqu/进入Shell后可以用response.css(ul.sellListContent li.clear)确认列表卡片数量再用response.css(.title a::attr(href)).get()确认详情链接能取到。Shell里能正确取到结果再把选择器复制进Spider。这样可以避免每次改代码都要跑全量也能更快区分是页面结构变化还是代码逻辑问题。6.2 用Stats统计解析成功率Scrapy的Stats在默认情况下会统计请求数、item产出数、异常数。在Spider里可以直观看到一次任务后的关键指标scrapy crawl cs_lianjia_spider -s CLOSESPIDER_ITEMCOUNT10CLOSESPIDER_ITEMCOUNT是Scrapy对爬虫收集到指定条数后自动停止的扩展适合快速验证小批量链路。任务结束后把item_scraped_count和downloader/request_count做除法就能得到粗略的解析成功率。正常情况下这个比例应该在90%以上如果明显偏低大概率是列表页解析部分失效而不是详情页的问题。为了让这个指标更容易观察我会在spider_closed信号里主动打印。def close(spider, reason): stats spider.crawler.stats scraped stats.get_value(item_scraped_count, 0) requests stats.get_value(downloader/request_count, 0) spider.logger.warning( parse_rate%.2f scraped%d requests%d, scraped / requests if requests else 0, scraped, requests )6.3 定时任务与运行状态检查数据链路稳定后把爬虫挂到定时任务里。Linux上最直接的是crontab但要正确指定Python解释器和项目目录。0 9 * * * cd /opt/house-spider /usr/bin/python3 -m scrapy crawl cs_lianjia_spider logs/crawl.log 21这段命令每天早上九点进入项目目录调用系统Python直接执行Scrapy模块。这里的关键是不要用python而要用/usr/bin/python3确保crontab PATH里找不到命令时也能正常运行。爬虫跑完后配合crawler.stats里的指标做简单告警如果item_scraped_count低于预设阈值就把日志推送到告警群表示页面可能改版。这套结构后续要改成分布式爬虫也很直接把Spider的请求队列接到Scrapy-Redis上即可。有条件的话把每日抓取和报价分析串成一条命令抓完数据后立即生成行情日报整个流程就不需要人工盯守了。本文还有配套的精品资源点击获取 Scrapy爬虫实战：长沙链家二手房数据采集与清洗
简介基于Python Scrapy框架实现的长沙链家二手房信息爬虫设计源码面向Python爬虫学习者、数据分析师及房地产研究人员用于批量采集长沙区域链家挂牌房源解决手动收集二手房信息效率低下的问题。压缩包共22个文件以8个Python源代码文件为核心覆盖爬虫主模块、管道处理、中间件、设置项等关键部分另有7个编译文件、4个XML配置、1个TXT说明文档、1个cfg配置文件及1个iml工程文件整体仅29KB结构清晰紧凑。目前已有465人学习适合想系统掌握Scrapy爬虫工程结构、提升数据采集实战能力的读者。运行该爬虫可自动获取房源标题、价格、地址、户型、面积、建造年代等结构化字段并直接保存为本地CSV方便使用Excel或Pandas完成数据清洗、价格区间统计与区域趋势分析。此外项目提供了完整的配置与说明便于二次开发和功能扩展作为课设、毕设或行业数据分析的参考项目都比较合适。1. 用Scrapy重新理解长沙链家二手房爬虫的设计想跟踪长沙某个片区的二手房挂牌价手动翻链家是最低效的方式可自己写爬虫时又容易踩进同一个坑列表页抓下来了但价格单位和翻页规则没处理好第二天页面一改版整个脚本报废。基于Python Scrapy框架的长沙链家二手房信息爬虫设计源码如果只看“源码”两个字很容易理解成一份能跑的脚本但真正值得研究的是Scrapy各项组件在区域站点采集场景下如何分工。Item负责固定字段Loader负责清洗Spider负责链接生成与解析Middleware负责频率控制Pipeline负责去重落库。这个结构同样可以平移到其他城市、其他分类信息站点。下面按数据模型、Spider、中间件、入库这条主线展开每一步都给可直接复用的代码。2. 从页面结构到数据模型长沙链家二手房Item与字段设计2.1 先拆页面再定义字段写Spider之前我会先做一次页面拆解。打开长沙链家二手房列表页能看到挂牌标题、小区名、商圈、户型、面积、朝向、装修、楼层、总价、单价等字段点进详情页还会出现房屋年限、产权、挂牌时间等额外信息。字段不是越多越好但统计要用到的一定不能漏。以常见的跟盘需求为例Item可以定义成下面这样。import scrapy class LianjiaHouseItem(scrapy.Item): house_id scrapy.Field() # 房源唯一编号来自详情链接 district scrapy.Field() # 行政区如 yueluqu、furongqu biz_circle scrapy.Field() # 商圈如 meixihu、binjiang community scrapy.Field() # 小区名 title scrapy.Field() # 挂牌标题 rooms scrapy.Field() # 户型原始文本如 3室2厅 area scrapy.Field() # 建筑面积float单位平米 orientation scrapy.Field() # 朝向如 南 北 decoration scrapy.Field() # 装修情况如 精装 floor scrapy.Field() # 楼层与总层数如 高楼层/共33层 total_price scrapy.Field() # 总价float单位万元 unit_price scrapy.Field() # 单价int单位元/平米 listing_date scrapy.Field() # 挂牌日期统一成 YYYY-MM-DD crawl_time scrapy.Field() # 本次抓取时间ISO 格式字符串这里有两个设计点要说明。第一是house_id不是页面直接展示的值而是从详情链接里提取的比如https://cs.lianjia.com/ershoufang/108456789.html去掉.html后得到108456789这就是稳定主键。后面的去重、增量更新、历史价格对比都依赖它所以不要把它和其他字段混在一起。第二是crawl_time应该放在Item里生成而不是等Pipeline入库时补时间这样重试下载或者消息队列延迟消费时记录下来的仍是真实抓取时刻。2.2 字段类型与链家页面样貌的映射陷阱字段定义好后还要处理页面文本和存储类型之间的差异。链家页面上显示的是给人看的内容直接存进数据库会带来大量脏数据。常见的映射关系我整理成了表格。字段页面样貌清洗操作建议存储类型total_price298万去掉“万”转floatREALunit_price11356元/平只提取数字INTarea117.6平米 / 117.6㎡去掉单位与空格REALfloor高楼层/共33层保留原文本TEXTlisting_date2024-03-18统一为日期文本TEXTrooms3室2厅保留原文本统计时再拆TEXT最大的坑是价格单位不统一总价以“万”为单位单价以“元/平”为单位清洗函数必须分开写。面积大概率要兼容“平米”和“㎡”两种写法我在实际项目中甚至遇到过“147.6平米”和“建面147.6㎡”混在同一个列表里。楼层字段建议保留原始字符串不要在这里拆成“高楼层”和“33层”两个值因为后续分析时拆分规则可能变化原始信息一旦丢就很难补回来。另外链家房源标签里的“满五”“满二”“唯一”属于状态信息如果要统计税费成本应该单独用tags字段保存。2.3 用ItemLoader统一清洗逻辑清洗逻辑如果散落在Spider的各个回调里同一个字段列表页要写一次详情页又要写一次改版时漏改一个就出问题。Scrapy提供ItemLoader可以把清洗函数集中声明。import re from scrapy.loader import ItemLoader from scrapy.loader.processors import MapCompose def clean_total_price(value): if value is None: return value return float(str(value).replace(万, ).strip()) def clean_unit_price(value): if value is None: return value nums re.findall(r\d, str(value)) return int(nums[0]) if nums else None def clean_area(value): if value is None: return value return float(re.sub(r[^0-9.], , str(value))) class HouseLoader(ItemLoader): default_item_class LianjiaHouseItem total_price_in MapCompose(clean_total_price) unit_price_in MapCompose(clean_unit_price) area_in MapCompose(clean_area)_in后缀表示这是输入处理器字段被写入Item之前逐值执行清洗_out后缀则是在输出最终Item时才执行。我用_in是希望清洗动作在进入Pipeline前完成后面Pipeline只做范围校验和落库。MapCompose会把一个列表中的所有值依次传入清洗函数所以每个清洗函数都要能容忍None和字符串并且重复执行结果一致。这样在Spider里写loader.add_css(total_price, span.totalPrice::text)时取出来就已经是float类型。3. 构建长沙链家二手房Spider翻页规则、列表解析与详情补全3.1 用start_requests参数化长沙六大区长沙链家二手房列表按行政区入口划分比如岳麓区和芙蓉区的列表路径不一样。如果每次都从城市总入口抓再靠字段里判断区域效率低且容易丢数据。更常见的做法是在start_requests里为每个行政区生成初始请求并把区域名通过cb_kwargs传给后续回调。import scrapy class CsLianjiaSpider(scrapy.Spider): name cs_lianjia_spider base_url https://cs.lianjia.com/ershoufang def start_requests(self): districts [yueluqu, furongqu, tianxinqu, kaifuqu, yuhuaqu, wangchengqu] for district in districts: yield scrapy.Request( urlf{self.base_url}/{district}/, callbackself.parse_list, cb_kwargs{district: district}, )用yield而不是直接声明start_urls是因为我想给每个请求附带区域参数。cb_kwargs里的内容会作为关键字参数传给回调函数这样parse_list再生成详情请求时也能一路把district和house_id带到parse_detail最终所有item都能追溯到来源区域。这个参数传递方式在Scrapy 2.x中是推荐的写法比在meta里塞字典更直观也减少漏传参数的问题。3.2 列表页卡片解析与翻页终止条件列表页的卡片结构相对稳定房源卡片通常在ul.sellListContent li.clear下标题链接是.title a。这里可以先用列表页字段组装一部分Item再把详情页请求发给parse_detail补全其他字段。import json from scrapy.http import Request def parse_list(self, response, district): for card in response.css(ul.sellListContent li.clear): detail_url card.css(.title a::attr(href)).get() if not detail_url: continue house_id detail_url.split(/)[-1].replace(.html, ) yield Request( urldetail_url, callbackself.parse_detail, cb_kwargs{district: district, house_id: house_id}, ) page_box response.css(.house-lst-page-box) if not page_box: return page_data json.loads(page_box.attrib.get(page-data, {})) cur_page int(page_data.get(curPage, 1)) total_page int(page_data.get(totalPage, 1)) if cur_page total_page: next_url f{self.base_url}/{district}/pg{cur_page 1}/ yield Request( urlnext_url, callbackself.parse_list, cb_kwargs{district: district}, )这段逻辑里需要注意两点。第一不要直接写死“下一页”按钮的选择器链家列表翻页状态实际放在.house-lst-page-box的page-data属性里从中读取curPage和totalPage更可靠。第二链家列表最多显示100页超过100页后要继续抓必须在区域下再按商圈拆分URL形如/ershoufang/{district}/{bizcircle}/。这也是为什么Item里要单独留biz_circle字段而不是用district替代。3.3 详情页解析补全列表页缺失的字段列表页能拿到价格、面积、户型而房屋年限、权属、挂牌时间这些信息要进详情页。详情页解析可以直接复用前面定义的HouseLoader保持清洗逻辑唯一。def parse_detail(self, response, district, house_id): loader HouseLoader(itemLianjiaHouseItem(), selectorresponse) loader.add_value(house_id, house_id) loader.add_value(district, district) loader.add_css(community, .communityName a::text) loader.add_css(rooms, .room .mainInfo::text) loader.add_css(area, .area .mainInfo::text) loader.add_css(orientation, .type .subInfo::text) loader.add_css(total_price, .total::text) loader.add_css(unit_price, .unitPriceValue::text) loader.add_css(listing_date, .subInfo::text) yield loader.load_item()使用selectorresponse后add_css里面写的是完整文档上下文的选择器因此.room .mainInfo这种组合写法不能省略否则rooms和area会同时命中多个页面元素把文本混在一起。listing_date也一样.subInfo在详情页里有多处必须用更具体的选择器定位。如果遇到抓取后某些字段为空优先用Scrapy Shell逐步验证选择器而不是急着换解析方案。另外提一个边界情况链家部分详情页存在动态加载内容或者页面被包在iframe里这时Scrapy原生请求拿不到完整HTML。常见做法是给Spider挂scrapy-playwright用无头浏览器渲染后再交给同一个解析函数。但浏览器渲染会明显拉低吞吐量性价比不高建议先确认普通请求里是否已经包含目标字段再决定要不要引入动态渲染。4. 防爬与稳定性自定义中间件、频率控制与请求重试4.1 链家常见的反爬触发点与合规应对链家对于频繁访问的应对主要是两类高频请求直接返回验证页或者短时间内数据异常导致状态码变成403/429。应对思路不是去绕验证码而是把请求节奏降到正常用户水平并让请求头更像真实浏览器。另一个原则是抓取范围控制在公开列表和详情页单次任务时间不宜拖得过长跑完就停不要无限循环。具体落地时我通常在settings里做几件基础配置开启AutoThrottle、设置下载延迟、限制单域名并发数再配一组真实UA做轮换。这些配置的推荐值如下。配置项推荐值说明DOWNLOAD_DELAY1.2同一个域名下两次请求的最小间隔单位秒RANDOMIZE_DOWNLOAD_DELAYTrue在下载延迟上叠加随机波动CONCURRENT_REQUESTS_PER_DOMAIN8每个域名最大并发请求数AUTOTHROTTLE_ENABLEDTrue根据服务器响应速度动态调整频率RETRY_HTTP_CODES[403, 429, 500, 502, 503]遇到这些状态码自动重试4.2 自定义DownloaderMiddleware随机UA与请求节流Scrapy默认的UA是Scrapy/x.x非常容易被识别。自定义一个下载中间件每次请求前从配置列表里随机选一个UA写进请求头。import random class RandomUserAgentMiddleware: def __init__(self, user_agents): self.user_agents user_agents classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist(USER_AGENTS)) def process_request(self, request, spider): request.headers.setdefault(bUser-Agent, random.choice(self.user_agents))这个中间件通过from_crawler从settings读取USER_AGENTS列表所以需要在settings里维护一个真实的UA列表比如Chrome和Edge各几个版本。用setdefault而不是直接赋值是为了保留某些请求里已经由上层逻辑设定的UA。真实浏览器UA可以从自己电脑浏览器的开发者工具里复制不需要编造。注意UA只是让请求看起来正常核心还是频率要慢下来UA轮换和下载延迟需要一起用。4.3 多出口IP池的接入方式单个IP抓上万条数据即使延迟调到1.5秒也可能因为总量过大触发限制。比较稳妥的做法是接入多出口IP资源在请求层设置代理。class ProxyPoolMiddleware: def process_request(self, request, spider): proxy spider.proxy_manager.get() if proxy: request.meta[proxy] proxy def process_response(self, request, response, spider): if response.status in (403, 429): spider.proxy_manager.remove(request.meta.get(proxy)) return request.replace(dont_filterTrue) return response这里的proxy_manager需要自己在Spider里实现例如从一组可用IP中轮询分配并在请求失败后把坏IP剔除。IP来源可以是公司出口资源也可以是商业代理服务商提供的合规代理不建议去抓免费代理稳定性差且容易引入风险。如果团队暂时没有代理资源用DOWNLOAD_DELAY加AUTOTHROTTLE控制节奏也能跑只是总耗时更长。4.4 重试策略与验证页特征识别Scrapy自带的RetryMiddleware已经够用没必要自己重新实现完整重试逻辑。在settings里声明即可。RETRY_ENABLED True RETRY_TIMES 3 RETRY_HTTP_CODES [403, 429, 500, 502, 503]但有些反爬页面返回的状态码是200页面内容却是验证页这种状态码层面的重试不会被触发。我的处理方式是写一个中间件识别页面上的异常特征一旦命中就记录并抛出一个特殊异常让该请求在后续重试中降频。class BlockedPageMiddleware: def process_response(self, request, response, spider): if response.status 200 and verify in response.url: spider.crawler.stats.inc_value(blocked/verify_page) return request.replace(dont_filterTrue) return response这段代码只做识别和重试不会去处理验证页面本身。看到这种页面出现频率升高应该停下来检查请求频率而不是继续加大并发。日志里加上blocked/verify_page这个统计项后可以在每次任务结束时快速判断爬虫是否健康。5. 数据入库与增量更新Pipeline让采集结果可追溯5.1 Pipeline的职责边界在Scrapy里ItemLoader已经完成了字段清洗Pipeline就不应该再重复清洗它的核心职责是校验、去重、落库。对于区域房源数据我更倾向于在Pipeline里做两件事丢弃明显无效的数据并把合法数据写入持久化存储。丢弃的数据要在Stats里计数方便任务结束后核对。5.2 用SQLite Pipeline在本地快速落库本地开发阶段用SQLite最方便不需要额外启动数据库服务。Pipeline里维护一张house表以house_id为主键用INSERT OR REPLACE实现覆盖更新。import sqlite3 from scrapy.exceptions import DropItem class SQLitePipeline: def __init__(self, db_path): self.db_path db_path self.conn None classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get(SQLITE_DB_PATH)) def open_spider(self, spider): self.conn sqlite3.connect(self.db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS house ( house_id TEXT PRIMARY KEY, district TEXT, community TEXT, rooms TEXT, area REAL, total_price REAL, unit_price INT, floor TEXT, crawl_time TEXT ) ) def process_item(self, item, spider): if not item.get(total_price) or item[total_price] 0: spider.crawler.stats.inc_value(item/dropped/empty_price) raise DropItem(empty total_price) self.conn.execute( INSERT OR REPLACE INTO house (house_id, district, community, rooms, area, total_price, unit_price, floor, crawl_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( item[house_id], item.get(district), item.get(community), item.get(rooms), item.get(area), item[total_price], item.get(unit_price), item.get(floor), item.get(crawl_time), )) self.conn.commit() return item def close_spider(self, spider): self.conn.close()from_crawler从settings读取数据库路径避免在代码里写死。open_spider在爬虫启动时创建数据库和表。INSERT OR REPLACE的意思是如果house_id已经存在就直接用新数据覆盖旧数据。这对抓取当前挂牌价是合理的但如果要做价格历史分析就不能只保留当前值必须同时记录每次变化。5.3 保存价格历史用审计表替代过度覆盖实际做市场跟踪时真正有价值的不只是最新价格而是价格随时间变化的曲线。常见做法是把当前快照写入house主表同时把每次抓取的结果追加到house_price_history表。CREATE TABLE IF NOT EXISTS house_price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, house_id TEXT NOT NULL, total_price REAL, unit_price INT, crawl_time TEXT, UNIQUE (house_id, total_price, crawl_time) );这张表每次写入的是一条历史记录不删旧数据。后续想看某个小区均价走势直接按house_id查历史表即可。如果某个房源在两次抓取之间价格没有变化通过UNIQUE(house_id, total_price, crawl_time)可以避免重复记录。要提醒的是入库操作本身比解析更耗时Pipeline处理速度要控制在Spider产出速度以上否则会导致内存积压实测中发现SQLite逐条commit在小数据量下没问题但超过5万条后建议改成批量提交每100条commit一次。5.4 扩容到MySQL时的改造点项目进入生产环境后SQLite的并发能力和运维监控都偏弱我会切换到MySQL。迁移成本主要在Pipeline的连接方式以及表结构里的字符集指定。MySQL表结构需要显式指定utf8mb4否则小区名里生僻字可能写入失败。其余字段设计可以保持一致这样Spider不需要改动只是替换Pipeline即可。6. 落盘后的验证与优化用Scrapy扩展核验数据质量6.1 用Scrapy Shell验证选择器正确性改动解析代码后不要直接跑步整个爬虫先用Scrapy Shell验证单页选择器。这是排查解析问题最有效的手段。scrapy shell https://cs.lianjia.com/ershoufang/yueluqu/进入Shell后可以用response.css(ul.sellListContent li.clear)确认列表卡片数量再用response.css(.title a::attr(href)).get()确认详情链接能取到。Shell里能正确取到结果再把选择器复制进Spider。这样可以避免每次改代码都要跑全量也能更快区分是页面结构变化还是代码逻辑问题。6.2 用Stats统计解析成功率Scrapy的Stats在默认情况下会统计请求数、item产出数、异常数。在Spider里可以直观看到一次任务后的关键指标scrapy crawl cs_lianjia_spider -s CLOSESPIDER_ITEMCOUNT10CLOSESPIDER_ITEMCOUNT是Scrapy对爬虫收集到指定条数后自动停止的扩展适合快速验证小批量链路。任务结束后把item_scraped_count和downloader/request_count做除法就能得到粗略的解析成功率。正常情况下这个比例应该在90%以上如果明显偏低大概率是列表页解析部分失效而不是详情页的问题。为了让这个指标更容易观察我会在spider_closed信号里主动打印。def close(spider, reason): stats spider.crawler.stats scraped stats.get_value(item_scraped_count, 0) requests stats.get_value(downloader/request_count, 0) spider.logger.warning( parse_rate%.2f scraped%d requests%d, scraped / requests if requests else 0, scraped, requests )6.3 定时任务与运行状态检查数据链路稳定后把爬虫挂到定时任务里。Linux上最直接的是crontab但要正确指定Python解释器和项目目录。0 9 * * * cd /opt/house-spider /usr/bin/python3 -m scrapy crawl cs_lianjia_spider logs/crawl.log 21这段命令每天早上九点进入项目目录调用系统Python直接执行Scrapy模块。这里的关键是不要用python而要用/usr/bin/python3确保crontab PATH里找不到命令时也能正常运行。爬虫跑完后配合crawler.stats里的指标做简单告警如果item_scraped_count低于预设阈值就把日志推送到告警群表示页面可能改版。这套结构后续要改成分布式爬虫也很直接把Spider的请求队列接到Scrapy-Redis上即可。有条件的话把每日抓取和报价分析串成一条命令抓完数据后立即生成行情日报整个流程就不需要人工盯守了。本文还有配套的精品资源点击获取