DeepSeek-R1 模型 Python 爬虫实战：智能数据采集与清洗

2.3 测试模型是否正常工作

写个简单的测试看看模型能不能用：

def test_model():
    """请用 Python 写一个简单的 requests 爬虫，爬取网页标题。"""
    prompt = "请用 Python 写一个简单的 requests 爬虫，爬取网页标题。"
    messages = [
        {"role": "user", "content": prompt}
    ]
    # 注意：DeepSeek-R1 系列建议不要用 system prompt
    # 所有指令都放在 user prompt 里
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            **generation_config
        )
    response = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    print("模型响应：")
    print(response)

# 运行测试
test_model()

如果看到模型输出了 Python 爬虫代码，说明环境配置成功了。

3. 智能爬虫代码生成：让 AI 写爬虫

3.1 基础静态页面爬取

假设我们要爬取一个新闻网站的文章列表。传统方法需要分析 HTML 结构，写 CSS 选择器或 XPath。现在让 AI 来帮忙：

def generate_static_crawler(url, description):
    """生成静态页面爬虫代码"""
    prompt = f"""
    我需要爬取这个网站：{url}
    网站描述：{description}
    请帮我生成完整的 Python 爬虫代码，要求：
    1. 使用 requests 获取页面
    2. 使用 BeautifulSoup 解析 HTML
    3. 提取文章标题、链接、发布时间
    4. 处理可能的异常（网络错误、解析错误）
    5. 将结果保存到 CSV 文件
    6. 添加适当的延迟避免被封
    7. 代码要有详细的注释
    请直接输出完整的 Python 代码，不要解释。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1500,  # 代码可能比较长
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return code

# 使用示例
url = "https://news.example.com"
description = "这是一个新闻网站，文章列表在 class 为'article-list'的 div 中，每篇文章在 class 为'article-item'的 div 里"
code = generate_static_crawler(url, description)
print("生成的爬虫代码：")
print(code)

3.2 动态页面处理（JavaScript 渲染）

很多现代网站用 JavaScript 动态加载内容，requests+BeautifulSoup 搞不定。这时候需要 Selenium：

def generate_dynamic_crawler(url, interaction_steps):
    """生成动态页面爬虫代码"""
    prompt = f"""
    我需要爬取这个动态网站：{url}
    页面特点：{interaction_steps}
    请生成使用 Selenium 的 Python 爬虫代码，要求：
    1. 使用 Chrome 浏览器驱动
    2. 处理页面加载等待
    3. 执行必要的交互操作（点击、滚动等）
    4. 提取动态加载的内容
    5. 处理弹窗和 cookie
    6. 使用 headless 模式（无头浏览器）
    7. 代码要健壮，处理各种异常情况
    请直接输出完整的 Python 代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=2000,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return code

# 使用示例
url = "https://spa.example.com"
steps = "页面初始只显示 10 条数据，滚动到底部会加载更多，需要模拟滚动操作"
code = generate_dynamic_crawler(url, steps)

4. 智能解析网页结构：AI 理解 HTML

4.1 自动分析页面结构

有时候我们面对一个复杂的网页，不知道该怎么提取数据。可以让 AI 帮忙分析：

def analyze_page_structure(html_content):
    """让 AI 分析 HTML 结构，建议提取方案"""
    prompt = f"""
    这是网页的 HTML 内容（部分）：
    {html_content[:2000]}  # 只传前 2000 字符避免太长
    请分析这个网页的结构，并告诉我：
    1. 可能的数据区域在哪里
    2. 用什么选择器提取数据比较合适
    3. 数据项的可能字段（标题、价格、描述等）
    4. 需要注意的特殊情况（分页、懒加载等）
    请用清晰的方式回答。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            **generation_config
        )
    analysis = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return analysis

# 使用示例
import requests
from bs4 import BeautifulSoup

url = "https://example.com"
response = requests.get(url)
soup = BeautifulSoup(response.content, 'html.parser')
# 获取页面主要内容的 HTML
main_content = str(soup.find('main') or soup.body)[:2000]
analysis = analyze_page_structure(main_content)
print("AI 分析结果：")
print(analysis)

4.2 生成自适应解析器

基于 AI 的分析，我们可以生成更智能的解析器：

def generate_adaptive_parser(html_samples):
    """生成能适应多种结构的解析器"""
    prompt = f"""
    我有多组 HTML 样本，它们结构类似但可能有细微差异：
    样本 1：
    {html_samples[0][:1000]}
    样本 2：
    {html_samples[1][:1000]}
    样本 3：
    {html_samples[2][:1000]}
    请帮我写一个 Python 函数，能够智能地从这些类似但不完全相同的结构中提取数据。
    要求：
    1. 函数应该能处理结构变化
    2. 使用多种选择器尝试
    3. 有回退机制（如果首选选择器失败）
    4. 返回结构化的数据
    5. 记录提取成功率
    请输出完整的函数代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1800,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    parser_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return parser_code

