Python全站链接爬取工具优化-支持过滤和断点续爬

Python全站链接爬取工具优化：支持过滤和断点续爬

标签：#Python #Playwright #爬虫 #AI知识库
日期：2026-05-03
摘要：本文介绍对全站链接爬取工具的优化升级，新增链接过滤、断点续爬、默认不下载文件三个优化点，让工具更加实用和人性化。

前言

上一篇文章发布后，我在实际使用中遇到了一些痛点：

• ❌ 某些网站的静态资源链接（如老旧版本的文档）不需要爬取
• ❌ 爬取过程中发现要排除某些链接，中断后重新开始太浪费时间
• ❌ 默认下载文件会污染本地目录，并且也浪费时间

于是我对工具进行了优化升级，本文分享这些改进。

一、优化点一览

二、优化后的源码

''' 全站站内链接爬取脚本 v2.0 功能：递归爬取指定网站的所有内部链接，支持过滤、断点续爬 '''frombs4importBeautifulSoupfromurllib.parseimporturljoin,urlparsefrommy_playwrightimportMyPlaywrightimportatexitimportCommonimportosdefget_internal_links(base_url:str,filters:set[str]=set())->set[str]:""" 递归爬取全站内部链接 Args: base_url: 目标网站根 URL filters: 需要过滤的 URL 前缀集合 Returns: visited: 所有发现的内部链接集合 """def_crawl_links(url:str)->None:'''爬取指定 URL 的所有内部链接，放到 unvisited 集合中'''print(f"{len(visited)}:{len(unvisited)}[+] 正在抓取:{url}")try:page.goto(url,wait_until='networkidle',timeout=30000)page.wait_for_timeout(500)# 等待确保页面加载完成html=page.content()soup=BeautifulSoup(html,'html.parser')forainsoup.find_all('a',href=True):href=a['href']full_url=urljoin(url,href)parsed=urlparse(full_url)# 过滤非内部链接ifnotparsed.netloc==target_netloc:continueif'#'infull_url:continueifparsed.schemenotin('http','https'):continueiffull_urlinvisited:continue# ⭐ 新增：过滤指定前缀的链接ifany(full_url.startswith(s)forsinfilters):continueunvisited.add(full_url)exceptExceptionase:print(f"[!] 请求失败:{url}-{e}")defon_exit()->None:'''⭐ 异常退出时保存当前状态'''ifunvisited:print('[!] 异常退出，正在保存当前状态到文件...')Common.WriteAllText('tmp_links.txt',str((visited,unvisited)))else:print('[!] 所有链接都被访问了')atexit.register(on_exit)visited:set[str]=set()# 已访问的 URL 集合unvisited:set[str]={base_url}# 未访问的 URL 集合# ⭐ 新增：断点续爬 - 从临时文件恢复状态ifos.path.exists('tmp_links.txt'):print('[!] 从临时文件读取状态...')visited,unvisited=eval(Common.ReadAllText('tmp_links.txt'))# ⭐ 新增：默认不下载文件page=MyPlaywright(headless=True,accept_downloads=False).page target_netloc=urlparse(base_url).netlocwhileunvisited:url=next(iter(unvisited))_crawl_links(url)visited.add(url)unvisited.remove(url)print(f"[✓] 已完成{len(visited)}条链接的爬取")returnvisiteddefsave_to_markdown(links,output_path='internal_links.md'):"""将链接列表保存为 Markdown 文件"""sorted_links=sorted(links)chunk_size=10chunks=[sorted_links[i:i+chunk_size]foriinrange(0,len(sorted_links),chunk_size)]markdown_content=[]markdown_content.append("# 全站内部链接列表\n")markdown_content.append(f"共发现 **{len(links)}** 条链接\n\n")markdown_content.append("---\n\n")foridx,chunkinenumerate(chunks,1):start_num=(idx-1)*chunk_size+1markdown_content.append(f"### 第{start_num}-{start_num+len(chunk)-1}条链接\n\n")forlinkinchunk:markdown_content.append(f"{link}\n")markdown_content.append("\n")withopen(output_path,'w',encoding='utf-8')asf:f.writelines(markdown_content)print(f"[✓] 已保存到:{output_path}")if__name__=='__main__':base_url='https://codemirror.net/'filters={'https://codemirror.net/5/'}# ⭐ 过滤旧版本文档links=get_internal_links(base_url,filters)save_to_markdown(links)

