文章目錄
- 一、代碼整體架構解析
- 二、各部分代碼詳解
- 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] = valuereturn 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)}條數據")breakexcept 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_completeddef 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 sessiondef 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 resultsdef 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 detailsdef 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] = valuereturn 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_dataexcept Exception as e:print(f"詳情頁解析異常: {str(e)}")return {}def crawl_full_data(session):"""完整爬取流程"""config = session.configtry:# 獲取總小區數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 = 0total_pages = 0all_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)}條數據")breakexcept Exception as e:print(f"第{retry+1}次重試: {str(e)}") time.sleep(random.uniform(0.2, 0.4))return all_dataif __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} 分鐘")
五、運行效果
六、特別說明(運行不了的時候看這里!)
- 記得替換main里面的各種參數,尤其是cookies!用的cookies是你自己瀏覽器登陸貝殼網,并且完成驗證之后的那個cookies!!!
- 如果沒有修改excel輸出路徑找不到輸出的文件,就在這個代碼文件所在的文件夾里面找
- 如果嫌爬取速度太慢可以自行修改time.sleep()里的時間,當然間隔越小被反爬的概率越大
- 不能保證網頁結構后續恒久不變,比如class的標簽變了需要重新修改對應的標簽,因此代碼也具有時效性~