Python贝壳网二手小区数据爬取(2025年3月更)

文章目录

一、代码整体架构解析
二、各部分代码详解
- 1. main()主函数解析
- 2. 会话初始化（伪装浏览器身份）
- 3. 动态参数生成（反爬虫核心机制）
- 4. 列表页抓取（获取小区列表）
- 5. 列表页解析（提取小区信息）
- 6. 多线程详情页抓取（高效采集）
- 7. 详情页解析（深度数据提取）
- 8. 主流程控制（程序大脑）
三、核心技术点总结
四、完整代码
五、运行效果
六、特别说明(运行不了的时候看这里！)

上次写了用python爬取安居客二手小区数据，这次应粉丝要求写一篇爬取贝壳网二手小区数据的教程，因为贝壳网的反爬策略比安居客的更为复杂，所以这次就大胆一试！！！
这次教程内容大部分都是由Ai帮我写的，所以可能不太详细！！ 欢迎私信或者评论区提问~~
先来看看网页,在这里找到小区，并选择具体的市县区域：
在这里插入图片描述
然后我们F12打开网络看看数据在哪里

我们在这个文档里面找到了对应的小区数据，言外之意只需要请求这个url即可得到想要的数据啦~
同时，在这些信息里面还有一个小区详情页的链接，里面有诸如小区的容积率、绿化率、开发商等等信息,我们一起给它爬下来
在这里插入图片描述

大体思路就是：

先用一些用户信息如浏览器类型、cookie等信息来包装爬虫
请求一遍小区列表的url来获取小区总数
通过小区总数来计算需要请求的页数，使用for循环来遍历每一页的小区
提取每页小区的数据，得到每个小区的详情页链接后继续发送请求来进入详情页
提取小区详情页数据
将所有结果保存为一个excel表格

废话不多说，直接上干货~

一、代码整体架构解析

# 导入必要库（相当于工具箱）
import requests  # 网络请求工具
import time      # 时间控制工具
import random    # 随机数生成器
import pandas as pd  # 数据表格工具
from bs4 import BeautifulSoup  # HTML解析器
import math      # 数学计算工具
from concurrent.futures import ThreadPoolExecutor, as_completed  # 多线程工具

下面是整体的函数流程以及每个函数大体的作用~

序号	函数名称	功能描述	输入参数	返回值
1	`init_session(config)`	初始化网络会话对象	`config`: 用户配置字典	`requests.Session`对象
2	`get_params(session)`	生成动态请求参数	`session`: 会话对象	请求参数字典
3	`fetch_list_page()`	抓取列表页数据	`session`, `page_url`	解析后的数据列表
4	`parse_list_page(html)`	解析列表页HTML内容	`html`: 页面源代码字符串	小区信息列表
5	`fetch_detail_batch()`	批量抓取详情页数据	`session`, `urls`	详情数据字典
6	`parse_detail_page()`	解析详情页完整信息	`session`, `url`	详细字段字典
7	`crawl_full_data()`	主控流程（分页抓取数据）	`session`	合并后的完整数据列表

二、各部分代码详解

1. main()主函数解析

要修改的地方主要有四个，其余的不需要特别的改动！！！！

城市(city)
市县(region)
Cookies
excel表格输出的路径

