不知不覺，已經(jīng)工作6年有余，恍恍惚惚，有機(jī)會滿足房子需求。

在收集房子信息過程中，做些記錄。

貝殼的功能很強(qiáng)大，但很難滿足小區(qū)、距離、教育、面積等多個方面的匹配，使用起來成本仍然較高。

針對以上情況，編寫該項目，收集鏈家的二手房數(shù)據(jù)。項目中，主要根據(jù)價格來篩選小區(qū)，并根據(jù)小區(qū)教育、同工作位置的距離來確定關(guān)注小區(qū)，再通過房子面積、總價、戶型來確定可以選擇的房子 列表，從而將購房精力集中在關(guān)注的重點小區(qū)和房子中。

當(dāng)然，每個人可以根據(jù)自己需求進(jìn)行調(diào)整。

一、基礎(chǔ)環(huán)境說明

1.1 基礎(chǔ)環(huán)境

1.1.1 python

1.1.2 request(加載頁面)

1.1.3 BeautifuSoup(提取信息 )

常用使用例子：

from bs4 import BeautifulSoup

soup = BeautifulSoup(a, "html.parser")

soup.title.text # '標(biāo)題'

# 一、提取標(biāo)簽

# 1.1 提取唯一標(biāo)簽

soup.h1

soup.find('h1')

soup.find_all('h1')[0]

# 1.2 提取多個標(biāo)簽

soup.find_all('h2')

# [

標(biāo)題2

,

標(biāo)題3

]

soup.find_all(['h1','h2'])

# [

標(biāo)題1

,

標(biāo)題2

,

標(biāo)題3

]

# 1.3 使用正則表達(dá)式

import re

soup.find_all(re.compile('^h'))

# [

標(biāo)題1

,

標(biāo)題2

,

標(biāo)題3

]

# 二、匹配屬性

# 2.1 匹配屬性1，直接將屬性名作為參數(shù)名，但是有些屬性不行，比如像a-b這樣的屬性

soup.find_all('p', id = 'p1') # 一般情況

soup.find_all('p', class_='p3') # class是保留字比較特殊，需要后面加一個_

# 2.2 匹配屬性2，最通用的方法

soup.find_all('p', attrs={'class':'p3'}) # 包含這個屬性就算，而不是只有這個屬性

soup.find_all('p', attrs={'class':'p3','id':'pp'}) # 使用多個屬性匹配

soup.find_all('p', attrs={'class':'p3','id':False}) # 指定不能有某個屬性

soup.find_all('p', attrs={'id':['p1','p2']}) # 屬性值是p1或p2

soup.find_all('p', attrs={'class':True}) # 含有class屬性即可

# 2.3 匹配屬性3，正則表達(dá)式匹配

import re

soup.find_all('p', attrs={'id':re.compile('^p')}) # 使用正則表達(dá)式

# 三、根據(jù)標(biāo)簽內(nèi)容文本來識別

# 3.1 匹配標(biāo)簽內(nèi)容1，正則表達(dá)式

import re

soup.find_all('p', text=re.compile('段落'))

soup.find_all('p',text=True)

# 3.2 匹配標(biāo)簽內(nèi)容2，傳入函數(shù)

def nothing(c):

return c not in ['段落1','段落2','文章']

soup.find_all('p',text=nothing)

def has_class_but_no_id(tag):

return tag.has_attr('class') and not tag.has_attr('id')

# 四、提取內(nèi)容

# 4.1 提取標(biāo)簽文本

soup.h.text # 多層嵌套也可以直接返回

soup.h.a.text # 也可以這樣

soup.body.text # 里面有多個內(nèi)容時 '\n標(biāo)題\n段落1\n段落2\n'

# 4.2 其他標(biāo)簽的屬性值

# 提取屬性值，像字典一樣提取，以下兩種方法等價

soup.h.a['href']

soup.h.a.get('href')

# 五、提取標(biāo)簽信息

print(i.name) # 提取標(biāo)簽名

print(i.attrs) # 提取標(biāo)簽所有屬性值