5. 反爬机制智能应对

5.1 识别反爬策略

让 AI 帮忙分析网站可能使用的反爬机制：

def analyze_anti_scraping(response_headers, html_content, behavior_notes):
    """分析反爬机制"""
    prompt = f"""
    我正在爬取一个网站，遇到了困难。以下是相关信息：
    响应头：
    {str(response_headers)[:500]}
    页面特征：
    {behavior_notes}
    HTML 中的可疑内容（如果有）：
    {html_content[:1000] if html_content else "无"}
    请分析：
    1. 这个网站可能使用了哪些反爬技术？
    2. 如何绕过这些限制？
    3. 需要调整哪些爬虫参数？
    4. 推荐的具体技术方案
    请给出详细的建议。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            **generation_config
        )
    advice = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return advice

# 使用示例
import requests

url = "https://protected-site.example.com"
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
try:
    response = requests.get(url, headers=headers, timeout=10)
    advice = analyze_anti_scraping(
        response.headers, response.text[:1000], "请求被拒绝，返回 403 错误，但用浏览器可以正常访问"
    )
    print("反爬分析建议：")
    print(advice)
except Exception as e:
    print(f"请求失败：{e}")

5.2 生成反爬绕过代码

基于分析结果，让 AI 生成具体的绕过代码：

def generate_bypass_code(anti_scraping_type, target_url):
    """生成绕过特定反爬机制的代码"""
    prompt = f"""
    网站：{target_url}
    反爬类型：{anti_scraping_type}
    请生成 Python 代码来绕过这种反爬机制。代码应该包括：
    1. 必要的库导入
    2. 请求头设置
    3. 会话管理
    4. 代理设置（如果需要）
    5. 验证码处理（如果需要）
    6. 速率限制策略
    7. 错误重试机制
    请输出完整可运行的代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1500,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    bypass_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return bypass_code

6. 智能数据清洗与处理

6.1 自动识别数据质量问题

爬取的数据往往有很多问题：格式不一致、缺失值、重复项等。让 AI 帮忙分析：

def analyze_data_quality(data_sample):
    """分析数据质量问题"""
    prompt = f"""
    这是一些爬取到的数据样本（JSON 格式）：
    {data_sample}
    请分析这些数据可能存在的质量问题：
    1. 格式不一致的地方
    2. 缺失值情况
    3. 异常值或错误数据
    4. 需要标准化的字段
    5. 数据清洗的建议步骤
    请详细列出问题和建议。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            **generation_config
        )
    analysis = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return analysis

# 使用示例
sample_data = [
    {"title": "产品 A", "price": "$100.00", "stock": "有货"},
    {"title": "产品 B", "price": "200 元", "stock": "缺货"},
    {"title": "产品 C", "price": "150", "stock": None},
    {"title": "产品 D", "price": "$三百", "stock": "有货"}
]
quality_report = analyze_data_quality(sample_data)
print("数据质量分析：")
print(quality_report)

6.2 生成智能清洗管道

基于分析结果，生成完整的数据清洗代码：

def generate_cleaning_pipeline(data_structure, quality_issues):
    """生成数据清洗管道代码"""
    prompt = f"""
    数据结构：
    {data_structure}
    发现的数据质量问题：
    {quality_issues}
    请生成一个完整的数据清洗管道，包括：
    1. 数据加载函数
    2. 针对每个问题的清洗函数
    3. 数据验证函数
    4. 清洗流程控制
    5. 结果保存
    要求使用 pandas 进行数据处理，代码要模块化、可重用。
    请输出完整的 Python 代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=2000,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    pipeline_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return pipeline_code

6.3 实体识别与标准化

对于文本数据，经常需要提取和标准化实体信息：