# 主程序入口（程序起点）
if __name__ == "__main__":
    # ================== 用户配置区域 ==================
    CONFIG = {
        "city": "fs",       # 目标城市拼音（如: 佛山->fs，上海->sh）
        "region": "nanhai", # 目标区域拼音（如: 南海区->nanhai）
        "cookies": {        # 必需Cookie
            'lianjia_uuid': '自修修改',
            'lianjia_token': '自行修改',
            'security_ticket': '自行修改'
        },
        "srcid": "自行修改"
    }
    #输出的excel路径
    output_name = f'{CONFIG["city"]}_{CONFIG["region"]}_小区数据.xlsx'
    # ================================================

    # 初始化会话
    session = init_session(CONFIG)
    
    # 执行爬取
    start_time = time.time()
    final_data = crawl_full_data(session)
    
    # 保存结果
    if final_data:
        df = pd.DataFrame(final_data)[[
            '小区名称', '参考均价', '成交信息', '出租信息', '行政区', '商圈', '建筑年代',
            '详情页均价', '建筑类型', '房屋总数', '楼栋总数', '绿化率', '容积率',
            '交易权属', '建成年代', '供暖类型', '用水类型', '用电类型',
            '物业费', '附近门店', '物业公司', '开发商', '详情页链接'
        ]] 
        df.to_excel(output_name, index=False)
        print(f"数据已保存至: {output_name}")
        print(f"总计 {len(df)} 条数据，耗时 {(time.time()-start_time)/60:.1f} 分钟")

🔍 参数说明：

city：目标城市拼音缩写（如佛山→fs，广州→gz）
region：目标区域拼音缩写（如天河区→tianhe）
cookies：登录贝壳网后浏览器生成的登录凭证（关键！没有它无法获取数据）
srcid：加密参数（需从网页源代码中复制，定期更新防止失效）

2. 会话初始化（伪装浏览器身份）

def init_session(config):
    session = requests.Session()  # 创建会话容器
    session.headers.update({      # 设置请求头（伪装浏览器）
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)...',  # 浏览器指纹
        'Referer': f'https://{config["city"]}.ke.com/'  # 来源页面
    })
    session.cookies.update(config["cookies"])  # 加载登录凭证
    return session

🛡️ 参数解析：

Session对象：保持TCP连接复用，提升访问速度
User-Agent伪装：模拟Chrome浏览器访问（防止被识别为爬虫）
Referer伪造：隐藏真实来源页面（如访问北京小区页时，显示来自bj.ke.com）
Cookie管理：自动携带登录凭证（相当于拿着钥匙开门）

3. 动态参数生成（反爬虫核心机制）

def get_params(session):
    return {
        '_t': str(int(time.time() * 1000)),  # 13位时间戳（防重复）
        'srcid': session.config['srcid']     # 设备指纹（防篡改）
   }

⏳ 时间戳的作用：防止重复请求被识别

time.time()：获取当前时间（精确到秒）
*1000：转换成毫秒级精度
int()：去掉小数部分
str()：转换成字符串

🔑srcid的重要性：

设备唯一标识符（类似手机的IMEI号）
需定期从网页源代码更新（右键网页→查看源代码→搜索srcid）

4. 列表页抓取（获取小区列表）

def fetch_list_page(session, page_url):
    time.sleep(random.uniform(0.2, 0.4))  # 随机等待0.2-0.4秒
    response = session.get(page_url, timeout=8)  # 发送网络请求
    return parse_list_page(response.text)  # 解析HTML内容

⏱️ 时间控制：

random.uniform(0.2,0.4)：生成0.2到0.4之间的随机数
time.sleep()：让程序暂停指定时间
目的：模拟人类浏览行为，防止触发反爬机制

🚨 异常处理：
如果请求失败（超时、404错误等），会自动跳过并打印错误信息

5. 列表页解析（提取小区信息）

def parse_list_page(html):
    soup = BeautifulSoup(html, 'html.parser')  # 创建HTML解析器
    items = soup.select('li.xiaoquListItem')      # 定位所有小区条目
    
    results = []
    for item in items:
        info = {
            '小区名称': item.select_one('.title a').text.strip(),  # 提取名称
            '参考均价': item.select_one('.totalPrice span').text + '元/㎡' if ... else '暂无数据'
            # 其他字段类似...
        }
        results.append(info)
    return results

🔍 CSS选择器用法：

select_one(‘.title a’)：选择class为"title"的元素下的第一个标签
.text.strip()：提取文本内容并去除两端空白
条件判断：如果某个元素不存在（如无均价信息），显示"暂无数据"

