Python爬虫+MongoDB+热力图：武汉链家二手房数据采集与分析实战

简介面向爬虫与数据分析学习者这套武汉链家房源采集与分析系统基于Python完整演示了从网页爬取、MongoDB存储到可视化输出的流程。项目成功抓取729条武汉链家数据对地址、户型、面积、价格等字段进行清洗入库并统计房源类型占比和房价水平最终以饼状图、柱形图以及热力图直观呈现适合房产数据研究和爬虫实践。压缩包共18个文件大小1.49MB核心包括爬取数据.py、可视化.py、toTxt.py三个Python脚本其中爬取数据.py负责解析页面并提取字段可视化.py用于生成统计图toTxt.py可导出数据配合fake_useragent.json模拟请求、thermalMap.html动态页面和多张PNG结果图另附README.md、说明文件.txt及附赠资源文档方便按步骤复现。源码中涉及requests、BeautifulSoup解析、Pandas统计、Matplotlib绘图还给出了MongoDB的存取写法可直接修改用于其他城市房源数据采集。已有75人学习下载可帮助入门者快速搭建完整的数据分析项目。1. 这个项目到底在解决什么729套房源撑起一条完整的数据链路729个武汉链家房源样本放在大数据语境里实在不算多。但这套项目的价值恰恰不在样本量而在它把「爬虫采集 → 数据清洗 → MongoDB 存储 → 聚合分析 → 热力图可视化」这条完整链路一次性走通了。对准备交课程设计、毕业设计或者想从零搭一套可复用的数据抓取分析骨架的人来说这是性价比很高的参考方向每一步都能在本地跑通每个环节都能看到实际输出数据规模不大但表结构、聚合查询、地理可视化这些技能点全都能练到。这套系统最常见的落地做法是先用 Python requests 抓取链家武汉二手房列表页解析出标题、小区、户型、面积、总价、单价、朝向、标签等字段然后写进 MongoDB 集合用文档模型承接这些半结构化字段接着通过 MongoDB 聚合管道或 pandas 做房源类型占比、房价水平分析最后用 folium 或 pyecharts 把带经纬度的房源落成热力图。下面按这条链路逐步讲每步给出可抄作业的代码和参数再补上我踩过的几个坑。2. 爬虫层requests 把武汉链家二手房页翻完字段解析和反爬参数一次说清2.1 链家二手房列表页的 URL 规则和请求头构造链家武汉二手房列表页的结构是按分页 URL 组织的第一页不带页码第二页起 URL 形如https://wh.lianjia.com/ershoufang/pg2/pg后面的数字就是页码数字。整个武汉区域默认列表大约有一百页左右标题里说的 729 条只是其中某一时间点抓下来的一个子集所以写爬虫时不能假设固定页数要在代码里判断「翻到没有数据就停」。请求头里最关键的是 User-Agent、Referer 和 Cookie。链家对无 UA 的裸 requests 请求大概率直接返回 419 或空页面UA 至少写成一个完整的 Chrome 版本串。Referer 写https://wh.lianjia.com/让服务端认为你是从首页点进来的。Cookie 不是必需的但带着一个浏览器里复制出来的 Cookie 能把触发验证码的概率降一档。下面是抓取前几页并解析房源列表项的最小脚本import time import random import requests from bs4 import BeautifulSoup HEADERS { 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://wh.lianjia.com/, Accept-Language: zh-CN,zh;q0.9, } def fetch_page(page_no: int): url https://wh.lianjia.com/ershoufang/pg{}/.format(page_no) if page_no 1 \ else https://wh.lianjia.com/ershoufang/ resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() return resp.text # 解析第一页验证字段结构后再批量翻页 html fetch_page(1) soup BeautifulSoup(html, lxml) items soup.select(li .info) print(本页房源条数:, len(items))逻辑说明先取第一页 HTML用 BeautifulSoup 的select(li .info)选中列表项容器如果打印出来的条数是 0说明页面结构变了或触发了反爬这时候不应该继续翻页而是先检查返回内容里有没有验证码关键字。fetch_page里把第一页和其他页分开处理是因为链家首页列表 URL 不带pg参数直接给pg1也能访问但保持与网站自身规则一致更稳妥。参数说明timeout10控制单次请求超时lxml解析速度比 html.parser 快解析大页面时差异明显raise_for_status()会在返回 4xx/5xx 时直接抛异常避免把错误页面当成正常数据往下解析。2.2 解析单条房源字段从 HTML 结构里把总价、单价、面积、标签抠出来链家列表页每条房源的 HTML 结构相对稳定标题在.title a里位置信息在.positionInfo里总价在.totalPrice span里单价在.unitPrice里房屋信息户型、面积、朝向、装修、楼层在.houseInfo这段文本里用换行符分隔。第一次写解析的人最容易在这里翻车直接取整段文本然后发现字段顺序偶尔不一致。def parse_item(item): title item.select_one(.title a).get_text(stripTrue) position item.select_one(.positionInfo).get_text( , stripTrue) house_info item.select_one(.houseInfo) total_price item.select_one(.totalPrice span).get_text(stripTrue) unit_price item.select_one(.unitPrice).get_text(stripTrue) # houseInfo 的典型文本: # 2室1厅 | 88.5平米 | 南 北 | 精装 | 中楼层(共33层) | 2016年建 | 板楼 parts [p.strip() for p in house_info.get_text(|, stripTrue).split(|)] return { title: title, position: position, house_type: parts[0] if len(parts) 0 else , area: float(parts[1].replace(平米, )) if len(parts) 1 else None, orientation: parts[2] if len(parts) 2 else , decorate: parts[3] if len(parts) 3 else , floor: parts[4] if len(parts) 4 else , total_price: float(total_price.replace(万, )), unit_price: int(unit_price.replace(元/平, ).replace(,, )), tags: [a.get_text(stripTrue) for a in item.select(.tag)], }逻辑说明get_text(|, stripTrue)里的分隔符参数很关键默认是空字符串文本会全部连在一起这里先按竖线分隔再清洗能把户型、面积、朝向这些字段稳定拆开。total_price和unit_price都是字符串必须去掉中文单位再转数值类型否则后面 MongoDB 里存的是字符串聚合运算时会得到一堆错误结果。参数说明面积字段从88.5平米里用replace(平米, )去掉单位再包float()单价里有逗号分隔符得先replace(,, )。tags可能为空列表这是正常的链家只给部分房源打标签。2.3 翻页策略与反爬节奏随机延时和失败重试怎么写才不显得刻意翻页抓取最忌讳的是固定间隔比如 1 秒一次看起来规律太明显。常见做法是在 1.5 到 3.5 秒之间取随机延时。另一个容易被忽略的点是单页多次请求如果某页解析出来条数为 0不要立刻判定「抓完了」链家偶发会返回一页空壳重试两次再放弃更稳。def crawl_pages(max_pages: int, max_retry: int 2): results [] for page_no in range(1, max_pages 1): for attempt in range(max_retry): try: html fetch_page(page_no) soup BeautifulSoup(html, lxml) items soup.select(li .info) if len(items) 0: raise ValueError(页面无房源列表疑似被反爬或页面结构变动) for it in items: results.append(parse_item(it)) break except Exception as e: print(fpage {page_no}, attempt {attempt1} failed: {e}) if attempt max_retry - 1: print(fpage {page_no} give up) time.sleep(random.uniform(2, 4)) time.sleep(random.uniform(1.5, 3.5)) return results逻辑说明双层循环把「翻页」和「重试」解耦内层失败只重试当前页不把整个任务打乱。break只在解析成功时执行连续失败两轮就跳过这一页保证整体任务能跑完。每页结束后再睡一段随机时间这不是玄学是把请求间隔打散降低被封概率。参数说明max_pages100是上限但不是目标实际抓多少页取决于运行时判断max_retry2适合链家这种偶发空页的情况超过两次基本就是 IP 风控再多重试只会加重封锁。这里数据先攒在 Python 列表里每 500 条可以写一次中间文件防止进程中断丢失全部数据。3. 存储层MongoDB 文档结构和 pymongo 批量入库替 MySQL 省下建表麻烦3.1 为什么这个场景用 MongoDB 而不是 MySQL链家房源数据天然是半结构化字段数量不完全一致有的有电梯有的没有有的有标签有的没有、标签字段是数组、后期还想加经纬度字段。如果用 MySQL要么预先设计一大张宽表字段固化后改起来麻烦要么拆多张表查询时 join对 729 条数据来说纯属过度设计。MongoDB 的优势在于文档模型可以直接把一条房源存成一个 JSON 对象字段有没有都不影响插入数组字段原样存储聚合管道还能直接对数组做$unwind展开统计。另外 Mongo 的可视化工具有 Robo 3T / MongoDB Compass写完数据能直接看图检查对课程设计答辩来说很加分。标题里明确点了 MongoDB说明原始项目也是认这个选型的。这里要注意一个细节MongoDB 服务端口默认 27017Windows 上如果安装的是 MongoDB Compass 而不是 Server那pymongo连接会一直失败。我遇到过不止一次「安装完连不上 localhost」的情况最后发现是只装了图形客户端真正的 mongod 服务没启动。3.2 集合结构设计一个文档存一条房源索引建在查询字段上集合名用lianjia_wh每个文档对应一条房源。设计文档结构时遵循一个原则面向「后面要怎么查」来设计而不是照抄页面字段。比如后面要做房价水平分析按行政区分组算均价那么district字段必须单独拆出来不能混在position字符串里。from pymongo import MongoClient, ASCENDING import pymongo.errors client MongoClient(mongodb://localhost:27017/, serverSelectionTimeoutMS3000) db client[wuhan_lianjia] col db[ershoufang] col.create_index([(district, ASCENDING)]) col.create_index([(total_price, ASCENDING)]) doc { title: 中南 经典小户型 近地铁, district: 武昌, region: 中南, house_type: 2室1厅, area: 88.5, orientation: 南 北, decorate: 精装, floor: 中楼层(共33层), total_price: 185.0, unit_price: 20904, position: 武昌 中南 中南路, tags: [近地铁, 满五唯一], crawl_time: 2025-01-10 20:30:00, url: https://wh.lianjia.com/ershoufang/xxx.html, } result col.insert_one(doc) print(inserted id:, result.inserted_id)逻辑说明district行政区和region商圈从position字符串里拆分出来存成独立字段这样之后按武昌、洪山、汉阳分组统计时直接用聚合管道的$group就能做。crawl_time存字符串而不是日期对象是出于可读性考虑后续要按时间过滤排序也能直接用字符串比较。url字段存原始链接方便出问题时回溯检查。参数说明serverSelectionTimeoutMS3000表示 3 秒内连不上 MongoDB 就报错避免 pymongo 默认 30 秒的漫长等待。create_index在数据量小的时候看起来多余但这是让你养成建索引的习惯数据量一旦上来没有索引的聚合查询会把 CPU 打满。3.3 批量入库和字段类型校验字符串数字混在同一个字段里的坑单个文档插入没性能问题但 729 条建议直接用insert_many一步完成。真正要小心的是数据清洗不到位导致 total_price 字段既有字符串又有浮点数MongoDB 允许这种混存但聚合求和时要么报错要么跳过排查起来很费劲。def clean_record(item: dict) - dict: record { title: str(item.get(title, )), district: str(item.get(district, )), total_price: float(item[total_price]), unit_price: int(item[unit_price]), area: float(item[area]) if item.get(area) else None, } # 类型校验Total_price 转 float 失败直接抛异常不做静默忽略 if record[total_price] 0: raise ValueError(ftotal_price 异常: {item}) return record records [clean_record(r) for r in parsed_results] try: col.insert_many(records, orderedFalse) except pymongo.errors.BulkWriteError as e: print(部分文档写入失败跳过重复键:, e.details.get(nInserted))逻辑说明clean_record做的是白名单映射只保留分析时用得到的字段而不是把整条原始 JSON 甩进 Mongo。orderedFalse让批量写入遇到单条错误时继续写后面的不会因为一条脏数据中断整个导入。BulkWriteError捕获后打印nInserted能立刻知道多少条成功多少条失败。参数说明orderedFalse适合清洗已经做完的场景能最大程度保留有效数据如果你的逻辑要求「要么全部成功要么全部不写」那就不要加这个参数。float(item[total_price])会直接抛出 ValueError 而不是静默转成 0这样脏数据能在入库前暴露而不是在聚合时才出现诡异结果。4. 分析层房源类型占比、房价水平、热力图用聚合管道和 folium 落地4.1 房源类型占比用 MongoDB 聚合管道的 $group $unwind 统计户型分布房源类型最常见的是按house_type里的「室」数统计比如 2室、3室占比多少。因为 house_type 是2室1厅这样的字符串直接按完整字段分组的话2室1厅和2室2厅会被拆成两组。更好的做法是用$project截取出第一个「室」字前面的数字再按这个一室、二室、三室分组统计。pipeline [ {$project: { rooms: {$substr: [$house_type, 0, 1]}, }}, {$group: {_id: $rooms, count: {$sum: 1}}}, {$sort: {count: -1}}, ] for doc in col.aggregate(pipeline): print(doc[_id], 室:, doc[count])逻辑说明$substr从house_type的索引 0 开始截 1 个字符正好拿到2室1厅里的2。这样聚合输出的就是2、3、4这些离散分组再做饼图就很干净。$sort按数量倒序排输出结果直接能看出武汉链家二手房市场的主力户型段。参数说明$substr在 MongoDB 4.4 之后的聚合框架里索引是字节位置对 ASCII 数字没问题如果 house_type 字段为空这里会得到空串分组时会多出一个_id: 的分组后续可以顺手加$match过滤掉。4.2 房价水平分析按行政区分组算均价用 pandas 补一个中位数视角均价计算不能只做平均729 条数据里如果混入几个千万级豪宅平均价会被拉得很高。更可靠的指标是「行政区均价 中位数」两个一起看。这里我习惯把 Mongo 数据抽出来转成 pandas DataFrame用 groupby 快速算比写复杂聚合管道直观。import pandas as pd cursor col.find({}, { _id: 0, district: 1, total_price: 1, area: 1, unit_price: 1 }) df pd.DataFrame(list(cursor)) df[unit_price] pd.to_numeric(df[unit_price], errorscoerce) district_stat df.groupby(district).agg( avg_price(unit_price, mean), median_price(unit_price, median), cnt(unit_price, count), ).sort_values(avg_price, ascendingFalse) print(district_stat)逻辑说明col.find的第二个参数是投影只取分析必需的 4 个字段不把 tags 这种数组字段一起拉出来减少网络和内存开销。pd.to_numeric(errorscoerce)把无法转数值的单价变成 NaN后续聚合自动跳过比 Python 层清洗好使。参数说明mean和median同时算的意义在于——如果某行政区的均价快 25000 而中位数只有 18000说明这个区有高价盘拉高了整体水平单看均价容易误判。cnt是每区的样本量样本少于 10 条的区域统计意义不大可视化时可以标注提示。4.3 热力图可视化经纬度从哪来folium 踩过的坐标偏移坑链家列表页里没有经纬度字段房源只有文字地址。常见做法是把「小区名 武汉」拼接后调高德或百度地理编码 API拿到经纬度。这一步有两个坑一是地理编码 API 有 QPS 限制729 个小区得加延时串行请求不能并发打二是高德返回的是 GCJ-02 坐标系而 folium 底图默认用 OpenStreetMap 的 WGS84 坐标系直接叠加会有约几百米的偏移小区级别的热力图看起来会偏到隔壁街区。import folium from folium.plugins import HeatMap # points [(lat, lng), ...] 来自地理编码结果 m folium.Map(location[30.59, 114.30], zoom_start11, tilesOpenStreetMap) HeatMap(points, radius15, blur10, max_zoom1).add_to(m) m.save(wuhan_house_price_heatmap.html)解决坐标偏移最省事的办法是换瓦片源把tiles改成https://webrd02.is.autonavi.com/appmaptile?langzh_cnsize1scale1style8x{x}y{y}z{z}并设置attr高德地图这样底图和经纬度都在 GCJ-02 坐标系里偏移问题消失。如果不想换底图就用百度坐标系的瓦片但地理编码 API 也要换成百度的混用才会有鬼畜偏移。参数说明radius15控制每个热力点的辐射半径数值越大热力图越平滑blur10是模糊半径影响色块边缘过渡max_zoom1限制热力图在放大的时候不重新计算权重亚像素级区域用起来更流畅。热力图渲染的权重默认是点的频次想按房价加权就把每个点写成(lat, lng, weight)三元组weight 取unit_price / 10000让高单价区域自然亮红。5. 武汉链家爬虫与 MongoDB 的五个坑现象、原因、解决都在这里5.1 单页返回 0 条数据不一定是对抗策略可能是页面结构改版现象脚本运行到第 20 页时某页解析出来的房源条数突然是 0但浏览器里打开同一 URL 能看到完整列表。原因有两层一是短时间内请求频率过高触发了链家的临时验证页二是链家前端类名在小范围改版.info这个 CSS 选择器没匹配上。解决先打印返回 HTML 的前 500 个字符做人工判断。如果包含「验证」「滑动」之类的关键字说明是被临时校验拦了停 30 秒再继续如果 HTML 正常但 select 结果为空用浏览器开发者工具重新定位列表项的选择器常见替代选择器是li[class*clear]。我在第一次写链家爬虫时就直接被第二类坑坑过某天.info选择器突然失效原因是链家把div.info改成了div.info--new。后来我把选择器改成li div:last-child这种更稳定的结构关系不再依赖单一类名。5.2 单价字段解析出 0 或乱码类名没选对取了隐藏节点现象部分房源解析出的unit_price是 0或者打印出「暂无单价」这种文案。原因链家在少数房源上不展示单价页面里的.unitPrice节点存在但文本为空另一个原因是误选到了同页的推荐广告卡片广告卡片结构长得跟房源卡片几乎一样。解决解析前做一层兜底取不到单价就用total_price * 10000 / area算一个估算值同时锁定列表项的父级容器不要全页面用select(.unitPrice)而是先定位到li .info再往下找避免抓到广告位。这个清洗逻辑我一般直接写进parse_item里解析出来单价为 0 的房源用公式回填并打一个标记字段price_source: estimated后续做均价统计时可以决定要不要排除。5.3 MongoDB 连接被拒绝mongod 服务没起来或版本不匹配现象pymongo 报ServerSelectionTimeoutError: localhost:27017 is not available但 MongoDB Compass 能正常打开。原因Compass 是图形客户端不是服务端Windows 上安装 MongoDB 时如果没有勾选 Install as a Servicemongod 进程不会随系统启动Compass 会自动连接一个嵌入式实例而 pymongo 连的是独立端口。解决以管理员身份打开命令行执行mongod --dbpath D:\data\db手动启动服务或者用sc create MongoDB binPath C:\mongodb\bin\mongod.exe --dbpath D:\data\db注册成 Windows 服务。Linux 上则要确认systemctl status mongod的状态是 active。连接 URI 里还可以加authSourceadmin如果你的 Mongo 开了认证的话。本地开发我通常不开认证但会绑定 127.0.0.1 而不是 0.0.0.0避免开发机被局域网扫到。5.4 热力图坐标偏了几个街区GCJ-02 和 WGS84 混用了现象房源点在北京地图上看起来整体往东北方向偏了大概 300 米单个点可能看不出来但热力图一叠加偏差很明显。原因高德/百度地理编码返回的都是 GCJ-02火星坐标而 folium 默认底图 OpenStreetMap 是 WGS84两个坐标系混用时点位整体漂移。解决要么换成高德瓦片源上文已写到要么自己在代码里做坐标纠偏。不要试图写纠偏函数——火星坐标转 WGS84 需要查表逼近算法网上流传的代码精度不一直接用高德瓦片最省心。注意百度坐标又不一样百度在 GCJ-02 基础上再加一层 BD-09 偏移所以同一份经纬度高德地图和百度地图点出来的位置也有差异。这个项目里自始至终用一个地图源就好。5.5 入库后聚合结果对不上数据类型不统一字符串和数字混存了现象$sum: $total_price算出来的总价是个天文数字或者直接报错。原因插入时total_price字段有的是字符串185万有的是浮点数185.0MongoDB 不强制类型一致导致聚合管道按数值累加时把字符串隐式转成了 0 或直接跳过。解决入库前用clean_record统一转换并在写完数据后跑一条验证语句bad_types col.count_documents({total_price: {$type: string}}) print(total_price 仍为字符串的文档数:, bad_types) if bad_types 0: col.update_many( {total_price: {$type: string}}, [{$set: {total_price: {$toDouble: $total_price}}}] )$type查询能快速找出混存记录$toDouble聚合更新符一步完成类型修复。这个验证脚本值得在每次入库后跑一遍二三十秒的事能省掉后面两小时的排查时间。6. 从一次性脚本到可持续任务增量采集、调度与断点续传的进阶做法6.1 增量采集判断逻辑用 crawl_time 字段做去重和断点标记729 条数据的采集是一次性快照但如果你想让这套系统每周自动更新就得考虑增量逻辑。最简单的做法是在 MongoDB 里查已存在的房源 URL用 URL 做唯一键。链家的房源 ID 在 URL 末尾天然适合做去重。插入前先建一个唯一索引col.create_index(url, uniqueTrue)这样insert_many时重复 URL 会触发DuplicateKeyError配合orderedFalse新数据正常写入旧数据被跳过。每次抓取时再给文档打上crawl_time datetime.now().strftime(%Y-%m-%d %H:%M:%S)就能区分哪些是本周新增、哪些是历史数据。6.2 调度部署APScheduler 定时任务和日志落地的习惯Windows 上做定时任务可以用计划任务程序但对 Python 脚本来说更灵活的方案是 APScheduler。写一个简单的间隔调度每天早上 8 点跑一次增量采集from apscheduler.schedulers.blocking import BlockingScheduler import logging logging.basicConfig( filenamelianjia_crawler.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s ) def job(): logging.info(crawler job start) # 采集、清洗、入库逻辑 logging.info(crawler job done) scheduler BlockingScheduler() scheduler.add_job(job, cron, hour8, minute0) scheduler.start()这里的关键是把日志写进文件而不是只 print 到控制台否则关掉命令窗口后任务跑没跑、跑成功没有你根本不知道。日志级别用 INFO采集失败、入库异常都要显式logging.exception记录堆栈。定时任务加日志是这套系统从「课程设计」变成「能长期用的数据服务」的转折点。我现在的习惯是每次改完采集逻辑先拿一页小数据做冒烟测试确认解析字段不报错再放给调度任务跑全量。这个习惯帮我避免了好几次「定时任务跑了三小时最后发现解析字段全错」的事故。希望帮到你。本文还有配套的精品资源点击获取