def generate_entity_extractor(text_samples, entity_types):
    """生成实体提取和标准化代码"""
    prompt = f"""
    文本样本：
    {text_samples}
    需要提取的实体类型：{entity_types}
    请生成 Python 代码来：
    1. 从文本中提取指定类型的实体
    2. 标准化实体格式（比如日期统一为 YYYY-MM-DD）
    3. 处理多语言或变体
    4. 验证提取结果
    可以使用正则表达式、关键词匹配或简单的 NLP 技术。
    请输出完整代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1800,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    extractor_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return extractor_code

7. 完整实战案例：电商价格监控爬虫

让我们用一个完整的例子，把上面所有技术用起来。假设我们要监控某个电商网站的商品价格变化。

7.1 项目架构设计

先让 AI 帮我们设计整个项目的架构：

def design_project_architecture(requirements):
    """设计爬虫项目架构"""
    prompt = f"""
    项目需求：{requirements}
    请设计一个完整的电商价格监控爬虫系统架构，包括：
    1. 项目目录结构
    2. 主要模块划分
    3. 数据流设计
    4. 配置管理方案
    5. 日志和监控
    6. 错误处理策略
    7. 部署建议
    请用清晰的格式描述。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1500,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    architecture = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return architecture

# 设计架构
requirements = """
监控 3 个电商网站的商品价格
每天定时爬取
检测价格变化并发送通知
数据存储到数据库
支持添加新的监控商品
"""
arch = design_project_architecture(requirements)
print("项目架构设计：")
print(arch)

7.2 生成核心爬虫代码

基于架构设计，生成主要爬虫代码：

def generate_ecommerce_crawler(site_info):
    """生成电商爬虫核心代码"""
    prompt = f"""
    电商网站信息：
    {site_info}
    请生成一个健壮的电商价格爬虫，要求：
    1. 支持多个商品页面爬取
    2. 处理登录状态（如果需要）
    3. 解析商品标题、价格、库存、评分
    4. 处理各种价格格式（原价、折扣价、会员价）
    5. 异常处理和重试机制
    6. 遵守 robots.txt
    7. 添加合理的延迟
    请输出完整的 Python 类代码。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=2500,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    crawler_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return crawler_code

7.3 价格变化检测与通知

def generate_price_monitor():
    """生成价格监控和通知代码"""
    prompt = """
    请生成一个价格变化监控系统，功能包括：
    1. 读取历史价格数据
    2. 比较当前价格和历史价格
    3. 检测价格下降、缺货、上新等情况
    4. 通过邮件或 Webhook 发送通知
    5. 生成价格趋势图表
    6. 数据存储到 SQLite 数据库
    请输出完整的 Python 代码，包括数据库模型、监控逻辑和通知功能。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=2200,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    monitor_code = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return monitor_code

8. 调试技巧与最佳实践

8.1 AI 辅助调试

当爬虫出问题时，让 AI 帮忙分析：

def debug_crawler(error_message, code_snippet, context):
    """使用 AI 调试爬虫问题"""
    prompt = f"""
    我的爬虫遇到了问题：
    错误信息：
    {error_message}
    相关代码：
    {code_snippet}
    上下文信息：
    {context}
    请帮我：
    1. 分析可能的原因
    2. 提供修复建议
    3. 如果有必要，给出修改后的代码
    4. 建议如何避免类似问题
    请详细解答。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            **generation_config
        )
    debug_advice = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return debug_advice

8.2 性能优化建议

def optimize_crawler_performance(current_code, performance_metrics):
    """优化爬虫性能"""
    prompt = f"""
    当前爬虫代码：
    {current_code[:1500]}
    性能指标：
    {performance_metrics}
    请提供性能优化建议，包括：
    1. 并发爬取优化
    2. 内存使用优化
    3. 网络请求优化
    4. 解析速度优化
    5. 资源清理建议
    请给出具体的代码改进方案。
    """
    messages = [{"role": "user", "content": prompt}]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1500,
            **{k: v for k, v in generation_config.items() if k != "max_new_tokens"}
        )
    optimization = tokenizer.decode(outputs[0][len(inputs["input_ids"][0]):], skip_special_tokens=True)
    return optimization

9. 总结与下一步建议

用 DeepSeek-R1-Distill-Llama-8B 来做 Python 爬虫开发，确实能省不少事。我实际用下来感觉最明显的好处是，不用再花大量时间研究每个网站的独特结构，AI 能帮你分析并生成合适的代码。处理反爬机制也比以前轻松，AI 能给出具体的绕过方案。

不过也要注意，AI 生成的代码不一定每次都完美，有时候需要你稍微调整一下。特别是对于特别复杂的网站，可能还需要结合一些传统的手动分析方法。我的建议是，把 AI 当作一个强大的助手，而不是完全依赖它。

如果你刚开始用，可以从简单的静态网站开始尝试，熟悉一下整个流程。等掌握了基本用法，再挑战那些有反爬的动态网站。记得要遵守网站的 robots.txt，合理控制爬取频率，别给人家服务器造成太大压力。

这个模型在代码生成和逻辑推理方面确实挺强的，但有时候会有点啰嗦，生成的内容比较长。你可以通过调整 temperature 参数来控制输出的随机性，找到最适合你任务的设置。