6. 多线程详情页抓取（高效采集）

def fetch_detail_batch(session, urls):
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 提交所有URL到线程池
        future_to_url = {executor.submit(parse_detail_page, url): url for url in urls}
        
        # 逐个获取结果
        for future in as_completed(futures):
            url = future_to_url[future]
            details[url] = future.result()
            time.sleep(random.uniform(0.2, 0.4))  # 保持访问节奏

🚀 多线程原理：

ThreadPoolExecutor(max_workers=3)：同时开启3个线程
as_completed()：哪个线程先完成就先处理结果
限速机制：每个请求间隔0.2-0.4秒，避免服务器压力过大

7. 详情页解析（深度数据提取）

def parse_detail_page(session, url):
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # 解析多列布局数据
    def extract_multi_column():
        data = {}
        for col in soup.select('.xiaoquInfoItemCol'):
            for item in col.select('.xiaoquInfoItem'):
                label = item.select_one('.xiaoquInfoLabel').text.strip()
                value = item.select_one('.xiaoquInfoContent').text.strip()
                data[label] = value
        return data
    
    # 提取关键字段
    detail_data = {
        '房屋总数': ''.join(filter(str.isdigit, multi_col_data.get('房屋总数', ''))) or '0',
        '绿化率': multi_col_data.get('绿化率', '').replace('%', '') if multi_col_data.get('绿化率') else '暂无数据'
        # 其他字段...
    }
    return detail_data

🔧 数据清洗技巧：

filter(str.isdigit, “总计1582户”)：提取纯数字（结果：“1582”）
replace(‘%’, ‘’)：去除百分比符号（结果：“35”）
容错处理：使用or和条件表达式处理缺失字段

8. 主流程控制（程序大脑）

def crawl_full_data(session):
    try:
        # 获取总小区数
        total = int(soup.select_one('h2.total span').text)
        total_pages = math.ceil(total / 30)  # 每页30条数据
        
        print(f"\n当前区域共有 {total} 个小区")
        print(f"需要爬取 {total_pages} 页数据\n")
        
    except Exception as e:
        print(f"获取总数失败: {str(e)}")
        total_pages = 1  # 异常时默认只爬取1页
    
    all_data = []
    for page in range(1, total_pages + 1):
        for retry in range(2):  # 最多重试2次
            try:
                list_data = fetch_list_page(page_url)
                detail_results = fetch_detail_batch(list_data)
                
                # 合并数据
                for item in list_data:
                    item.update(detail_results.get(item['详情页链接'], {}))
                
                all_data.extend(list_data)
                print(f"第{page}页完成，累计{len(all_data)}条数据")
                break
            except Exception as e:
                print(f"第{retry+1}次重试失败: {str(e)}")
        time.sleep(random.uniform(0.2, 0.4))  # 页间延迟
    return all_data

📊 流程控制要点：

智能分页：自动计算总页数（例如100个小区→4页）
双重保障：每页最多重试2次，确保数据完整性
数据合并：将列表页基础信息与详情页数据合并
限速机制：页间访问间隔0.2-0.4秒

三、核心技术点总结

数据清洗三板斧

过滤非数字：filter(str.isdigit, text) → 保留纯数字
文本替换：.replace(old, new) → 删除/替换特定字符
条件赋值：value if condition else default → 处理缺失数据

容错机制设计

 .get(key, default)：安全获取字典值，避免KeyError or '默认值'：当结果为空时提供兜底方案
 if condition：严格校验数据存在性

字符串处理技巧

.strip()：去除首尾空白符
.split()：按空白符分割字符串
' '.join(list)：用空格连接列表元素

四、完整代码

import requests
import time
import random
import pandas as pd
from bs4 import BeautifulSoup
import math
from concurrent.futures import ThreadPoolExecutor, as_completed

