用Python爬虫+Scapy抓包，手把手教你从零搭建一个自己的期末复习资料库（附完整代码）

用Python构建智能复习资料库：爬虫与抓包技术实战

期末考试季总是让人手忙脚乱，各种讲义、习题和PPT散落在不同平台，手动收集效率低下。作为计算机专业学生，我们完全可以用技术手段解决这个问题——构建一个自动化、智能化的个人复习资料库。本文将带你从零开始，整合Python爬虫与网络抓包技术，打造一个能自动收集、分类存储学习资料的系统。

1. 项目规划与技术选型

在开始编码前，明确项目目标和选择合适的技术栈至关重要。我们的复习资料库需要实现三个核心功能：资源自动收集、内容解析存储和反爬策略应对。

技术栈对比分析：

选择Requests+BeautifulSoup+Scapy组合，既能覆盖大多数学术资源网站，又能应对基础反爬措施。对于特别复杂的动态网站，可以后续引入Selenium作为补充。

提示：在实际开发中，建议先从最简单的Requests开始，遇到障碍再逐步引入更复杂的技术，避免过早优化。

2. 基础爬虫框架搭建

让我们先构建一个稳健的爬虫框架，包含请求管理、异常处理和基础解析功能。模块化设计将使后续扩展更加容易。

import requests from bs4 import BeautifulSoup import time import os from urllib.parse import urljoin, urlparse class AcademicSpider: def __init__(self, base_url, output_dir="materials"): self.base_url = base_url self.output_dir = output_dir self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Accept-Language': 'en-US,en;q=0.9' }) self.visited_urls = set() self.create_output_dir() def create_output_dir(self): if not os.path.exists(self.output_dir): os.makedirs(self.output_dir) os.makedirs(os.path.join(self.output_dir, 'pdfs')) os.makedirs(os.path.join(self.output_dir, 'ppts')) os.makedirs(os.path.join(self.output_dir, 'html')) def is_valid_url(self, url): parsed = urlparse(url) return bool(parsed.netloc) and parsed.netloc in self.base_url def download_file(self, url, file_type): try: response = self.session.get(url, stream=True, timeout=10) response.raise_for_status() filename = os.path.basename(urlparse(url).path) or f"file_{int(time.time())}" save_path = os.path.join(self.output_dir, file_type, filename) with open(save_path, 'wb') as f: for chunk in response.iter_content(1024): f.write(chunk) return True except Exception as e: print(f"下载失败 {url}: {e}") return False def extract_links(self, url): try: response = self.session.get(url, timeout=8) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') # 识别并下载资源文件 for link in soup.find_all('a', href=True): href = link['href'].lower() if href.endswith('.pdf'): pdf_url = urljoin(url, link['href']) self.download_file(pdf_url, 'pdfs') elif href.endswith(('.ppt', '.pptx')): ppt_url = urljoin(url, link['href']) self.download_file(ppt_url, 'ppts') # 返回页面上的所有链接供进一步爬取 return [urljoin(url, a['href']) for a in soup.find_all('a', href=True) if self.is_valid_url(urljoin(url, a['href']))] except Exception as e: print(f"解析页面失败 {url}: {e}") return [] def crawl(self, start_url, max_depth=2): queue = [(start_url, 0)] while queue: url, depth = queue.pop(0) if url in self.visited_urls or depth > max_depth: continue print(f"正在抓取: {url} (深度 {depth})") self.visited_urls.add(url) time.sleep(1) # 礼貌性延迟 links = self.extract_links(url) queue.extend((link, depth+1) for link in links if link not in self.visited_urls)

这个基础框架已经具备了几个关键特性：

• 会话管理：使用requests.Session保持连接，提高效率
• 资源识别：自动检测并下载PDF和PPT文件
• 礼貌爬取：设置1秒间隔，避免对服务器造成负担
• 广度优先搜索：控制爬取深度，防止无限递归
• 异常处理：对各种网络错误进行捕获和记录

3. 使用Scapy分析网络请求

当遇到反爬机制时，我们需要更深入地理解网站如何运作。Scapy可以帮助我们分析原始网络流量，识别关键API请求和验证机制。

