Python爬虫实战：链家二手房房源数据抓取全解析-洪萨配资

引言

在房产大数据时代，获取准确的二手房源信息对于市场分析、投资决策和学术研究都具有重要意义。本文将详细介绍如何使用Python最新技术栈构建一个高效、稳定的链家二手房数据爬虫系统。我们将从爬虫原理、技术选型、代码实现到数据存储进行全面讲解。

技术栈选型

Python 3.10+：使用最新Python版本确保性能和安全
Playwright：微软开源的无头浏览器，替代传统的Selenium
BeautifulSoup4 & lxml：HTML解析库
asyncio & aiohttp：异步IO处理，提升爬取效率
Pandas & NumPy：数据处理与分析
SQLAlchemy：数据库ORM框架
FastAPI：可选的数据API服务

项目结构设计

text

lianjia_crawler/ ├── config/ # 配置文件 │ └── settings.py ├── spiders/ # 爬虫核心 │ ├── base_spider.py │ └── lianjia_spider.py ├── models/ # 数据模型 │ ├── database.py │ └── models.py ├── utils/ # 工具函数 │ ├── logger.py │ ├── proxy_manager.py │ └── user_agent.py ├── middlewares/ # 中间件 │ └── anti_anti_crawler.py ├── storage/ # 数据存储 │ └── data_handler.py ├── requirements.txt # 依赖文件 └── main.py # 主程序入口

完整代码实现

1. 环境配置与依赖安装

python

# requirements.txt playwright==1.40.0 beautifulsoup4==4.12.0 lxml==5.0.0 aiohttp==3.9.0 pandas==2.1.0 sqlalchemy==2.0.0 asyncpg==0.29.0 pydantic==2.5.0 asyncio python-dotenv==1.0.0

2. 配置文件

python

# config/settings.py from pydantic_settings import BaseSettings from typing import List class Settings(BaseSettings): # 爬虫配置 MAX_CONCURRENT_REQUESTS: int = 5 REQUEST_TIMEOUT: int = 30 DELAY_RANGE: tuple = (1, 3) # 链家配置 BASE_URL: str = "https://bj.lianjia.com" START_URLS: List[str] = [ "https://bj.lianjia.com/ershoufang/pg{}/" ] # 数据库配置 DATABASE_URL: str = "postgresql+asyncpg://user:password@localhost:5432/lianjia" # 代理配置 USE_PROXY: bool = False PROXY_POOL_URL: str = "" # 日志配置 LOG_LEVEL: str = "INFO" class Config: env_file = ".env" settings = Settings()

3. 数据模型定义

python

# models/models.py from sqlalchemy import Column, Integer, String, Float, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base = declarative_base() class HouseListing(Base): """二手房房源数据模型""" __tablename__ = "house_listings" id = Column(String(50), primary_key=True) title = Column(String(200)) total_price = Column(Float) unit_price = Column(Float) district = Column(String(50)) region = Column(String(50)) community = Column(String(100)) floor = Column(String(50)) area = Column(Float) room_type = Column(String(50)) orientation = Column(String(50)) decoration = Column(String(50)) build_year = Column(Integer) listing_date = Column(DateTime) follow_count = Column(Integer) view_count = Column(Integer) tags = Column(Text) url = Column(String(500)) created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) def __repr__(self): return f"<HouseListing {self.title} - {self.total_price}万>"

4. 异步数据库连接

python

# models/database.py from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from config.settings import settings class Database: def __init__(self): self.engine = create_async_engine( settings.DATABASE_URL, echo=False, future=True, pool_size=20, max_overflow=0 ) self.async_session = sessionmaker( self.engine, class_=AsyncSession, expire_on_commit=False ) async def init_db(self): """初始化数据库""" async with self.engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async def get_session(self) -> AsyncSession: """获取数据库会话""" async with self.async_session() as session: yield session db = Database()

5. 核心爬虫类实现

python

