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

文章目录

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

    

    上次写了用python爬取安居客二手小区数据，这次应粉丝要求写一篇爬取贝壳网二手小区数据的教程，因为贝壳网的反爬策略比安居客的更为复杂，所以这次就大胆一试！！！

    这次教程内容大部分都是由Ai帮我写的，所以可能不太详细！！ 欢迎私信或者评论区提问~~

    先来看看网页,在这里找到小区，并选择具体的市县区域：

    

    然后我们F12打开网络看看数据在哪里

    

    

    我们在这个文档里面找到了对应的小区数据，言外之意只需要请求这个url即可得到想要的数据啦~

    同时，在这些信息里面还有一个小区详情页的链接，里面有诸如小区的容积率、绿化率、开发商等等信息,我们一起给它爬下来

    

    ---

    大体思路就是：

    1. 先用一些用户信息如浏览器类型、cookie等信息来包装爬虫
    2. 请求一遍小区列表的url来获取小区总数
    3. 通过小区总数来计算需要请求的页数，使用for循环来遍历每一页的小区
    4. 提取每页小区的数据，得到每个小区的详情页链接后继续发送请求来进入详情页
    5. 提取小区详情页数据
    6. 将所有结果保存为一个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()主函数解析

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

    1. 城市(city)
    2. 市县(region)
    3. Cookies
    4. 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的重要性：

          1. 设备唯一标识符（类似手机的IMEI号）
          2. 需定期从网页源代码更新（右键网页→查看源代码→搜索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秒

                    三、核心技术点总结

                    1. 数据清洗三板斧

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

                    1. 容错机制设计

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

                    1. 字符串处理技巧

                    .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} 分钟")
                    

                    五、运行效果

                    

                    

                    六、特别说明(运行不了的时候看这里！)

                    1. 记得替换main里面的各种参数，尤其是cookies！用的cookies是你自己浏览器登陆贝壳网，并且完成验证之后的那个cookies！！！
                    2. 如果没有修改excel输出路径找不到输出的文件，就在这个代码文件所在的文件夹里面找
                    3. 如果嫌爬取速度太慢可以自行修改time.sleep()里的时间，当然间隔越小被反爬的概率越大
                    4. 不能保证网页结构后续恒久不变，比如class的标签变了需要重新修改对应的标签，因此代码也具有时效性~