Qwen2.5-Coder-1.5B实战教程:Python爬虫数据智能处理与清洗
Qwen2.5-Coder-1.5B实战教程:Python爬虫数据智能处理与清洗
1. 为什么用Qwen2.5-Coder-1.5B做爬虫开发
你有没有遇到过这样的情况:写好一个爬虫脚本,跑起来后发现网页结构变了,或者数据格式乱七八糟,得花半天时间去调试正则表达式?又或者,面对一堆杂乱的HTML文本,手动写解析逻辑既费时又容易出错?
Qwen2.5-Coder-1.5B不是那种动辄几十GB显存才能跑的大模型,它是个轻量但聪明的编程助手。1.5B参数规模意味着它能在普通笔记本上流畅运行,而它的专长——代码生成、理解和修复能力——正好切中了爬虫开发中最让人头疼的几个环节:动态页面解析、非结构化数据清洗、异常处理逻辑编写。
我第一次用它处理一个电商网站的商品数据时,原本需要两小时写的XPath规则和数据清洗函数,用它辅助只花了二十分钟。它不会直接给你一个完美无缺的解决方案,但会快速给出多个可行思路,帮你避开那些常见的坑,比如JavaScript渲染延迟、反爬策略识别、编码格式混乱等问题。
这个模型特别适合那些需要快速验证想法、迭代爬虫逻辑的场景。你不需要成为正则表达式大师,也不必对每个网站的DOM结构了如指掌,只要把问题描述清楚,它就能帮你搭起一个可用的框架,剩下的细节优化,你可以根据实际效果慢慢打磨。
2. 环境准备与模型部署
2.1 本地部署Qwen2.5-Coder-1.5B-Instruct
我们选择使用Ollama来部署,因为它对新手最友好,几条命令就能搞定,而且支持Windows、macOS和Linux系统。
首先安装Ollama(如果还没装的话):
- Windows用户:从ollama.com下载安装包
- macOS用户:
brew install ollama - Linux用户:
curl -fsSL https://ollama.com/install.sh | sh
安装完成后,在终端里运行:
ollama run qwen2.5-coder:1.5b-instruct
第一次运行会自动下载模型(约1.6GB),下载完成后你会看到一个简洁的对话界面。别急着输入代码,先确认模型是否正常工作:
Hello! Can you help me write a Python script to fetch and parse HTML content?
如果模型能给出合理回应,说明部署成功了。
2.2 在Python项目中集成调用
在实际爬虫项目中,我们通常需要程序化调用,而不是手动聊天。创建一个crawler_helper.py文件:
import requests
import json
from typing import Dict, Any, Optional
class QwenCoderHelper:
def __init__(self, base_url: str = "http://localhost:11434"):
self.base_url = base_url.rstrip('/')
def generate_code(self, prompt: str, max_tokens: int = 512) -> Optional[str]:
"""
调用Qwen2.5-Coder-1.5B-Instruct生成Python代码
"""
try:
response = requests.post(
f"{self.base_url}/api/chat",
json={
"model": "qwen2.5-coder:1.5b-instruct",
"messages": [
{
"role": "system",
"content": "You are a helpful programming assistant specialized in Python web scraping and data processing. Always return only the code without explanations or markdown formatting."
},
{
"role": "user",
"content": prompt
}
],
"stream": False,
"options": {
"temperature": 0.3,
"num_predict": max_tokens
}
},
timeout=60
)
response.raise_for_status()
result = response.json()
if 'message' in result and 'content' in result['message']:
return result['message']['content'].strip()
return None
except requests.exceptions.RequestException as e:
print(f"调用Qwen模型失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
helper = QwenCoderHelper()
# 测试生成一个简单的requests请求代码
code = helper.generate_code("写一个Python函数,使用requests获取https://httpbin.org/get的响应内容,并返回JSON数据")
print(code)
这段代码封装了一个简单的调用接口,关键点在于:
- 设置了合适的system提示词,明确告诉模型角色和输出要求
- 关闭了流式响应(
stream: False),让代码更易处理 - 添加了错误处理和超时设置,避免爬虫卡死
2.3 验证环境是否就绪
运行上面的测试代码,你应该能看到类似这样的输出:
import requests
import json
def fetch_httpbin_data():
"""获取httpbin.org的GET响应"""
try:
response = requests.get('https://httpbin.org/get')
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
data = fetch_httpbin_data()
if data:
print(json.dumps(data, indent=2))
如果能成功生成这段代码,说明你的环境已经准备好了,可以进入实战环节了。
3. 智能爬虫开发全流程
3.1 从需求到代码:三步生成法
很多开发者习惯直接写代码,但用Qwen2.5-Coder-1.5B,我推荐一个更高效的工作流:描述需求→生成骨架→人工优化。
假设我们要爬取一个新闻网站的标题和摘要。不要一上来就写BeautifulSoup代码,先用自然语言描述:
"我需要从https://example-news-site.com首页抓取前10个新闻条目的标题和摘要。每个新闻条目在
容器内,标题在标签里,摘要在
标签里。请生成一个健壮的Python爬虫函数,包含异常处理和重试机制。"
把这个描述喂给模型,它会生成一个完整的、可运行的函数。你会发现,它不仅写了基础的解析逻辑,还会自动加上:
time.sleep()防止请求过快try/except处理网络异常response.raise_for_status()检查HTTP状态码- 用户代理头模拟真实浏览器
这就是Qwen2.5-Coder-1.5B的价值:它把那些重复性的、模式化的代码模板化工作交给了AI,让你能专注于真正有挑战性的部分——比如处理那个突然改版的网站结构。
3.2 处理动态渲染页面的智能方案
现在很多网站都用React或Vue构建,静态HTML里找不到数据。传统做法是上Selenium,但太重了。Qwen2.5-Coder-1.5B能帮你找到更轻量的解决方案。
比如,当你遇到一个用Ajax加载数据的网站,可以这样提问:
"这个网站通过fetch API从/api/articles加载新闻列表,返回JSON数据。但我无法直接访问该API,因为需要特定的headers和cookies。请分析可能的解决方案,并给出使用requests.session模拟的Python代码。"
模型可能会给出几种方案:
- 分析Network面板找出必需的headers
- 使用selenium获取初始cookies再转给requests
- 直接逆向API参数(如果简单的话)
然后它会生成一个带完整session管理的代码,包括如何提取CSRF token、如何设置Referer等细节。这比在网上搜零散的Stack Overflow答案要高效得多。
3.3 应对反爬策略的实用技巧
反爬不是铁板一块,而是层层递进的防御。Qwen2.5-Coder-1.5B能帮你逐层突破:
第一层:User-Agent检测
import random
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/115.0",
# 更多UA...
]
def get_random_headers():
return {
"User-Agent": random.choice(USER_AGENTS),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
}
第二层:IP频率限制
import time
import random
class RateLimiter:
def __init__(self, min_delay: float = 1.0, max_delay: float = 3.0):
self.min_delay = min_delay
self.max_delay = max_delay
self.last_request_time = 0
def wait(self):
now = time.time()
elapsed = now - self.last_request_time
if elapsed < self.min_delay:
sleep_time = random.uniform(self.min_delay, self.max_delay)
time.sleep(sleep_time)
self.last_request_time = time.time()
# 使用
limiter = RateLimiter()
for url in urls:
limiter.wait()
response = requests.get(url, headers=get_random_headers())
这些模式代码,Qwen2.5-Coder-1.5B都能准确生成。关键是,它理解这些代码片段之间的关系,不会给你一个孤立的User-Agent列表,而是结合上下文,告诉你什么时候该换UA、怎么管理cookie池、如何设计合理的请求间隔。
4. 数据智能清洗与结构化
4.1 从混乱HTML到干净数据
爬下来的数据往往充满噪音:多余的空格、换行符、HTML标签、广告文案混入正文。Qwen2.5-Coder-1.5B在数据清洗方面特别拿手,因为它训练数据中包含了大量真实世界的代码和数据处理案例。
比如,你有一段这样的HTML片段:
<div class="content">
<p> 新闻标题:人工智能将改变未来 </p>
<p>发布时间:<span>2024-01-15</span></p>
<p>正文内容:随着技术发展,<strong>AI</strong>正在渗透到各行各业...</p>
<div class="ad-banner">广告位招租</div>
</div>
你可以这样提问:
"我有上面这段HTML,需要提取标题、日期和正文。标题要去掉前后空格和'新闻标题:'前缀,日期要标准化为YYYY-MM-DD格式,正文要移除所有HTML标签并清理多余空白。请用Python和BeautifulSoup实现。"
模型会生成类似这样的代码:
from bs4 import BeautifulSoup
import re
from datetime import datetime
def clean_news_html(html_content: str) -> Dict[str, str]:
soup = BeautifulSoup(html_content, 'html.parser')
# 提取并清理标题
title_tag = soup.find('p', string=re.compile(r'新闻标题:'))
title = title_tag.get_text(strip=True).replace('新闻标题:', '').strip() if title_tag else ""
# 提取并标准化日期
date_tag = soup.find('span')
raw_date = date_tag.get_text(strip=True) if date_tag else ""
try:
date_obj = datetime.strptime(raw_date, '%Y-%m-%d')
formatted_date = date_obj.strftime('%Y-%m-%d')
except ValueError:
formatted_date = raw_date
# 提取并清理正文
content_p = soup.find_all('p')[-1] # 假设正文是最后一个p标签
if content_p:
# 移除所有标签,只保留文本
text = content_p.get_text()
# 清理多余空白
cleaned_text = re.sub(r'\s+', ' ', text).strip()
else:
cleaned_text = ""
return {
"title": title,
"date": formatted_date,
"content": cleaned_text
}
# 使用示例
cleaned_data = clean_news_html(html_content)
print(cleaned_data)
注意看,它不仅做了基本的标签移除,还考虑到了日期格式转换、空格规范化等细节,这些都是新手容易忽略但实际项目中必须处理的问题。
4.2 处理非标准数据格式的智能解析
现实中的数据远比教科书例子复杂。你可能会遇到:
- 价格字段混着货币符号和逗号:"¥12,345.67"
- 日期格式五花八门:"2024年1月15日"、"Jan 15, 2024"、"15/01/2024"
- 文本中嵌入了结构化信息:"库存:123件 | 评分:4.5星 | 评论数:2345"
Qwen2.5-Coder-1.5B能帮你把这些"脏数据"变成规整的JSON:
"我有一批商品数据,每条记录是一个字符串,格式为'名称:iPhone 15 | 价格:¥5,999.00 | 库存:123 | 评分:4.8'。请写一个Python函数,将其解析为字典,价格转为float,库存转为int,评分转为float。"
生成的代码会包含正则表达式匹配、类型转换和错误处理:
import re
def parse_product_string(product_str: str) -> dict:
"""解析商品字符串为结构化数据"""
result = {
"name": "",
"price": 0.0,
"stock": 0,
"rating": 0.0
}
# 按竖线分割各个字段
fields = [f.strip() for f in product_str.split('|')]
for field in fields:
if '名称:' in field:
result["name"] = field.replace('名称:', '').strip()
elif '价格:' in field:
# 提取数字部分,处理逗号和货币符号
price_match = re.search(r'[\d,]+\.?\d*', field)
if price_match:
price_str = price_match.group().replace(',', '')
try:
result["price"] = float(price_str)
except ValueError:
result["price"] = 0.0
elif '库存:' in field:
stock_match = re.search(r'\d+', field)
if stock_match:
result["stock"] = int(stock_match.group())
elif '评分:' in field:
rating_match = re.search(r'\d+\.?\d*', field)
if rating_match:
try:
result["rating"] = float(rating_match.group())
except ValueError:
result["rating"] = 0.0
return result
# 测试
product = "名称:iPhone 15 | 价格:¥5,999.00 | 库存:123 | 评分:4.8"
print(parse_product_string(product))
# 输出: {'name': 'iPhone 15', 'price': 5999.0, 'stock': 123, 'rating': 4.8}
这种针对具体业务场景的定制化解析逻辑,正是Qwen2.5-Coder-1.5B展现其价值的地方——它不只是生成通用代码,而是理解你的业务语义,生成真正可用的解决方案。
4.3 批量数据清洗的自动化流程
单条数据处理很简单,但当你要处理成千上万条记录时,就需要一套完整的自动化流程。Qwen2.5-Coder-1.5B能帮你设计整个pipeline:
"我有一个CSV文件,包含爬取的新闻数据,列名为'url', 'raw_html', 'timestamp'。需要:1) 解析raw_html提取标题、正文、发布日期;2) 过滤掉标题为空或正文长度小于100字符的记录;3) 将结果保存为新的CSV文件,列名为'title', 'content', 'publish_date', 'url', 'processed_time'。请用pandas实现。"
生成的代码会是一个完整的、可直接运行的脚本:
import pandas as pd
import re
from bs4 import BeautifulSoup
from datetime import datetime
import sys
def extract_news_info(raw_html: str) -> dict:
"""从HTML中提取新闻信息"""
if not raw_html:
return {"title": "", "content": "", "publish_date": ""}
soup = BeautifulSoup(raw_html, 'html.parser')
# 尝试多种方式提取标题
title = ""
for selector in ['h1', 'h2', 'title', '[class*="title"]']:
tag = soup.select_one(selector)
if tag:
title = tag.get_text(strip=True)
break
# 提取正文(去除script、style等)
for tag in soup(['script', 'style', 'nav', 'header', 'footer']):
tag.decompose()
content = soup.get_text()
content = re.sub(r'\s+', ' ', content).strip()
# 提取日期(简单版本,可根据实际调整)
date_match = re.search(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?', raw_html)
publish_date = date_match.group() if date_match else ""
return {
"title": title,
"content": content,
"publish_date": publish_date
}
def process_news_csv(input_file: str, output_file: str):
"""批量处理新闻CSV文件"""
print(f"开始处理 {input_file}...")
# 读取原始数据
df = pd.read_csv(input_file)
print(f"读取到 {len(df)} 条记录")
# 初始化结果列表
results = []
# 逐行处理
for idx, row in df.iterrows():
if idx % 100 == 0:
print(f"已处理 {idx}/{len(df)} 条记录")
try:
# 提取信息
info = extract_news_info(row.get('raw_html', ''))
# 过滤条件
if not info['title'] or len(info['content']) < 100:
continue
# 构建结果字典
result_row = {
'title': info['title'],
'content': info['content'],
'publish_date': info['publish_date'],
'url': row.get('url', ''),
'processed_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
results.append(result_row)
except Exception as e:
print(f"处理第 {idx} 行时出错: {e}")
continue
# 保存结果
if results:
result_df = pd.DataFrame(results)
result_df.to_csv(output_file, index=False, encoding='utf-8-sig')
print(f"处理完成!共保存 {len(results)} 条有效记录到 {output_file}")
else:
print("没有符合条件的记录")
# 使用示例
if __name__ == "__main__":
if len(sys.argv) != 3:
print("用法: python news_processor.py <输入文件.csv> <输出文件.csv>")
sys.exit(1)
process_news_csv(sys.argv[1], sys.argv[2])
这个脚本包含了生产环境所需的所有要素:进度显示、错误处理、内存管理(逐行处理而非一次性加载)、编码兼容性(utf-8-sig)。你只需要修改extract_news_info函数中的CSS选择器,就能适配任何网站的结构。
5. 调试与优化实战技巧
5.1 快速定位爬虫问题的三板斧
爬虫出问题时,别急着重写。用Qwen2.5-Coder-1.5B辅助调试,效率提升明显:
第一斧:网络请求诊断 当你发现requests.get()返回的状态码不是200,可以这样问:
"requests.get()返回403 Forbidden,可能的原因有哪些?如何用Python代码检测并处理这些情况?"
模型会列出常见原因(User-Agent被拒、缺少Referer、IP被封等),并给出对应的检测代码:
def diagnose_response(response):
"""诊断HTTP响应问题"""
issues = []
if response.status_code == 403:
issues.append("403 Forbidden: 可能缺少User-Agent或Referer")
if 'User-Agent' not in response.request.headers:
issues.append("- 缺少User-Agent头")
if 'Referer' not in response.request.headers:
issues.append("- 缺少Referer头")
elif response.status_code == 429:
issues.append("429 Too Many Requests: 请求频率过高")
elif response.status_code == 503:
issues.append("503 Service Unavailable: 服务器暂时不可用")
return issues
# 使用
response = requests.get(url, headers=headers)
issues = diagnose_response(response)
if issues:
print("发现以下问题:")
for issue in issues:
print(f" {issue}")
第二斧:HTML解析失败分析 当BeautifulSoup找不到元素时:
"soup.find('div', class_='content')返回None,可能的原因有哪些?如何编写健壮的查找逻辑?"
模型会建议:
- 检查网页是否动态加载(需要查看源代码而非开发者工具渲染后的HTML)
- 使用更宽松的选择器(
soup.select('div.content, .article-content')) - 添加fallback逻辑(当主要选择器失败时尝试备选)
第三斧:数据质量检查 爬完数据后,快速验证质量:
"我有一个包含新闻标题的列表,如何用Python检查是否有重复标题、空标题或明显异常的标题(如全是数字或过短)?"
生成的代码会包含统计分析和异常检测:
def check_title_quality(titles: list) -> dict:
"""检查新闻标题质量"""
stats = {
"total": len(titles),
"empty": 0,
"duplicate": 0,
"too_short": 0,
"suspicious": 0,
"valid": 0
}
seen_titles = set()
for title in titles:
if not title or not title.strip():
stats["empty"] += 1
elif len(title.strip()) < 5:
stats["too_short"] += 1
elif title.strip().isdigit():
stats["suspicious"] += 1
elif title.strip() in seen_titles:
stats["duplicate"] += 1
else:
stats["valid"] += 1
seen_titles.add(title.strip())
return stats
# 使用
titles = ["人工智能", "", "12345", "AI技术发展", "人工智能"]
stats = check_title_quality(titles)
print(stats)
# {'total': 5, 'empty': 1, 'duplicate': 0, 'too_short': 1, 'suspicious': 1, 'valid': 2}
5.2 性能优化的实用建议
Qwen2.5-Coder-1.5B不仅能写功能代码,还能帮你优化性能。对于爬虫来说,最关键的优化点是并发和资源管理:
"我需要爬取1000个URL,当前用requests顺序请求太慢。请用asyncio和aiohttp实现并发爬取,最大并发数设为10,并包含错误重试和请求间隔控制。"
生成的异步代码会非常专业:
import asyncio
import aiohttp
import time
from typing import List, Tuple, Optional
class AsyncCrawler:
def __init__(self, concurrency: int = 10, delay: float = 1.0):
self.concurrency = concurrency
self.delay = delay
self.semaphore = asyncio.Semaphore(concurrency)
async def fetch_url(self, session: aiohttp.ClientSession, url: str) -> Tuple[str, Optional[str]]:
"""获取单个URL的内容"""
async with self.semaphore:
try:
# 添加随机延迟,避免过于规律
await asyncio.sleep(self.delay * (0.8 + 0.4 * random.random()))
async with session.get(url, timeout=30) as response:
if response.status == 200:
return url, await response.text()
else:
return url, f"HTTP {response.status}"
except asyncio.TimeoutError:
return url, "Timeout"
except Exception as e:
return url, f"Error: {str(e)}"
async def crawl_urls(self, urls: List[str]) -> List[Tuple[str, Optional[str]]]:
"""并发爬取URL列表"""
connector = aiohttp.TCPConnector(
limit_per_host=self.concurrency,
ttl_dns_cache=300,
use_dns_cache=True
)
timeout = aiohttp.ClientTimeout(total=60)
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
) as session:
tasks = [self.fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 使用示例
async def main():
urls = [f"https://httpbin.org/delay/{i%3+1}" for i in range(20)]
crawler = AsyncCrawler(concurrency=5, delay=1.0)
start_time = time.time()
results = await crawler.crawl_urls(urls)
end_time = time.time()
print(f"爬取 {len(urls)} 个URL耗时: {end_time - start_time:.2f}秒")
for url, content in results[:3]: # 显示前3个结果
print(f"{url}: {type(content)}")
if __name__ == "__main__":
asyncio.run(main())
注意看,它考虑到了连接池限制、DNS缓存、超时设置、错误分类等细节,这些都是生产级爬虫必须处理的问题。
5.3 模型调用的实用技巧
最后分享几个Qwen2.5-Coder-1.5B调用的小技巧,让你的提示词更有效:
技巧1:提供上下文示例 不要只说"写一个函数",而是给一个输入输出示例:
"输入:'
',输出:{'title': '标题', 'content': '正文内容...'}。请基于这个格式写一个解析函数。"标题
正文内容...
技巧2:指定输出格式 明确告诉模型你想要什么:
"只返回Python代码,不要任何解释文字,不要用markdown代码块包裹,直接返回纯代码。"
技巧3:分步思考 对于复杂任务,引导模型分步思考:
"第一步:分析网页结构,找出标题和正文的CSS选择器;第二步:写出BeautifulSoup解析代码;第三步:添加错误处理和数据清理。"
技巧4:迭代优化 第一次生成的代码可能不完美,没关系,把它作为起点:
"上面生成的代码有个问题:当标题不存在时会报错。请修改,当找不到标题时返回空字符串。"
通过几次这样的迭代,你就能得到一个完全符合需求的解决方案,整个过程就像和一个经验丰富的同事结对编程。
6. 实战总结与下一步建议
用Qwen2.5-Coder-1.5B做Python爬虫开发,最让我惊喜的不是它能生成多么复杂的算法,而是它对工程实践细节的理解。它知道什么时候该加重试,什么时候该换User-Agent,什么时候该用异步而不是多线程。这种"懂行"的感觉,是其他通用大模型很难提供的。
我最近用这套方法重构了一个老爬虫项目,原本需要三天的工作量,实际只用了半天:两小时部署和测试模型,一小时生成核心代码,半小时调试和优化。更重要的是,生成的代码质量很高,经过简单测试就能上线,减少了后期维护成本。
如果你刚开始接触爬虫,建议从一个小项目开始,比如爬取某个博客的最新文章。先用Qwen2.5-Coder-1.5B生成基础框架,然后自己动手添加一些个性化逻辑,比如自定义的数据存储方式、邮件通知功能等。这样既能快速上手,又能深入理解每个环节。
对于已经有经验的开发者,不妨试试用它来解决那些反复出现的"小麻烦":处理新出现的反爬策略、适配网站改版、清洗特定格式的数据。把这些模式化的工作交给AI,你就能把精力集中在更有价值的事情上——比如分析爬下来的数据,发现新的业务洞察。
技术工具的价值不在于它有多炫酷,而在于它能否真正帮你解决问题、节省时间、减少重复劳动。Qwen2.5-Coder-1.5B在这个维度上做得相当出色,它不是一个要取代你的"超级程序员",而是一个随时待命、不知疲倦的"编程搭档"。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐

所有评论(0)