print(i.has_attr('href')) # 檢查標(biāo)簽是否有某屬性

# 五、示例

soup.find('p', attrs={'class':'first'}).text # '文字1'

soup.find_all('p') # [

文字1

,

文字2

], 再分別從中提取文字

soup.find('ul', attrs={'class':'list1'}).find_all('li') # [

列表1第1項, 列表1第2項]

# 代碼參考：https://zhuanlan.zhihu.com/p/35354532

1.1.4 地理位置信息(百度API)

調(diào)用方式1：

def geocodeB(address):

base = "http://api.map.baidu.com/geocoder?address=%s&output=json&key=yourak&city=上海" % address

response = requests.get(base)

if response.status_code == 200:

answer = response.json()

if "location" in answer['result'] and "level" in answer['result']:

return (address,

# round(answer['result']['location']['lng'], 5),

answer['result']['location']['lng'],

# round(answer['result']['location']['lat'], 5),

answer['result']['location']['lat'],

answer['result']["level"])

else:

logger.error("geocodeB %s warning:%s" % (address, answer))

return None

else:

logger.error("geocodeB %s Error" % address)

return None

調(diào)用方式2：

def geocodeB2(address):

from urllib.request import urlopen, quote

from urllib.parse import quote_plus

import hashlib, json

# 以get請求為例http://api.map.baidu.com/geocoder/v2/?address=百度大廈&output=json&ak=yourak

queryStr = '/geocoder/v2/?address=%s&city=上海&output=json&ak=$yourak$' % address

# 對queryStr進(jìn)行轉(zhuǎn)碼，safe內(nèi)的保留字符不轉(zhuǎn)換

encodedStr = quote(queryStr, safe="/:=&?#+!$,;'@()*[]")

# 在最后直接追加上yoursk

rawStr = encodedStr + '$yoursn$'

sn = hashlib.md5(quote_plus(rawStr).encode("utf8")).hexdigest()

url = 'http://api.map.baidu.com%s&sn=%s' % (encodedStr, sn)

req = urlopen(url)

res = req.read().decode() # 將其他編碼的字符串解碼成unicode

answer = json.loads(res) # 對json數(shù)據(jù)進(jìn)行解析

if "location" in answer['result'] and "level" in answer['result']:

return answer['result']['location']['lat'], answer['result']['location']['lng']

else:

logger.error("geocodeB %s warning:%s" % (address, answer))

return None

調(diào)用方式3：

def geocode_by_baidu(address):

from geopy.geocoders import baidu

apikey = '$yourak$' # 從網(wǎng)站申請 http://lbsyun.baidu.com/apiconsole/key?application=key

sn = '$yoursn$'

g = baidu.Baidu(api_key=apikey, security_key=sn, timeout=200)

a = g.geocode(address)

# return (round(a.latitude, 6), round(a.longitude, 6))

return a.latitude, a.longitude

1.1.5 地理獲取距離計算(geopy)

# x and y is (lat,lng)

def get_distance(x, y):

from geopy.distance import geodesic

return round(geodesic(x, y).km, 3)

1.1.6 解決懶加載和滾動加載(selenium)

Selenium是一個用于Web應(yīng)用程序測試的工具。

Selenium測試直接運(yùn)行在瀏覽器中，就像真正的用戶在操作一樣。支持的瀏覽器包括IE（7, 8, 9, 10, 11），Firefox，Safari，Chrome，Opera等。

使用python爬蟲調(diào)用selenium來模擬正常用戶訪問瀏覽器.

1.2 主要問題

1.2.1 懶加載問題

1.2.2 滾動加載問題

1.2.3 IP訪問限制

二、前期準(zhǔn)備

2.1 分析獲取的需求

個人買房需求：

預(yù)算：400萬，最多不超過450萬；

教育：2梯隊學(xué)區(qū)房

戶型：二房以上

房齡：1990年后

面積：60平米以上

交通：離世紀(jì)大道乘公交不超過1小時

===》

1、學(xué)區(qū)情況，根據(jù)小區(qū)攻略的教育評分來過濾，確定小區(qū)范圍