from scapy.all import sniff, IP, TCP, Raw from collections import defaultdict import json class TrafficAnalyzer: def __init__(self, target_domain): self.target_domain = target_domain self.api_endpoints = defaultdict(int) self.auth_headers = set() def packet_handler(self, packet): if IP in packet and TCP in packet: ip_pkt = packet[IP] tcp_pkt = packet[TCP] # 只分析目标域名的流量 if self.target_domain in ip_pkt.dst: payload = bytes(tcp_pkt.payload) try: # 尝试解析HTTP请求 if b'HTTP' in payload[:20]: http_lines = payload.decode('utf-8', errors='ignore').split('\r\n') request_line = http_lines[0] # 记录API端点 if any(method in request_line for method in ['GET', 'POST', 'PUT']): path = request_line.split(' ')[1] self.api_endpoints[path] += 1 # 提取认证头信息 for line in http_lines[1:]: if line.startswith(('Authorization:', 'X-API-Key:')): self.auth_headers.add(line) except UnicodeDecodeError: pass def start_capture(self, duration=60): print(f"开始捕获 {self.target_domain} 的流量...") sniff(filter=f"tcp and host {self.target_domain}", prn=self.packet_handler, timeout=duration) def generate_report(self): report = { "most_frequent_endpoints": sorted(self.api_endpoints.items(), key=lambda x: x[1], reverse=True)[:5], "authentication_headers": list(self.auth_headers) } with open('traffic_report.json', 'w') as f: json.dump(report, f, indent=2) return report # 使用示例 analyzer = TrafficAnalyzer("example.edu") analyzer.start_capture(120) report = analyzer.generate_report()

这个流量分析器能帮助我们：

1. 识别网站最频繁访问的API端点
2. 发现隐藏的认证机制
3. 理解网站的数据加载方式
4. 找出可能用于反爬的请求头

注意：在实际使用中，请确保你有权限监控网络流量。仅分析你拥有或有权测试的网站流量。

4. 应对常见反爬策略

学术资源网站常用的反爬手段包括请求频率限制、用户行为分析和验证码。下面是一些实用对策：

1. 请求速率控制

import random from functools import wraps def randomized_delay(min_wait=1, max_wait=3): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): delay = random.uniform(min_wait, max_wait) time.sleep(delay) return func(*args, **kwargs) return wrapper return decorator # 使用方法 @randomized_delay(0.5, 2.5) def make_request(url): return requests.get(url)

2. 请求头轮换

USER_AGENTS = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)', 'Mozilla/5.0 (X11; Linux x86_64)' ] def rotate_headers(): return { 'User-Agent': random.choice(USER_AGENTS), 'Accept': 'text/html,application/xhtml+xml', 'Accept-Language': 'en-US,en;q=0.5', 'Referer': 'https://www.google.com/' }

3. 处理验证码

对于简单验证码，可以使用OCR库尝试自动识别：

import pytesseract from PIL import Image import io def solve_captcha(image_bytes): try: image = Image.open(io.BytesIO(image_bytes)) text = pytesseract.image_to_string(image) return text.strip() except Exception: return None

对于复杂验证码，建议：

• 手动输入后保存cookies复用
• 寻找无验证码的API接口
• 使用专业验证码解决服务

4. 会话保持技巧

def maintain_session(spider, login_url, credentials): # 先获取登录页获取CSRF token login_page = spider.session.get(login_url) soup = BeautifulSoup(login_page.text, 'html.parser') csrf_token = soup.find('input', {'name': 'csrf_token'})['value'] # 提交登录表单 credentials['csrf_token'] = csrf_token response = spider.session.post(login_url, data=credentials) # 检查登录是否成功 if 'welcome' in response.url: print("登录成功！会话已建立") return True return False

5. 资料存储与检索系统

收集到的资料需要有效组织才能发挥最大价值。我们设计一个基于SQLite的存储系统，支持全文检索。

