目标
https://opendev.org/openstack/neutron 是Openstack 中的一个网络组件项目,这次示例目标是爬取这个项目中所有文件的URL,有了URL再去检索内容,简单的就只剩下写写正则啦。
页面分析
用元素选择器定到目标位置,通过观察发现整个目标内容都包裹在<tbody>
标签中。内容分为两种元素:文件
和目录
,它们都被包裹在<td class="name four wide">
标签中。其中文件的<td>
标签中包含了<span class="octicon-file-directory">
标签,目录的<td>
标签中包含了<span class="octicon-file-text">
标签。如下表:
爷爷标签 | 父标签 | 子标签 |
---|---|---|
tbody | <td class="name four wide"> | <span class="octicon-file-directory"> |
- | - | <span class="octicon-file-text"> |
文件
是末梢,目录
下包含了目录
和文件
。点击文件
超链接能进入内容呈现页面。点击目录
的链接进入到它所包含内容的呈现页面。因此我们只需要一层层不断进入目录中,拿到该目录所包含的所有文件的URL,即算是完成目标。
爬取思路
我们把爬取的目标抽象成树型结构,先假设这是一颗无环的生成树,不对爬到的URL进行重复检测。因此只需要借助队列,对树进行层次遍历即可。
工具
爬网页用的两库已经很熟悉了,除此之外我还需要用到python自带的队列库queue
。我要用到 queue.Queue() 类中提供了三个方法,入队put()
、出队get()
和判队空empty()
。get()
如果在空队列上操作,会一直阻塞等待新元素入队。
from bs4 import BeautifulSoup
import requests
from queue import Queue
我准备设置两个队列,一个是用于暂存URL的cache
,一个用于输出爬取成果的result
。
处理流程如下:
先来看看伪码的实现
def 爬取(url):
html = 访问(url)
for td in html:
if 在<td>中找到了<span class="octicon octicon-file-directory">:
cache.入队(td.url)
elif 在<td>中找到了<span class="octicon octicon-file-text">:
result.入队(td.url)
def 多线程处理result
print(result.出队())
while 队列不为空:
url = 首元素出队
爬取(url)
完整代码:
from queue import Queue
from bs4 import BeautifulSoup
import requests
import threading
cache = Queue()
result = Queue()
link = 'https://opendev.org/openstack/neutron'
cache.put(link)
def find_links(url, queue1, queue2):
try:
html = requests.get(url, timeout=30)
html.raise_for_status
html.encoding = html.apparent_encoding
except:
html.text = ''
soup = BeautifulSoup(html.text, "html.parser")
tds = soup.find('tbody').find_all('td', class_='name four wide')
for td in tds:
if td.find('span', class_='octicon octicon-file-directory') is not None:
# directory enqueue 'cache'
href = 'https://opendev.org' + td.find('a')['href']
queue1.put(href)
elif td.find('span', class_='octicon octicon-file-text') is not None:
# file enqueue 'result'
href = 'https://opendev.org' + td.find('a')['href']
queue2.put(href)
def process_result(queue):
while True:
url = queue.get()
print('Result:{}'.format(url))
threading.Thread(target=process_result, args=(result,)).start()
while not cache.empty():
url = cache.get_nowait()
print('Accessing:{}'.format(url))
find_links(url, cache, result)
处理result队列数据的函数是放在另外一个线程里跑的,面对这种异步的情况,我希望每当队列里的元素全部被取空之后,它就会停下来等待新元素入队。get()方法恰好就是为这种场景设计的,不过这里还有点小问题,因为它无法自己停下来,我放到后面完善这个地方。
处理cache队列的循环是同步的,即每次取一个URL拿给find_links
去跑,跑完之后后再取下一个URL。因此它没有“停-等”的必要,get_nowait()方法恰好就是为这种场景设计的。
完善功能
前面的场景假设有点单纯,实际情况来看,网站中会存超链接指向闭环的情况(环路),还会有许多重复的链接。因此它们看起来更像一张由超连接组成的有向图
。由于我们并不关注它是链接还是被链接(出入度),因此这张图可以化简成无向图
。这样一来爬取整站的问题就变成了图中各节点的遍历。
防止环路爬取的有效办法是设置 visited 标识符、重复URL检测。我准备使用后一种方法,因为用Python实现这个很简单。
图的遍历有两种方法:深度优先
和广度优先
。前者使用递归的方法实现,后者依靠队列。我认为深度优先在处理层级数未知的情况,函数反复递归可能会出现意想不到的情况,并且很耗费内存。广度优先就很有意思了,因为它依靠队列,因此无论是使用外部存储或是增加处理节点,都会变的很容易扩展,再者就实现原理来讲,它更容易理解。
处理流程
前面使用的队列库没有元素重复检测方法,因此我需要重新封装一个队列,实质上它就是一个list。
先看看伪码的实现:
def 爬取(url):
html = 访问(url)
for td in html:
if 在<td>中找到了<span class="octicon octicon-file-directory">:
if cache.重复检测(url):
cache.入队(td.url)
elif 在<td>中找到了<span class="octicon octicon-file-text">:
if result.重复检测(url):
result.入队(td.url)
def 多线程处理result:
print(result.出队())
while 队列不为空:
url = 首元素出队
爬取(url)
完整代码:
from bs4 import BeautifulSoup
import requests
import threading
import time
class LQueue:
def __init__(self):
self._queue = []
self.mutex = threading.Lock()
self.condition = threading.Condition(self.mutex)
def put(self, item):
with self.condition:
self._queue.append(item)
self.condition.notify()
def get(self, timeout=60):
with self.condition:
endtime = time.time() + timeout
while True:
if not self.empty():
return self._queue.pop(0)
else:
remaining = endtime - time.time()
if remaining <= 0.0:
# 等待超时后,返回 None
return None
self.condition.wait(remaining)
def get_nowait(self):
with self.condition:
try:
return self._queue.pop(0)
except:
IndexError('Empty queue')
def has(self, item):
if item in self._queue:
return True
else:
return False
def empty(self):
if self._queue.__len__() == 0:
return True
else:
return False
cache = LQueue()
result = LQueue()
link = 'https://opendev.org/openstack/neutron'
cache.put(link)
single = True
def find_links(url, queue1, queue2):
print('Accessing:{}'.format(url))
try:
html = requests.get(url, timeout=30)
html.raise_for_status
html.encoding = html.apparent_encoding
except:
html.text = ''
soup = BeautifulSoup(html.text, "html.parser")
tds = soup.find('tbody').find_all('td', class_='name four wide')
for td in tds:
if td.find('span', class_='octicon octicon-file-directory') is not None:
# directory enqueue 'cache'
href = 'https://opendev.org' + td.find('a')['href']
if not queue1.has(href):
queue1.put(href)
elif td.find('span', class_='octicon octicon-file-text') is not None:
# file enqueue 'result'
href = 'https://opendev.org' + td.find('a')['href']
if not queue2.has(href):
queue2.put(href)
def process_result(queue):
while True:
# get()方法等待35秒后就会超时,在超时前依然没有新元素进入 cache
# 队列,就说明当所有链接都爬完了。这时候 get() 会返回None,判断返回值
# 如果是None就结束任务。在这设置35秒的超时时间是由于我设置访问URL的超时时间为30。
url = queue.get(35)
if url is None:
break
print('Result:{}'.format(url))
threading.Thread(target=process_result, args=(result,)).start()
while not cache.empty():
url = cache.get_nowait()
find_links(url, cache, result)