# spiders/lianjia_spider.py import asyncio import re import json from typing import List, Dict, Optional from datetime import datetime import aiohttp from bs4 import BeautifulSoup from playwright.async_api import async_playwright import random import logging from urllib.parse import urljoin from pydantic import BaseModel, ValidationError from config.settings import settings from models.models import HouseListing from utils.logger import setup_logger from utils.user_agent import get_random_user_agent from middlewares.anti_anti_crawler import AntiAntiCrawlerMiddleware logger = setup_logger(__name__) class HouseItem(BaseModel): """数据验证模型""" id: str title: str total_price: float unit_price: float district: str region: str community: str floor: str area: float room_type: str orientation: str decoration: str build_year: Optional[int] listing_date: datetime follow_count: int view_count: int tags: List[str] url: str class LianJiaSpider: """链家二手房爬虫""" def __init__(self): self.base_url = settings.BASE_URL self.start_urls = settings.START_URLS self.session = None self.playwright = None self.browser = None self.context = None self.middleware = AntiAntiCrawlerMiddleware() async def init_session(self): """初始化aiohttp会话""" timeout = aiohttp.ClientTimeout(total=settings.REQUEST_TIMEOUT) self.session = aiohttp.ClientSession( timeout=timeout, headers={ 'User-Agent': get_random_user_agent(), 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', } ) async def init_browser(self): """初始化Playwright浏览器""" self.playwright = await async_playwright().start() self.browser = await self.playwright.chromium.launch( headless=True, args=[ '--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox', '--disable-setuid-sandbox' ] ) self.context = await self.browser.new_context( user_agent=get_random_user_agent(), viewport={'width': 1920, 'height': 1080}, ignore_https_errors=True ) # 添加stealth插件避免检测 await self.context.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); window.chrome = { runtime: {} }; """) async def crawl_list_pages(self, max_pages: int = 100): """爬取列表页""" listings = [] for start_url in self.start_urls: for page in range(1, max_pages + 1): url = start_url.format(page) logger.info(f"爬取列表页: {url}") try: # 使用Playwright处理动态内容 page_content = await self.fetch_with_playwright(url) if not page_content: logger.warning(f"页面内容为空: {url}") continue # 解析列表页 page_listings = await self.parse_list_page(page_content, url) listings.extend(page_listings) # 随机延迟 await asyncio.sleep(random.uniform(*settings.DELAY_RANGE)) # 检查是否到达末页 if not await self.has_next_page(page_content): logger.info("已到达最后一页") break except Exception as e: logger.error(f"爬取列表页失败 {url}: {str(e)}") continue return listings async def fetch_with_playwright(self, url: str) -> Optional[str]: """使用Playwright获取页面内容""" page = None try: page = await self.context.new_page() # 设置请求头 await page.set_extra_http_headers({ 'User-Agent': get_random_user_agent(), 'Referer': self.base_url, 'Accept-Language': 'zh-CN,zh;q=0.9' }) # 导航到页面 response = await page.goto(url, wait_until='networkidle', timeout=60000) if response.status != 200: logger.error(f"请求失败，状态码: {response.status}") return None # 等待关键元素加载 await page.wait_for_selector('.sellListContent', timeout=30000) # 获取页面内容 content = await page.content() # 执行JavaScript获取隐藏数据 additional_data = await page.evaluate(""" () => { const data = {}; // 提取页面上的JSON-LD数据 const scriptTags = document.querySelectorAll('script[type="application/ld+json"]'); scriptTags.forEach(script => { try { const jsonData = JSON.parse(script.textContent); Object.assign(data, jsonData); } catch(e) {} }); return data; } """) if additional_data: logger.debug(f"获取到额外数据: {additional_data}") return content except Exception as e: logger.error(f"Playwright获取页面失败: {str(e)}") return None finally: if page: await page.close() async def parse_list_page(self, html: str, page_url: str) -> List[Dict]: """解析列表页""" soup = BeautifulSoup(html, 'lxml') listings = [] # 查找房源列表 house_items = soup.select('.sellListContent li[class^="clear LOGCLICKDATA"]') for item in house_items: try: house_data = await self.extract_house_data(item, page_url) if house_data: listings.append(house_data) except Exception as e: logger.error(f"解析房源失败: {str(e)}") continue return listings async def extract_house_data(self, item, page_url: str) -> Optional[Dict]: """提取单个房源数据""" try: # 房源ID house_id = item.get('data-lj_action_housedel_id', '') if not house_id: return None # 标题 title_elem = item.select_one('.title a') title = title_elem.text.strip() if title_elem else '' # 总价 price_elem = item.select_one('.totalPrice span') total_price = float(price_elem.text.strip()) if price_elem else 0 # 单价 unit_price_elem = item.select_one('.unitPrice') unit_price_text = unit_price_elem.text.strip() if unit_price_elem else '' unit_price = float(re.search(r'(\d+)', unit_price_text).group(1)) if unit_price_text else 0 # 位置信息 position_elem = item.select_one('.positionInfo a') position_text = position_elem.text.strip() if position_elem else '' district, region = self.parse_position(position_text) # 小区 community_elem = item.select_one('.houseInfo a') community = community_elem.text.strip() if community_elem else '' # 房屋信息 house_info_elem = item.select_one('.houseInfo') house_info = house_info_elem.text.strip() if house_info_elem else '' area, room_type, floor = self.parse_house_info(house_info) # 关注信息 follow_elem = item.select_one('.followInfo') follow_text = follow_elem.text.strip() if follow_elem else '' follow_count, view_count = self.parse_follow_info(follow_text) # 标签 tag_elems = item.select('.tag span') tags = [tag.text.strip() for tag in tag_elems] # URL relative_url = title_elem.get('href', '') if title_elem else '' url = urljoin(self.base_url, relative_url) # 构建数据字典 house_data = { 'id': house_id, 'title': title, 'total_price': total_price, 'unit_price': unit_price, 'district': district, 'region': region, 'community': community, 'area': area, 'room_type': room_type, 'floor': floor, 'follow_count': follow_count, 'view_count': view_count, 'tags': json.dumps(tags, ensure_ascii=False), 'url': url, 'listing_date': datetime.now() } # 验证数据 try: validated_data = HouseItem(**house_data) return validated_data.dict() except ValidationError as e: logger.warning(f"数据验证失败: {e}") return None except Exception as e: logger.error(f"提取房源数据失败: {str(e)}") return None def parse_position(self, position_text: str) -> tuple: """解析位置信息""" if '·' in position_text: parts = position_text.split('·') if len(parts) >= 2: return parts[0].strip(), parts[1].strip() return '', '' def parse_house_info(self, house_info: str) -> tuple: """解析房屋信息""" area = 0 room_type = '' floor = '' try: parts = house_info.split('|') if len(parts) >= 3: # 面积 area_match = re.search(r'([\d\.]+)平米', parts[1].strip()) if area_match: area = float(area_match.group(1)) # 户型 room_type = parts[0].strip() # 楼层 floor = parts[2].strip() except Exception as e: logger.debug(f"解析房屋信息失败: {e}") return area, room_type, floor def parse_follow_info(self, follow_text: str) -> tuple: """解析关注信息""" follow_count = 0 view_count = 0 try: parts = follow_text.split('/') if len(parts) >= 2: follow_match = re.search(r'(\d+)', parts[0]) view_match = re.search(r'(\d+)', parts[1]) if follow_match: follow_count = int(follow_match.group(1)) if view_match: view_count = int(view_match.group(1)) except Exception as e: logger.debug(f"解析关注信息失败: {e}") return follow_count, view_count async def has_next_page(self, html: str) -> bool: """检查是否有下一页""" soup = BeautifulSoup(html, 'lxml') next_button = soup.select_one('.page-box .next') return next_button is not None and 'disabled' not in next_button.get('class', []) async def crawl_detail_pages(self, listings: List[Dict]): """爬取详情页补充信息""" detailed_listings = [] for listing in listings[:10]: # 限制爬取数量，避免请求过多 try: detail_data = await self.fetch_detail_page(listing['url']) if detail_data: listing.update(detail_data) detailed_listings.append(listing) # 随机延迟 await asyncio.sleep(random.uniform(2, 5)) except Exception as e: logger.error(f"爬取详情页失败 {listing['url']}: {str(e)}") continue return detailed_listings async def fetch_detail_page(self, url: str) -> Optional[Dict]: """爬取详情页""" page = None try: page = await self.context.new_page() await page.goto(url, wait_until='networkidle', timeout=60000) # 等待详情信息加载 await page.wait_for_selector('.overview', timeout=30000) # 执行JavaScript提取详情数据 detail_data = await page.evaluate(""" () => { const data = {}; // 提取装修信息 const decorationElem = document.querySelector('.base .content ul li:last-child'); if (decorationElem) { data.decoration = decorationElem.textContent.replace('装修', '').trim(); } // 提取朝向 const orientationElem = document.querySelector('.base .content ul li:nth-child(7)'); if (orientationElem) { data.orientation = orientationElem.textContent.replace('房屋朝向', '').trim(); } // 提取建筑年代 const yearElem = document.querySelector('.area .subInfo'); if (yearElem) { const yearMatch = yearElem.textContent.match(/(\\d{4})/); if (yearMatch) { data.build_year = parseInt(yearMatch[1]); } } return data; } """) return detail_data except Exception as e: logger.error(f"获取详情页数据失败: {str(e)}") return None finally: if page: await page.close() async def save_to_database(self, listings: List[Dict]): """保存数据到数据库""" from models.database import db async with db.async_session() as session: try: for listing in listings: # 检查是否已存在 existing = await session.get(HouseListing, listing['id']) if existing: # 更新现有记录 for key, value in listing.items(): setattr(existing, key, value) existing.updated_at = datetime.now() else: # 创建新记录 house = HouseListing(**listing) session.add(house) await session.commit() logger.info(f"成功保存 {len(listings)} 条数据到数据库") except Exception as e: await session.rollback() logger.error(f"保存数据失败: {str(e)}") raise async def run(self, max_pages: int = 10): """运行爬虫""" try: # 初始化 await self.init_browser() await self.init_session() # 爬取列表页 logger.info("开始爬取列表页...") listings = await self.crawl_list_pages(max_pages) logger.info(f"爬取到 {len(listings)} 条房源数据") # 爬取详情页（可选） if listings: logger.info("开始爬取详情页...") detailed_listings = await self.crawl_detail_pages(listings) # 保存到数据库 await self.save_to_database(detailed_listings) # 导出为CSV await self.export_to_csv(detailed_listings) logger.info("爬虫任务完成！") except Exception as e: logger.error(f"爬虫运行失败: {str(e)}") raise finally: # 清理资源 await self.cleanup() async def export_to_csv(self, listings: List[Dict]): """导出数据为CSV""" import pandas as pd if listings: df = pd.DataFrame(listings) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"lianjia_houses_{timestamp}.csv" # 选择需要的列 columns_to_export = [ 'id', 'title', 'total_price', 'unit_price', 'district', 'region', 'community', 'area', 'room_type', 'floor', 'orientation', 'decoration', 'build_year', 'follow_count', 'view_count', 'listing_date', 'url' ] df = df[[col for col in columns_to_export if col in df.columns]] df.to_csv(filename, index=False, encoding='utf-8-sig') logger.info(f"数据已导出到 {filename}") async def cleanup(self): """清理资源""" if self.session: await self.session.close() if self.context: await self.context.close() if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop()