2、根據(jù)a. 小區(qū)的中的房子的價格，使用預(yù)算過濾；b.小區(qū)的位置，通過距離來過濾交通，不滿足 的小區(qū)

3、通過符合要求的小區(qū)列表，來針對每個小區(qū)獲取房子列表，并確定跟蹤重點小區(qū)

特別說明：

1、為什么不直接獲取房子呢？房子無法判斷是否滿足教育；如果通過房子找小區(qū)，再找教育，考慮房子比小區(qū)多出幾個數(shù)量級，會有更多的時間浪費

2、通過預(yù)算和面積需求，可以確定房子的單價，通過單價來篩選小區(qū)，減少小區(qū)范圍。

2.2 分析頁面路徑

2.2.1 獲取小區(qū)列表

1、小區(qū)列表的鏈接分析

由于鏈家僅顯示前100頁內(nèi)容，而整個上海的小區(qū)顯然比100頁更多，故根據(jù)區(qū)來獲取小區(qū)。

其中

bp5ep7.5為價格在5-7.5萬的區(qū)間，bp為begin price；ep為end price。

pg為page的頁面

2、小區(qū)是否有評價的判斷

可以根據(jù)第一步獲取的小區(qū)列表中，查看小區(qū)是否存在小區(qū)攻略標(biāo)簽來判斷是否有小區(qū)評價信息

特別說明：并不是每一個小區(qū)，都可以查看到小區(qū)的教育評分

示例鏈接：https://m.ke.com/sh/xiaoqu/5011000016009/，可以獲取到小區(qū)的整體的評分

2.2.2 根據(jù)小區(qū)，獲取攻略

小區(qū)的攻略地址為：

https://m.ke.com/sh/xiaoqu/5011000016009/gonglueV2.html?click_source=m_resblock_detail#review

對于小區(qū)，有總體評分和分項評分，其中分項評分包含建筑品質(zhì)、戶型設(shè)計、交通條件、教育質(zhì)量、商業(yè)環(huán)境、花園景觀、物業(yè)管理等評分。

每個人可以根據(jù)自己的需求，使用不同的評分項進(jìn)行小區(qū)過濾。

例如，我優(yōu)先考慮教育，則以教育條件進(jìn)行主要過濾條件，要求教育8分以上，而其他的要求6.5分以上。

2.2.3 根據(jù)小區(qū)，獲取房子列表

小區(qū)的房子列表：

增加過濾：https://m.ke.com/sh/ershoufang/bp350ep450l2l3ba67ea70c5011000016009

其中，bp350ep450為價格區(qū)間為350-450萬；l2l3戶型為2室或3室；ba67ea70為面積在67-70平米；c5011000016009 為小區(qū)編號為5011000016009。

三、項目代碼實現(xiàn)

3.1 獲取小區(qū)

def get_xiaoqu_list(self, area, save_path):

page_size = 100

# 由于僅收集上海，故未對多城市處理

fieldnames = ['area', 'page', 'xiaoqu_id', 'url', 'name', "brief", "loc", "build_type", "build_year", "price",

"have_gonglue"]

# 如果不存在，則創(chuàng)建一個空CSV文件，包含表頭

# 如果已存在，則將記錄已處理的記錄情況（針對IP限制，需要跑多次情況）

handled_list = []

if os.path.isfile(save_path):

with open(save_path, encoding='utf-8-sig') as csvfile:

reader = csv.DictReader(csvfile)

for row in reader:

handled_list.append("%s_%s" % (row['area'], row['page']))

else:

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

writer.writeheader()

handled_set = set(handled_list)

logger.info(

"get_xiaoqu_list, have handled:%s " % (len(handled_set)))

# 針對上海各區(qū)進(jìn)行處理

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

for page_num in range(1, page_size):

# https://m.ke.com/sh/xiaoqu/pudong/pb4ep4.5pg10/

url = "https://m.ke.com/sh/xiaoqu/%s/bp5ep7.5pg%s/" % (area, str(page_num))

if "%s_%s" % (area, page_num) in handled_set:

logger.info("%s has been handled." % url)

continue

else:

logger.info(url)

# 獲取頁面內(nèi)容

r = requests.get(url=url, headers=self.page_headers)

html = r.content

lj = BeautifulSoup(html, 'html.parser')

page_items = lj.find_all('li', attrs={'class': 'pictext'})

# 解析各頁中的小區(qū)列表

if len(page_items) > 0:

for item in page_items:

xiaoqu_url = item.a.get('href')

xiaoqu_id = xiaoqu_url.split("/")[-2]

xiaoqu_gonglue = item.find_all("p", attrs={"class": "gonglue_title"})

if len(xiaoqu_gonglue) == 0:

is_gonglue = 0

else:

is_gonglue = 1

xiaoqu_info = item.find_all("div", attrs={"class": "item_list"})[0]

xiaoqu_name = xiaoqu_info.find_all("div", attrs={"class": "item_main"})[0].string

xiaoqu_brief = xiaoqu_info.find_all("div", attrs={"class": "item_other"})[0].string.strip(

"\n\r \"")

xiaoqu_brief = " ".join(xiaoqu_brief.split())

xiaoqu_loc = xiaoqu_brief.split()[0]

build_type = xiaoqu_brief.split()[1]

build_year = re.search(r' (?P\d{1,})年建成', xiaoqu_brief, re.I)

if build_year:

xiaoqu_build = build_year.group("build_year")

else:

xiaoqu_build = ""

xiaoqu_price = xiaoqu_info.find_all("span", attrs={"class": "price_total"})[0].em.string

xiaoqu_dict = {

"area": area,

"page": page_num,

"xiaoqu_id": xiaoqu_id,

"url": xiaoqu_url,

"name": xiaoqu_name,

"brief": xiaoqu_brief,

"loc": xiaoqu_loc,

"build_type": build_type,

"build_year": xiaoqu_build,

"price": xiaoqu_price,

"have_gonglue": is_gonglue

}

writer.writerow(xiaoqu_dict)

else:

# 表面已到最后一頁

break

handled_set.update({"%s_%s" % (area, page_num)})

3.2 根據(jù)小區(qū)列表，獲取包含攻略的小區(qū)

3.2.1 根據(jù)單個頁面獲取小區(qū)詳細(xì)信息

# 根據(jù)指定小區(qū)的id，獲取小區(qū)的攻略信息

def get_xiaoqu_gonglue_dict(self, id):

url = "https://m.ke.com/sh/xiaoqu/%s/gonglueV2.html?click_source=m_resblock_detail#review" % id

logger.info(url)

# 根據(jù)url加載頁面

# https://m.ke.com/sh/xiaoqu/5011000007603/gonglueV2.html?click_source=m_resblock_detail#review

html = requests.get(url=url, headers=self.page_headers).content

lj = BeautifulSoup(html, 'html.parser')

loc_node = lj.find('div', attrs={'class': 'head_location'})

if loc_node is not None:

loc_name = loc_node.string

else:

loc_name = ""

cpt_content = lj.find_all('div', attrs={'id': 'review'})[0]

totoal_score = cpt_content.find('div', attrs={'class': "review_score"}).get_text().replace("綜合測評得分", "")

review_txt = ""

if cpt_content.find('div', attrs={'class': "review_txt_box"}) is not None:

review_txt = cpt_content.find('div', attrs={'class': "review_txt_box"}).get_text().strip(" \n\r")

review_list_txt = cpt_content.find('ul', attrs={'class': "review_list"})

review_list = review_list_txt.find_all('li')

other = ""

jianzhu_score = huxing_score = jiaotong_score = shangye_score = jiaoyu_score = jingguan_score = wuye_score = ""

for item in review_list:

key = item.span.string

value = item.progress.get('value')

if key == "建筑品質(zhì)":

jianzhu_score = value

elif key == "戶型設(shè)計":

huxing_score = value

elif key == "交通條件":

jiaotong_score = value

elif key == "教育質(zhì)量":

jiaoyu_score = value

elif key == "商業(yè)環(huán)境":

shangye_score = value

elif key == "花園景觀":