def init_session(config):
    """初始化会话对象"""
    session = requests.Session()
    session.headers.update({
        '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': f'https://{config["city"]}.ke.com/xiaoqu/{config["region"]}/'
    })
    session.cookies.update(config["cookies"])
    session.config = config  # 存储配置信息
    return session

def get_params(session):
    """生成动态请求参数"""
    return {
        '_t': str(int(time.time() * 1000)),
        'srcid': session.config['srcid']
    }

def fetch_list_page(session, page_url):
    """抓取列表页数据"""
    try:
        time.sleep(random.uniform(0.2, 0.4))
        response = session.get(page_url, params=get_params(session), timeout=8)
        response.raise_for_status()
        return parse_list_page(response.text)
    except Exception as e:
        print(f"列表页请求失败: {str(e)}")
        return []

def parse_list_page(html):
    """解析列表页信息"""
    soup = BeautifulSoup(html, 'html.parser')
    items = soup.select('li.xiaoquListItem')
    
    results = []
    for item in items:
        try:
            info = {
                '小区名称': item.select_one('.title a').text.strip(),
                '参考均价': f"{item.select_one('.totalPrice span').text}元/㎡" if item.select_one('.totalPrice') else '暂无数据',
                '成交信息': item.select_one('.houseInfo a[href*="chengjiao"]').text.strip() if item.select_one('.houseInfo a[href*="chengjiao"]') else "暂无成交",
                '出租信息': item.select_one('.houseInfo a[href*="zufang"]').text.strip() if item.select_one('.houseInfo a[href*="zufang"]') else "暂无出租",
                '行政区': item.select_one('.district').text.strip() if item.select_one('.district') else "未知区域",
                '商圈': item.select_one('.bizcircle').text.strip() if item.select_one('.bizcircle') else "未知商圈",
                '建筑年代': ' '.join(item.select_one('.positionInfo').stripped_strings).split('/')[-1].strip() if item.select_one('.positionInfo') else "未知",
                '详情页链接': item.select_one('a.maidian-detail[href]')['href']
            }
            results.append(info)
        except Exception as e:
            print(f"解析异常: {str(e)}")
    return results

def fetch_detail_batch(session, urls):
    """批量获取详情页数据"""
    details = {}
    with ThreadPoolExecutor(max_workers=3) as executor:
        future_to_url = {executor.submit(parse_detail_page, session, url): url for url in urls}
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            details[url] = future.result()
            time.sleep(random.uniform(0.2, 0.4))
    return details

