WSWP(用python寫網路爬蟲)筆記 一:實現簡單爬蟲
wswp中的程式碼是通過python2的語法來寫的,在學習的過程中個人比較喜歡python3,因此準備將wswp的示例程式碼用python3重寫一遍,以加深映像。
開始嘗試構建爬蟲
識別網站所用技術和網站所有者
構建網站所使用的技術型別的識別和尋找網站所有者很有用處,比如web安全滲透測試中資訊收集的環節對這些資訊的收集將對後續的滲透步驟有很重要的作用。對於爬蟲來說,識別網站所使用的技術和網站所有者雖然不是很重要,但也能從中獲取到很多資訊。
檢查構建網站的技術型別可通過一個很有用的模組builtwith中的函式來實現。
pip install builtwith
需要注意的是,python3中安裝好builtwith以後需對builtwith的__init__.py檔案進行糾錯處理,寫一個測試程式碼根據程式碼執行報錯進行修改就行了。修改完成後,便能使用其中的函式進行解析處理了。
import builtwith
print(builtwith.parse('http://example.webscraping.com')
結果如下:
尋找網站所有者可通過python-whois模組中的函式進行解析。
pip install python-whois
測試程式碼如下:
import whois
print(whois.whois('example.webscraping.com')
結果如下:
第一個爬蟲
python3中將urllib2模組的函式整合到了urllib模組中,通過不同的分類進行更細化的管理。
下載網頁
爬蟲進行網頁資料爬取的第一個步驟便是先將網頁下載下來。
# download_v1
import urllib.request
def download(url):
return urllib.request.urlopen(url).read()
當傳入需要下載的url時,上述函式可將對應的網頁下載並返回其html。但是上述函式並沒有對可能遇到的異常情況進行處理,比如頁面不存在,為了避免這些異常,改進版如下:
# downloader.py
# download_v2
import urllib.request
import urllib.error
def download(url):
print("正在下載:" , url)
try:
html = urllib.request.urlopen(url).read()
except urllib.error.URLError as e:
print("下載錯誤:", e.reason)
html = None
return html
將上述程式碼寫進一個檔案,將此檔案當做模板使用即可進行測試。
通過mian檔案引入上述函式,對一個不存在的網頁進行訪問:
# main.py
from downloader import download
if __name__ == '__main__':
url = 'http://www.freebuf.com/articles/rookie/151327.html'
html = download(url)
print(html)
結果如下:
重試下載
爬蟲在爬取資料時遇到某些錯誤是臨時性的,比如 503 Service Unavailable錯誤。對於臨時錯誤可通過嘗試重新下載。
新增重試下載功能的下載函式如下:
# downloader.py
# download_v3
import urllib.request
import urllib.error
def download(url, num_retries=3):
print("正在下載:", url)
try:
html = urllib.request.urlopen(url).read()
except urllib.error.URLError as e:
print("下載錯誤:", e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500<=e.code<600:
# 只對5xx錯誤進行重新下載嘗試
return download(url, num_retries - 1)
return html
設定使用者代理
一些網站會對python使用的預設代理進行禁封,因此需要控制使用者代理的控制。
# downloader.py
# download_with_useragent
import urllib.request
import urlllib.error
def download(url, user_agent='wswp', num_retries=2):
print('正在下載:', url)
headers = {'User-agent':user-agent}
request = urllib.request.Request(url, headers=headers)
try:
html = urllib.request.urlopen(request).read()
except urllib.error.URLError as e:
print('下載錯誤:', e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500<=e.code<600:
# 針對5xx錯誤進行重新下載
return download(url, user_agent, num_retries - 1)
return html
連結爬蟲
通過跟蹤所有連結的方式,可以很容易的下載整個網頁的頁面,通過正則表示式確定哪些頁面需要下載。
# linkCrawler.py
import re
from downloader import download
def get_links(html):
"""
返回html中所有的連結
"""
webpage_regex = re.compile('<a\sherf="(.*?)"', re.IGNORECASE)
# Python3中預設返回的資料為bytes,需要轉換為str
return webpage_regex.findall(str(html))
def linkCrawler(seed_url, link_regex):
"""
seed_url: 需要爬取的第一個url
link_regex: 連結匹配的正則表示式
"""
crawl_queue = [seed_url]
while crawl_queue:
url = crawl_queue.pop()
html = download(url)
for link in get_links(html):
if re.match(link_regex, link):
crawl_queue.append(link)
測試結果如下:
可以看出當下載 ‘/places/default/index/1’這個連結的時候出錯了,因為這只是連結的路徑部分,沒有協議和伺服器部分,也就是說這是一個相對連線。如果需正確的爬取,則需要將此連結轉換為絕對連結的形式。
#linkCrawlwer.py
# linkCrawler_v2
# 在上一步的函式中修改如下內容
# 新增
import urllib.parse
def linkCrawler(seed_url, link_regex):
crawl_queue = [seed_url]
while crawl_queue:
url = crawl_queue.pop()
html = download(url)
# filter for links matching our regular expression
for link in get_links(html):
if re.match(link_regex, link):
mLink = urllib.parse.urljoin(seed_url, link)
crawl_queue.append(mLink)
測試結果如下:
由結果可以看出,出現了重複的爬取。為什麼呢?因為在這些連結中相互之間存在連結。為了避免重複爬取相同發的連結,需要去重處理。修改後的linkCrawler函式如下:
def linkCrawler(seed_url, link_regex):
crawl_queue = [seed_url]
# 通過set型別進行去重處理
seen = set(crawl_queue)
while crawl_queue:
url = crawl_queue.pop()
html = download(url)
for link in get_links(html):
if re.match(link_regex, link):
link = urllib.parse.urljoin(seed_url, link)
if link not in seen:
seen.add(link)
crawl_queue.add(link)
到此為止,第一個簡單的爬蟲已經實現了,雖然功能還有點簡陋,只是爬取網頁的連結,在後面通過新增功能便能使這個爬蟲的功能更加強大。
高階功能
解析robots.txt
通過解析robots.txt檔案,避免下載禁止爬取的URL,通過python自帶的robotparser模組就能輕鬆完成這項工作。
import urllib.robotparser
rp = urllib.robotparser.RobotFileParser()
rp.set_url('http://example.webscraping.com/robotx.txt')
print(rp.reader())
# 可通過can_fentch()函式確定制定的使用者代理是否允許訪問網頁
url = 'http://example.webscraping.com'
user_agent = 'BadCrawler'
print(rp.can_fetch(user_agent, url))
如果希望爬蟲只爬取robots.txt中規定可訪問的頁面,則可以在linkCrawler中獲取link的時候進行判斷。
支援代理
有時候訪問的網站需要通過代理才能訪問。新增代理引數的download函式如下:
# downloader.py
import urllib.request
import urllib.error
import urllib.parse
def download(url, user_agent='wswp', proxy=None, num_retries=2):
print("Downloading:", url)
headers = {'User-agent': user_agent}
request = urllib.request.Request(url,headers=headers)
opener = urllib.request.build_opener()
if proxy:
proxy_params = { urllib.parse.urlparse(url).scheme: proxy}
opener.add_handler(urllib.request.ProxyHandler(proxy_params))
try:
#html = urlopen(request).read()
html = opener.open(request).read()
except urllib.error.URLError as e:
print("Download Error:", e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500 <= e.code < 600:
download(url,user_agent, proxy, num_retries - 1)
return html
下載限速
如果爬取網站的速度過快,可能會面臨被禁止訪問或者是伺服器過載的情況。為了降低這些風險,可在兩次下載之間新增延時。
# Throttle.py
import urllib.parse
import datetime
import time
class Throttle:
"""
對一個域名訪問時在前後兩次訪問之間新增一個延時。
"""
def __init__(self, delay):
self.delay = delay
# 同一個域名的上一次訪問時間
self.domains = {}
def wait(self, url):
domain = urllib.parse.urlparse(url).netloc
last_accessed = self.domains.get(domain)
if self.delay > 0 and last_accessed is not None:
sleep_secs = self.delay - (datetime.datetime.now() - last_accessed).seconds
if sleep_secs > 0:
# 域名剛被訪問過,需要延時
time.sleep(sleep_secs)
# 更新上次訪問時間
self.domains[domain] = datetime.datetime.now()
避免爬蟲死迴圈
通過新增一個引數記錄當前網頁經歷過了多少個連結 – 深度。當達到最大深度時,爬蟲就不再向連結佇列中新增網頁中的連結了。修改linkCrawler中的seen變數,修改為一個字典型別,增加頁面深度的記錄。
def linkCrawler(seed_url, link_regex=None, max_depth= 2, ...):
crawl_queue = set([seed_url])
# 通過set型別進行去重處理
seen = {seed_url:0}
...
while crawl_queue:
url = crawl_queue.pop()
# 檢查url是否可以被採集
if rp.can_fetch(user_agent, url):
throttle.wait(url)
html = ...
depth = seen[url]
if depth != max_depth:
if link_regex:
links.extend(link for link in get_links(html) if re.match(link_regex, link))
for link in links:
link = normalize(seed_url, link)
if link not in seen:
seen[link] = depth + 1
crawl_queue.append(link)
綜合以上功能的爬蟲
根據以上內容的學習,已經可以大體構建出一個linkCrawler爬蟲的架構,大體可分為下載器模組和連結獲取模組兩個大模組。整合完成的終極版本程式碼如下:
"""
downloader.py
"""
import urllib.request
import urllib.parse
import urllib.error
def download(url, headers, proxy, numRetries, data=None):
print("正在下載:", url)
request = urllib.request.Request(url, data, headers)
opener = urllib.request.build_opener()
if proxy:
proxyParams = { urllib.parse.urlparse(url).scheme: proxy }
opener.add_handler(urllib.request.ProxyHandler(proxyParams))
try:
response = opener.open(request)
html = response.read()
code = response.code
except urllib.error.URLError as e:
print('下載錯誤:', e.reason)
html = ''
if hasattr(e, 'code'):
code = e.code
if numRetries > 0 and 500<=code<600:
return download(url, headers, proxy, numRetries - 1, data)
else:
code = None
return html
連結獲取模組linkCrawler.py:
import re
import urllib.parse
import urllib.robotparser
from collections import deque
from .Throttle import Throttle
from .downloader import download
def getRobots(url):
"""
Initial robots parser for this domain
:param url:
:return:
"""
rp = urllib.robotparser.RobotFileParser()
rp.set_url(urllib.parse.urljoin(url, '/robots.txt'))
rp.read()
return rp
def normalize(seedUrl, link):
"""
Normalize this URL by removing hash and adding domain
:param seedUrl:
:param link:
:return:
"""
link, _ = urllib.parse.urldefrag(link) # remove hash to avoid duplicates
return urllib.parse.urljoin(seedUrl, link)
def sameDomain(url1, url2):
"""
Return True if both URL's belong to same domain
:param url1:
:param url2:
:return:
"""
return urllib.parse.urlparse(url1).netloc == urllib.parse.urlparse(url2).netloc
def getLinks(html):
"""
Return a list of links from html
:param html:
:return:
"""
webpageRegex = re.compile('<a\shref="(.*?)"',re.IGNORECASE)
return webpageRegex.findall(str(html))
def linkCrawler(seedUrl, linkRegx=None, delay=5, maxDepth=-1, maxUrls=-1, headers=None, userAgent='wswp', proxy=None, numRetries=1):
"""
Crawl from the given seed URL following links matched by linkRegex
:param seedUrl: 起始url
:param linkRegx: 連結匹配的正則表示式
:param delay: 延遲時間
:param maxDepth: 最深的層次
:param maxUrls: 最多的url數量
:param headers: http請求頭
:param userAgent: http頭中的userAgent選項
:param proxy: 代理地址
:param numRetries: 重新下載次數
:return:
"""
crawlQueue = deque([seedUrl])
seen = { seedUrl:0}
numUrls = 0
rp = getRobots(seedUrl)
throttle = Throttle(delay)
headers = headers or {}
if userAgent:
headers['User-agent'] = userAgent
while crawlQueue:
url = crawlQueue.pop()
if rp.can_fetch(userAgent, url):
throttle.wait(url)
html = download(url, headers, proxy=proxy, numRetries=numRetries)
links = []
depth = seen[url]
if depth != maxDepth:
if linkRegx:
links.extend(link for link in getLinks(html) if re.match(linkRegx, link))
for link in links:
link = normalize(seedUrl, link)
if link not in seen:
seen[link] = depth + 1
if sameDomain(seedUrl, link):
crawlQueue.append(link)
numUrls += 1
if numUrls == maxUrls:
break
else:
print('Blocked by robots.txt')
延時模組Throttle.py:
import time
import datetime
import urllib.parse
class Throttle:
"""
Throttle downloading by sleeping between requests to same domain
"""
def __init__(self, delay):
# Amount of delay between downloads for each domain
self.delay = delay
# timestamp of when a domain was last accessed
self.domain = {}
def wait(self, url):
domain = urllib.parse.urlparse(url).netloc
lastAccessed = self.domain.get(domain)
if self.delay > 0 and lastAccessed is not None:
sleepSec = self.delay - (datetime.datetime.now() - lastAccessed).seconds
if sleepSec > 0:
time.sleep(sleepSec)
self.domain[domain] = datetime.datetime.now()
測試程式碼main.py:
from .linkCrawler import linkCrawler
if __name__ == "__main__":
linkCrawler('http://example.webscraping.com', '.*?/(index/view)', delay=3,numRetries=3,userAgent='BadCrawler')
linkCrawler('http://example.webscraping.com', '.*?/(index/view)', delay=3, numRetries=3, maxDepth=1,userAgent='GoodCrawler')