使用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 来进行异步请求。

具体步骤如下:

  1. 抓取页面内容:通过 aiohttp 进行异步抓取页面,使用 beautifulsoup4 解析页面内容,并提取站内链接。
  2. 站内链接跟踪:检查站内的所有链接,并递归处理每个站内页面的链接,直到没有新的页面为止。
  3. 过滤特定地址:在解析链接时,忽略特定的 URL 模式。
  4. 日志记录:对于不能打开的链接,记录错误日志,注明具体哪个页面的哪个链接出错。
  5. 控制并发:通过 asyncio.Semaphore 来限制每秒的请求次数,确保不超过每秒 10 个请求。
  6. 检测资源:对页面中的图片等资源链接进行状态检测。

直接上代码

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
等它检查即可....

标签: Python

相关文章

图片Base64编码

CSR生成

图片无损放大

图片占位符

Excel拆分文件