三、优化详解

1️⃣ 链接过滤

# 新增 filters 参数defget_internal_links(base_url:str,filters:set[str]=set())->set[str]:...ifany(full_url.startswith(s)forsinfilters):continue# 跳过过滤的链接

使用示例：

# 过滤 codemirror 旧版本文档filters={'https://codemirror.net/5/'}# 或者过滤多个filters={'https://example.com/api/','https://example.com/docs/v1/',}

2️⃣ 断点续爬

核心思路：用visited和unvisited两个集合分离管理，异常退出时保存状态。

正常执行流程： ┌─────────────────────────────────────┐ │ unvisited = {url1, url2, url3...} │ │ visited = {} │ └─────────────────────────────────────┘ ↓ ┌─────────────────────────────────────┐ │ 取出一个 url1，解析所有子链接 │ │ unvisited = {url2, url3, url4...} │ │ visited = {url1} │ └─────────────────────────────────────┘ ↓ (循环直到 unvisited 为空)

异常恢复流程：

# 程序启动时检查临时文件ifos.path.exists('tmp_links.txt'):print('[!] 从临时文件读取状态...')visited,unvisited=eval(Common.ReadAllText('tmp_links.txt'))

使用场景：

用户：开始爬取... 总数：50 [+] 正在抓取: https://example.com/page45... 总数：51 [+] 正在抓取: https://example.com/page46... ^C 中断 (Ctrl+C) 用户：发现 page46 不需要，过滤掉它 filters = {'https://example.com/page46'} 用户：重新运行... [!] 从临时文件读取状态... [+] 正在抓取: https://example.com/page47... # ⭐ 从断点继续

3️⃣ 默认不下载文件

# 新增 accept_downloads=Falsepage=MyPlaywright(headless=True,accept_downloads=False).page

这样网页中的下载链接就不会触发自动下载，保持工作目录整洁。

四、关键技术点总结

🔑 核心改进：从递归到循环

🔑 状态持久化

# 使用 atexit 注册退出回调atexit.register(on_exit)defon_exit():ifunvisited:Common.WriteAllText('tmp_links.txt',str((visited,unvisited)))

⚠️ 注意：这里使用eval()反序列化有安全风险，生产环境建议用json替代。

五、使用效果

以 CodeMirror 官网为例：

用户：python crawl.py [+] 正在抓取: https://codemirror.net/... [+] 正在抓取: https://codemirror.net/5/... # 自动过滤 [✓] 已完成 128 条链接的爬取 用户：发现不需要 /5/ 版本 filters = {'https://codemirror.net/5/'} 用户：rm internal_links.md && python crawl.py [!] 从临时文件读取状态... [+] 正在抓取: https://codemirror.net/6/... # 从断点继续 [✓] 已完成 118 条链接的爬取

六、后续计划

这个工具将继续迭代，未来计划：

• 📦 封装成命令行工具，支持--url、--filter、--output参数
• 🔧 支持从配置文件读取过滤规则
• 📊 增加进度显示和预估剩余时间
• 🤖 集成到 AI Agent，实现自动化学习流程

七、总结

📌 要点回顾

1. 链接过滤：filters参数支持排除指定前缀的链接
2. 断点续爬：通过visited/unvisited分离 + 临时文件实现
3. 默认不下载：accept_downloads=False保持目录整洁
4. 核心改进：从递归改为循环，状态更可控

📚 相关资源

• 上一篇文章：Python实现全站链接爬取工具：助力打造AI知识库
• 学习人最好的时代：技术学习实践分享

本文为本人原创，首发于掘金。
如果你有任何问题或想法，欢迎在评论区交流！