jingguan_score = value

elif key == "物業(yè)管理":

wuye_score = value

else:

other = " %s:%s " % (key, value)

peitao_node = lj.find('div', attrs={"class": "box peitao card_box"})

map_api_node = peitao_node.find('img') if peitao_node is not None else None

if map_api_node is not None:

map_api = map_api_node.get('src')

else:

map_api = ""

def get_geo_from_mapapi(map_api):

geo = re.search(r'center=(?P[\d.]+),(?P[\d.]+)', map_api, re.I)

if geo:

lat = geo.group("lat")

lng = geo.group("lng")

else:

lat = lng = None

return lat, lng

lat, lng = get_geo_from_mapapi(map_api)

gonglue_dict = {

"xiaoqu_id": id,

"loc_name": loc_name,

"total_score": totoal_score,

"review_txt": review_txt if review_txt is not None else "",

"jianzhu_score": jianzhu_score if jianzhu_score is not None else "",

"huxing_score": huxing_score if huxing_score is not None else "",

"jiaotong_score": jiaotong_score if jiaotong_score is not None else "",

"jiaoyu_score": jiaoyu_score if jiaoyu_score is not None else "",

"shangye_score": shangye_score if shangye_score is not None else "",

"jingguan_score": jingguan_score if jingguan_score is not None else "",

"wuye_score": wuye_score if wuye_score is not None else "",

"map_api": map_api,

"lng": lng if lng is not None else "",

"lat": lat if lat is not None else "",

"other": other

}

return gonglue_dict

3.2.2 根據(jù)列表，生成所有攻略信息列表

# 根據(jù)第一步獲取的小區(qū)列表，逐項生成攻略列表

def handle_gonglue_by_xiaoqu(self, file_path, save_path, if_distance=False, local_geo=None):

# 判斷參數(shù)是否正確

if if_distance == True and local_geo is None:

logger.error("in handle_gonglue_by_xiaoqu, if_distance's Ture, local_geo can't be None")

exit(1)

# 生成小區(qū)列表

url_list = []

with open(file_path, encoding='utf-8-sig') as csvfile:

reader = csv.DictReader(csvfile)

for row in reader:

if row['have_gonglue'] == "1":

url_list.append(row['xiaoqu_id'])

# 如果攻略列表已存在，則統(tǒng)計已處理的記錄

handled_list = []

fieldnames = ['xiaoqu_id', 'loc_name', 'total_score', "review_txt", "jianzhu_score", "huxing_score",

"jiaotong_score", "jiaoyu_score", "shangye_score", "jingguan_score", "wuye_score",

"map_api", "lat", "lng", "distance", "other"]

if os.path.isfile(save_path):

with open(save_path, encoding='utf-8-sig') as csvfile:

reader = csv.DictReader(csvfile)

for row in reader:

handled_list.append(row['xiaoqu_id'])

else:

# 如果不存在，則創(chuàng)建一個空CSV文件，包含表頭

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

writer.writeheader()

handled_set = set(handled_list)

logger.info("handle_gonglue_by_xiaoqu, the length of url_list: %s" % len(url_list))

# 針對每一個小區(qū)列表，獲取攻略信息

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

for xiaoqu_id in url_list:

if xiaoqu_id not in handled_set:

gonglue_dict = self.get_xiaoqu_gonglue_dict(id=xiaoqu_id)

if if_distance:

distance = get_distance((gonglue_dict["lat"], gonglue_dict["lng"]), local_geo)

gonglue_dict["distance"] = distance

writer.writerow(gonglue_dict)

handled_set.update({xiaoqu_id})

else:

logger.info("xiaoqu %s is handled" % xiaoqu_id)

3.3 根據(jù)攻略列表，生成關(guān)注的房子列表

3.3.1 獲取單個小區(qū)的房子列表

# 根據(jù)小區(qū)id，獲取小區(qū)的滿足條件的房子列表

def get_houselist_by_xiaoqu(self, xiaoqu_id):

# https://m.ke.com/sh/ershoufang/bp350ep450l2l3ba67ea70c5011000009590