6. 工具函数

python

# utils/user_agent.py import random USER_AGENTS = [ # Chrome "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # Firefox "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0", # Safari "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15", ] def get_random_user_agent(): """获取随机User-Agent""" return random.choice(USER_AGENTS)

7. 反反爬虫中间件

python

# middlewares/anti_anti_crawler.py import asyncio import random from typing import Dict, Optional import hashlib import time class AntiAntiCrawlerMiddleware: """反反爬虫中间件""" def __init__(self): self.request_history = [] self.blocked_count = 0 async def process_request(self, url: str, headers: Dict) -> Dict: """处理请求，添加反反爬虫策略""" # 随机延迟 delay = random.uniform(1, 5) await asyncio.sleep(delay) # 动态更新请求头 headers.update({ 'X-Requested-With': 'XMLHttpRequest', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'same-origin', 'Sec-Fetch-User': '?1', }) # 添加时间戳 headers['X-Timestamp'] = str(int(time.time() * 1000)) # 添加指纹（简化版） fingerprint = self.generate_fingerprint(headers) headers['X-Fingerprint'] = fingerprint return headers def generate_fingerprint(self, headers: Dict) -> str: """生成浏览器指纹""" fingerprint_str = f"{headers.get('User-Agent', '')}{int(time.time())}" return hashlib.md5(fingerprint_str.encode()).hexdigest() def check_anti_crawler(self, html: str) -> bool: """检查是否触发反爬虫""" warning_signs = [ "访问验证", "安全检查", "restricted", "blocked", "请输入验证码", "访问过于频繁" ] for sign in warning_signs: if sign in html: return True return False async def handle_blocked(self): """处理被封锁的情况""" self.blocked_count += 1 wait_time = 60 * (2 ** min(self.blocked_count, 4)) # 指数退避 logger.warning(f"检测到反爬虫，等待 {wait_time} 秒") await asyncio.sleep(wait_time)