import sqlite3 from datetime import datetime import fitz # PyMuPDF class KnowledgeBase: def __init__(self, db_path="knowledge.db"): self.conn = sqlite3.connect(db_path) self._init_db() def _init_db(self): cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS materials ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, file_path TEXT UNIQUE NOT NULL, file_type TEXT NOT NULL, source_url TEXT, content_text TEXT, keywords TEXT, added_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 启用全文搜索 (FTS5) cursor.execute(''' CREATE VIRTUAL TABLE IF NOT EXISTS materials_fts USING fts5(title, content_text, keywords) ''') self.conn.commit() def index_pdf(self, filepath): """提取PDF文本内容并索引""" try: doc = fitz.open(filepath) text = "" for page in doc: text += page.get_text() title = os.path.basename(filepath) keywords = self._extract_keywords(text) cursor = self.conn.cursor() cursor.execute(''' INSERT INTO materials (title, file_path, file_type, content_text, keywords) VALUES (?, ?, 'pdf', ?, ?) ''', (title, filepath, text, keywords)) # 更新全文搜索索引 cursor.execute(''' INSERT INTO materials_fts (rowid, title, content_text, keywords) VALUES (last_insert_rowid(), ?, ?, ?) ''', (title, text, keywords)) self.conn.commit() return True except Exception as e: print(f"索引PDF失败 {filepath}: {e}") return False def _extract_keywords(self, text, top_n=10): """简单的关键词提取""" from collections import Counter import re words = re.findall(r'\w{4,}', text.lower()) word_counts = Counter(words) return ','.join(word for word, _ in word_counts.most_common(top_n)) def search(self, query): """全文检索""" cursor = self.conn.cursor() cursor.execute(''' SELECT m.title, m.file_path, snippet(materials_fts, 2, '<b>', '</b>', '...', 64) FROM materials m JOIN materials_fts fts ON m.id = fts.rowid WHERE materials_fts MATCH ? ORDER BY rank LIMIT 10 ''', (query,)) return cursor.fetchall() def close(self): self.conn.close() # 使用示例 kb = KnowledgeBase() kb.index_pdf("materials/pdfs/computer_networks.pdf") results = kb.search("TCP 三次握手") for title, path, snippet in results: print(f"{title}: {snippet}\n-> {path}")

这个知识库系统提供：

• PDF文本内容提取
• 自动关键词生成
• 快速全文检索
• 搜索结果高亮显示
• 按相关性排序

6. 项目部署与自动化

为了让资料库保持更新，我们可以设置定时任务自动运行爬虫，并通过简单的Web界面提供访问。

1. 使用APScheduler设置定时任务

from apscheduler.schedulers.blocking import BlockingScheduler def scheduled_crawl(): spider = AcademicSpider("https://ocw.example.edu") spider.crawl("https://ocw.example.edu/cs101") kb = KnowledgeBase() for root, _, files in os.walk("materials/pdfs"): for file in files: if file.endswith(".pdf"): kb.index_pdf(os.path.join(root, file)) kb.close() scheduler = BlockingScheduler() scheduler.add_job(scheduled_crawl, 'cron', day_of_week='mon,wed,fri', hour=2) scheduler.start()

2. 简易Web界面（使用Flask）

from flask import Flask, render_template, request app = Flask(__name__) kb = KnowledgeBase() @app.route('/') def home(): query = request.args.get('q', '') results = [] if query: results = kb.search(query) return render_template('search.html', query=query, results=results) if __name__ == '__main__': app.run(port=5000)

对应的HTML模板（templates/search.html）：

<!DOCTYPE html> <html> <head> <title>个人知识库</title> <style> .result { margin-bottom: 20px; padding: 15px; border: 1px solid #ddd; } .snippet { color: #666; } .path { font-size: 0.8em; color: #999; } </style> </head> <body> <h1>复习资料检索系统</h1> <form method="get"> <input type="text" name="q" value="{{ query }}" placeholder="输入搜索关键词"> <button type="submit">搜索</button> </form> {% if query %} <h2>搜索结果</h2> {% if results %} {% for title, path, snippet in results %} <div class="result"> <h3>{{ title }}</h3> <p class="snippet">{{ snippet|safe }}</p> <p class="path">{{ path }}</p> </div> {% endfor %} {% else %} <p>没有找到相关结果</p> {% endif %} {% endif %} </body> </html>

7. 实际应用中的经验分享

在开发过程中，我遇到了几个值得注意的问题。首先是网站结构变化导致爬虫失效，解决方法是为重要爬虫添加监控通知：

def check_crawler_health(spider, test_url): try: links = spider.extract_links(test_url) if not links: send_alert_email("爬虫可能失效，未获取到任何链接") return bool(links) except Exception as e: send_alert_email(f"爬虫健康检查失败: {str(e)}") return False

另一个常见问题是资源去重。相同的讲义可能在不同页面出现，通过内容哈希可以避免重复存储：

import hashlib def file_hash(filepath): with open(filepath, 'rb') as f: return hashlib.md5(f.read()).hexdigest() def is_duplicate(filepath): kb = KnowledgeBase() cursor = kb.conn.cursor() current_hash = file_hash(filepath) cursor.execute('SELECT file_path FROM materials WHERE file_hash=?', (current_hash,)) return cursor.fetchone() is not None

最后，对于需要登录的学术平台，建议使用环境变量存储凭证，并添加两步验证支持：

import os from getpass import getpass def get_credentials(): username = os.getenv('UNIV_USERNAME') password = os.getenv('UNIV_PASSWORD') if not username: username = input("请输入学号: ") if not password: password = getpass("请输入密码: ") return { 'username': username, 'password': password }