def parse_detail_page(session, url):
    """解析详情页完整信息"""
    try:
        time.sleep(random.uniform(0.6, 1.0))
        response = session.get(url, params=get_params(session), timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        
        # 辅助函数：安全提取单列信息
        def safe_extract_single(label_text):
            try:
                item = soup.find('span', class_='xiaoquInfoLabel', string=label_text)
                return item.find_next('span', class_='xiaoquInfoContent').text.strip()
            except:
                return '暂无数据'

        # 辅助函数：处理多列信息
        def extract_multi_column():
            data = {}
            columns = soup.select('.xiaoquInfoItemCol')
            for col_idx, col in enumerate(columns):
                items = col.select('.xiaoquInfoItem')
                for item in items:
                    label = item.select_one('.xiaoquInfoLabel').text.strip()
                    value = item.select_one('.xiaoquInfoContent').text.strip()
                    data[label] = value
            return data

        # 处理多列区域数据
        multi_col_data = extract_multi_column()
        
        # 处理单行区域数据（物业费、附近门店等）
        detail_data = {
            '建筑类型': multi_col_data.get('建筑类型', '暂无数据'),
            '房屋总数': ''.join(filter(str.isdigit, multi_col_data.get('房屋总数', ''))) or '0',
            '楼栋总数': ''.join(filter(str.isdigit, multi_col_data.get('楼栋总数', ''))) or '0',
            '绿化率': multi_col_data.get('绿化率', '').replace('%', '').strip(),
            '容积率': multi_col_data.get('容积率', '暂无数据'),
            '交易权属': multi_col_data.get('交易权属', '暂无数据'),
            '建成年代': multi_col_data.get('建成年代', '暂无数据'),
            '供暖类型': multi_col_data.get('供暖类型', '暂无数据'),
            '用水类型': multi_col_data.get('用水类型', '暂无数据'),
            '用电类型': multi_col_data.get('用电类型', '暂无数据'),
            # 处理单行区域
            '物业费': safe_extract_single('物业费').split('元')[0].strip(),
            '附近门店': ' '.join(safe_extract_single('附近门店').replace('\n', ' ').split()),
            '物业公司': safe_extract_single('物业公司'),
            '开发商': safe_extract_single('开发商'),
            '详情页均价': f"{soup.select_one('.xiaoquUnitPrice').text.strip()}元/㎡" if soup.select_one('.xiaoquUnitPrice') else '暂无数据'
        }
        
        return detail_data
        
    except Exception as e:
        print(f"详情页解析异常: {str(e)}")
        return {}

def crawl_full_data(session):
    """完整爬取流程"""
    config = session.config
    try:
        # 获取总小区数
        response = session.get(
            f"https://{config['city']}.ke.com/xiaoqu/{config['region']}/",
            params=get_params(session)
        )
        soup = BeautifulSoup(response.text, 'html.parser')
        total = int(soup.select_one('h2.total span').text)
        total_pages = math.ceil(total / 30)

        # 打印统计信息
        print(f"\n当前区域共有 {total} 个小区")
        print(f"需要爬取 {total_pages} 页数据\n")
        
    except Exception as e:
        print(f"获取总数失败: {str(e)}")
        total = 0
        total_pages = 0

    all_data = []
    for page in range(1, total_pages + 1):
        page_url = f"https://{config['city']}.ke.com/xiaoqu/{config['region']}/p{page}"
        
        for retry in range(2):
            try:
                list_data = fetch_list_page(session, page_url)
                detail_urls = [item['详情页链接'] for item in list_data]
                detail_results = fetch_detail_batch(session, detail_urls)
                
                for item in list_data:
                    item.update(detail_results.get(item['详情页链接'], {}))
                
                all_data.extend(list_data)
                print(f"第{page}页完成，累计{len(all_data)}条数据")
                break
            except Exception as e:
                print(f"第{retry+1}次重试: {str(e)}")        
        time.sleep(random.uniform(0.2, 0.4))
    
    return all_data

if __name__ == "__main__":
    # ================== 用户配置区域 ==================
    CONFIG = {
        "city": "fs",       # 目标城市拼音（如: 佛山->fs，上海->sh）
        "region": "nanhai", # 目标区域拼音（如: 南海区->nanhai）
        "cookies": {        # 必需Cookie
            'lianjia_uuid': '自行修改',
            'lianjia_token': '自行修改',
            'security_ticket': '自行修改'
        },
        "srcid": '自行修改'
    }
    #输出的excel路径
    output_name = f'{CONFIG["city"]}_{CONFIG["region"]}_小区数据.xlsx'
    # ================================================

    # 初始化会话
    session = init_session(CONFIG)
    
    # 执行爬取
    start_time = time.time()
    final_data = crawl_full_data(session)
    
    # 保存结果
    if final_data:
        df = pd.DataFrame(final_data)[[
            '小区名称', '参考均价', '成交信息', '出租信息', '行政区', '商圈', '建筑年代',
            '详情页均价', '建筑类型', '房屋总数', '楼栋总数', '绿化率', '容积率',
            '交易权属', '建成年代', '供暖类型', '用水类型', '用电类型',
            '物业费', '附近门店', '物业公司', '开发商', '详情页链接'
        ]]
        df.to_excel(output_name, index=False)
        print(f"数据已保存至: {output_name}")
        print(f"总计 {len(df)} 条数据，耗时 {(time.time()-start_time)/60:.1f} 分钟")