8. 主程序入口

python

# main.py import asyncio import argparse import sys from spiders.lianjia_spider import LianJiaSpider from utils.logger import setup_logger logger = setup_logger(__name__) async def main(): """主函数""" parser = argparse.ArgumentParser(description='链家二手房爬虫') parser.add_argument('--pages', type=int, default=10, help='爬取的页数') parser.add_argument('--max-workers', type=int, default=5, help='最大并发数') parser.add_argument('--export-csv', action='store_true', help='导出CSV文件') args = parser.parse_args() try: # 初始化爬虫 spider = LianJiaSpider() # 运行爬虫 await spider.run(max_pages=args.pages) except KeyboardInterrupt: logger.info("用户中断程序") sys.exit(0) except Exception as e: logger.error(f"程序运行失败: {e}") sys.exit(1) if __name__ == "__main__": # 安装Playwright浏览器 import subprocess subprocess.run(["playwright", "install", "chromium"]) # 运行主程序 asyncio.run(main())

高级功能扩展

1. 分布式爬虫支持

python

# spiders/distributed_spider.py import redis import json from typing import List import asyncio class DistributedLianJiaSpider(LianJiaSpider): """分布式爬虫""" def __init__(self, redis_url: str = "redis://localhost:6379/0"): super().__init__() self.redis = redis.Redis.from_url(redis_url) self.task_queue = "lianjia:task:queue" self.result_queue = "lianjia:result:queue" async def distribute_tasks(self, urls: List[str]): """分发任务到队列""" for url in urls: task = { 'url': url, 'timestamp': asyncio.get_event_loop().time(), 'retry_count': 0 } self.redis.rpush(self.task_queue, json.dumps(task)) async def consume_tasks(self): """消费任务""" while True: task_json = self.redis.blpop(self.task_queue, timeout=30) if task_json: task = json.loads(task_json[1]) result = await self.process_task(task) if result: self.redis.rpush(self.result_queue, json.dumps(result))

2. 数据监控与可视化

python

# monitor/dashboard.py from fastapi import FastAPI import pandas as pd import plotly.express as px from sqlalchemy import func from models.database import db from models.models import HouseListing app = FastAPI(title="链家数据监控面板") @app.get("/api/stats") async def get_statistics(): """获取统计信息""" async with db.async_session() as session: # 基本统计 total_count = await session.scalar( func.count(HouseListing.id) ) avg_price = await session.scalar( func.avg(HouseListing.unit_price) ) # 区域统计 district_stats = await session.execute( func.count(HouseListing.id).label('count'), func.avg(HouseListing.unit_price).label('avg_price'), HouseListing.district ).group_by(HouseListing.district).all() return { 'total_count': total_count, 'avg_price': round(avg_price, 2) if avg_price else 0, 'district_stats': [ { 'district': stat[2], 'count': stat[0], 'avg_price': round(stat[1], 2) } for stat in district_stats ] }