使用Python3开发一个网站内链检查工具
1. 指定域名站点爬取与站内链接跟踪
- 输入:指定一个域名(如
https://www.lvtao.net
),工具需要抓取该站点的所有页面内容。 - 目标:递归跟踪并访问站点内的所有链接。只有同域名的站内链接(即根域名相同的链接)才会被进一步爬取和分析。
- 处理:对于每一个被访问的页面,提取所有站内链接,并继续抓取这些页面的内容,直到所有的站内链接都被爬取完毕。
2. 忽略特定的地址
- 规则:忽略特定的 URL 前缀。假设你指定的地址是
https://www.lvtao.net/url.html
或/url.htm
开头的 URL,这类链接需要在爬取时被忽略,不做进一步分析。 - 忽略逻辑:在提取和解析链接时,检查每个链接是否以指定的前缀开头,如果是则忽略此链接。
3. 错误链接日志记录
- 日志要求:记录所有不能打开的链接,并明确指出在哪个页面中发现了这个错误链接。
- 输出格式:假设页面
https://www.lvtao.net/a.html
中包含的链接https://www.lvtao.net/b.html
不能打开,日志需要如下记录:Broken link: https://www.lvtao.net/b.html found on page: https://www.lvtao.net/a.html
- 检测内容:不仅仅是状态码的检测,对于网页类链接,需要实际抓取页面内容,确保内容能够正确加载;对于附件(如 PDF、图片等),只需检测其状态码是否正常。
4. 并发控制
- 请求速率限制:为防止对服务器造成过大的压力,需要控制并发请求的速度。每秒最多只能发出 10 个 HTTP 请求。
- 资源链接处理:对图片、视频、PDF 等资源类链接(非网页链接),只需要检查其状态码,不需要进一步解析内容。
5. 网页和附件的处理差异
- 网页链接:如果检测到页面的响应头 Content-Type 为 text/html,表示该链接是一个网页,抓取其 HTML 内容并继续递归检查该页面中的站内链接。
- 附件链接:对于 Content-Type 为附件类型(如 application/pdf、image/jpeg、application/zip 等)的链接,只需检查状态码,不进行内容解析。
这个需求涉及一个爬虫的开发,爬取指定网站的内容,检查站内链接的有效性,同时需要忽略特定的链接并控制并发。我们可以通过Python的 requests、beautifulsoup4、aiohttp 和 asyncio 等库来实现这个工具。为了控制并发和对图片资源进行检测,我们可以使用 aiohttp 来进行异步请求。
具体步骤如下:
- 抓取页面内容:通过 aiohttp 进行异步抓取页面,使用 beautifulsoup4 解析页面内容,并提取站内链接。
- 站内链接跟踪:检查站内的所有链接,并递归处理每个站内页面的链接,直到没有新的页面为止。
- 过滤特定地址:在解析链接时,忽略特定的 URL 模式。
- 日志记录:对于不能打开的链接,记录错误日志,注明具体哪个页面的哪个链接出错。
- 控制并发:通过 asyncio.Semaphore 来限制每秒的请求次数,确保不超过每秒 10 个请求。
- 检测资源:对页面中的图片等资源链接进行状态检测。
直接上代码
import aiohttp
import asyncio
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
import time
# 设置站点和忽略的URL前缀
BASE_URL = "https://www.lvtao.net"
IGNORE_URLS = ["/url.html"] # 忽略的URL模式
MAX_CONCURRENT_REQUESTS = 10 # 每秒请求数限制
# 检查是否是站内链接
def is_internal_link(url):
parsed_url = urlparse(url)
return parsed_url.netloc == "" or parsed_url.netloc == urlparse(BASE_URL).netloc
# 检查是否应该忽略该链接
def should_ignore_link(url):
for ignore in IGNORE_URLS:
if url.startswith(urljoin(BASE_URL, ignore)):
return True
return False
# 获取所有站内链接
def get_internal_links(html, base_url):
soup = BeautifulSoup(html, 'html.parser')
links = set()
for link in soup.find_all('a', href=True):
href = link['href']
full_url = urljoin(base_url, href)
if is_internal_link(full_url) and not should_ignore_link(full_url):
links.add(full_url)
return links
# 记录错误日志
def log_broken_link(page_url, broken_link):
with open('broken_links.log', 'a') as f:
f.write(f"Broken link: {broken_link} found on page: {page_url}\n")
# 请求网页内容,并根据内容类型进行处理
async def fetch(session, page_url, link_url, semaphore):
async with semaphore:
try:
async with session.get(link_url) as response:
# 检查状态码
if response.status != 200:
log_broken_link(page_url, link_url)
return None, False
# 获取响应头中的 Content-Type
content_type = response.headers.get('Content-Type', '').lower()
# 如果是网页内容,则继续处理
if 'text/html' in content_type:
html = await response.text()
return html, True
# 对于附件类的文件,只记录状态,不做进一步处理
else:
return None, True
except Exception as e:
log_broken_link(page_url, link_url)
return None, False
# 处理页面并递归抓取站内链接
async def process_page(session, url, visited, semaphore):
if url in visited:
return
visited.add(url)
html, success = await fetch(session, url, url, semaphore) # 请求页面的内容
if success and html:
links = get_internal_links(html, url) # 获取页面中的站内链接
tasks = [process_link(session, url, link, visited, semaphore) for link in links] # 针对每个站内链接递归处理
await asyncio.gather(*tasks)
# 处理站内的单个链接并递归
async def process_link(session, page_url, link_url, visited, semaphore):
if link_url in visited:
return
visited.add(link_url)
html, success = await fetch(session, page_url, link_url, semaphore) # 检查该链接
if success and html: # 如果是HTML页面,继续解析链接
links = get_internal_links(html, link_url) # 获取该链接页面中的站内链接
tasks = [process_link(session, link_url, link, visited, semaphore) for link in links]
await asyncio.gather(*tasks)
# 主异步函数
async def main():
visited = set()
semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS) # 控制并发量
async with aiohttp.ClientSession() as session:
await process_page(session, BASE_URL, visited, semaphore)
# 运行爬虫
if __name__ == '__main__':
start_time = time.time()
asyncio.run(main())
print(f"Finished in {time.time() - start_time} seconds")
将这个代码保存为check.py
文件
然后我们在命令行下安装扩展模块pip install aiohttp beautifulsoup4 lxml -i https://pypi.mirrors.ustc.edu.cn/simple
然后执行python3 check.py
等它检查即可....
版权声明:本文为原创文章,版权归 全栈开发技术博客 所有。
本文链接:https://www.lvtao.net/dev/python-check-link.html
转载时须注明出处及本声明