# bp350ep450 表示價格開始和結(jié)束

# l2l3 戶型2室和3室

# ba67ea70 面積67-70

# c5011000009590 小區(qū)編號

url = "https://m.ke.com/sh/ershoufang/bp350ep450l2l3ba60ea90c%s" % xiaoqu_id

html = requests.get(url=url, headers=self.page_headers).content

house_list = []

lj = BeautifulSoup(html, 'html.parser')

# 頁面中包含多個列表，包含當(dāng)前搜索以及推薦其他小區(qū)

view_body = lj.find('div', attrs={'class': 'list-view-section-body'})

item_list = view_body.find_all('div', attrs={'class': 'lj-track', 'data-click-event': 'SearchClick'})

for item in item_list:

house_body = item.find("div", attrs={'class': 'kem__house-tile-ershou'})

house_id = house_body.get("data-id")

logger.info("handle house_id:%s" % house_id)

house_txt = house_body.find("div", attrs={'class': 'house-text'})

house_title = house_txt.find("div", attrs={"class": 'house-title'}).text

house_desc = house_txt.find("div", attrs={"class": 'house-desc'}).string

house_price_total = house_txt.find("span", attrs={"class": "price-total"}).strong.string

house_price_unit = house_txt.find("span", attrs={"class": "price-unit"}).string.strip("元/平")

house_dict = {

"xiaoqu_id": xiaoqu_id,

"house_id": house_id,

"title": house_title,

"desc": house_desc,

"price_total": house_price_total,

"price_unit": house_price_unit

}

house_list.append(house_dict)

return house_list

3.3.2 根據(jù)攻略列表，生成房子列表

# 根據(jù)攻略列表，提取關(guān)注的小區(qū)，再逐項獲取列表

def handle_hoselist_by_gonglue(self, file_path, save_path, filter_func=None):

xiaoqu_list = []

with open(file_path, encoding='utf-8-sig') as csvfile:

reader = csv.DictReader(csvfile)

for row in reader:

if filter_func is not None:

if filter_func(row):

# 將小區(qū)的ID，加入到處理列表中

xiaoqu_list.append((row["xiaoqu_id"], row["loc_name"], row["distance"]))

else:

xiaoqu_list.append((row["xiaoqu_id"], row["loc_name"], row["distance"]))

handled_list = []

fieldnames = ['xiaoqu_id', 'xiaoqu_name', 'distance', 'house_id', 'title', "desc", "price_total", "price_unit"]

if os.path.isfile(save_path):

with open(save_path, encoding='utf-8-sig') as csvfile:

reader = csv.DictReader(csvfile)

for row in reader:

handled_list.append(row['xiaoqu_id'])

else:

# 如果不存在，則創(chuàng)建一個空CSV文件，包含表頭

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

writer.writeheader()

handled_set = set(handled_list)

logger.info(

"handle_hoselist_by_xiaoqu, to be handled: %s, have handled:%s " % (len(xiaoqu_list), len(handled_set)))

with open(save_path, "a+", newline='\n', encoding='utf-8-sig') as csvfile:

writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

for xiaoqu_id, xiaoqu_loc_name, distance in xiaoqu_list:

if xiaoqu_id not in handled_set:

logger.info("handle xiaoqu:%s" % xiaoqu_id)

house_list = self.get_houselist_by_xiaoqu(xiaoqu_id)

if len(house_list) > 0:

for house_dict in house_list:

house_dict["xiaoqu_name"] = xiaoqu_loc_name

house_dict["distance"] = distance

writer.writerow(house_dict)

else:

house_dict = {

"xiaoqu_id": xiaoqu_id,

"xiaoqu_name": xiaoqu_loc_name,

"distance": distance

}

writer.writerow(house_dict)

logger.info("小區(qū)：%s %s have no match house." % (xiaoqu_id, xiaoqu_loc_name))

handled_set.update({xiaoqu_id})

else:

logger.info("%s is handled" % xiaoqu_id)

總結(jié)

以上是生活随笔為你收集整理的python爬虫requests源码链家_python的爬虫项目(链家买二手房）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。