集团网站开发,apache 做网站,在线装修设计软件,国内知名的wordpress网站Qwen2.5-32B-Instruct爬虫开发实战#xff1a;数据采集与清洗 做爬虫的朋友们#xff0c;不知道你们有没有这样的经历#xff1a;辛辛苦苦写好的爬虫脚本#xff0c;运行几天后突然就失效了#xff0c;要么被网站封IP#xff0c;要么页面结构变了数据抓不到。更头疼的是…Qwen2.5-32B-Instruct爬虫开发实战数据采集与清洗做爬虫的朋友们不知道你们有没有这样的经历辛辛苦苦写好的爬虫脚本运行几天后突然就失效了要么被网站封IP要么页面结构变了数据抓不到。更头疼的是抓下来的数据乱七八糟清洗起来比写爬虫本身还费时间。最近我在尝试用大模型来辅助爬虫开发发现Qwen2.5-32B-Instruct在这方面表现相当不错。这模型在代码生成和指令跟随上做了深度优化正好能解决我们爬虫开发中的几个痛点。今天就跟大家分享一下怎么用这个模型来增强爬虫功能特别是智能反反爬策略和数据结构化处理这块。1. 为什么选择Qwen2.5-32B-Instruct做爬虫助手先说说我为什么选这个模型。爬虫开发其实挺特殊的它不像一般的应用开发很多时候需要处理一些“非标准”问题。比如网站的反爬机制千奇百怪每个网站的数据结构也不一样。Qwen2.5-32B-Instruct有几个特点特别适合爬虫场景指令跟随能力强这模型能精准理解你的要求。你说“输出为JSON格式包含标题、价格、发布时间三个字段”它真的会按这个格式来不会自作主张加一堆乱七八糟的东西。代码生成质量高爬虫本质上就是写代码Qwen2.5在代码生成这块表现很好生成的代码可读性强而且会考虑一些实际场景的细节。上下文长支持128K的上下文这意味着你可以把整个网页的HTML代码、你的需求说明、还有之前的对话历史都放进去它都能处理。结构化输出特别擅长生成JSON、表格这类结构化数据这对数据清洗和整理太重要了。我用过不少模型来做爬虫辅助有些模型生成的代码看着漂亮但一运行就报错。有些模型理解不了复杂的网页结构。Qwen2.5-32B-Instruct算是平衡得比较好的一个。2. 智能请求头优化让爬虫更像真人反爬的第一道防线通常是请求头检测。很多网站会检查User-Agent、Referer这些字段如果看起来像机器人在访问直接就给你拦了。传统做法是我们手动维护一个User-Agent列表随机选一个用。但这样还不够因为真实的浏览器请求头是一整套的不只是User-Agent。用Qwen2.5我们可以让它智能生成更真实的请求头。看看这个例子import requests from typing import Dict, List import random class SmartHeaderGenerator: def __init__(self): self.browsers [ { name: Chrome, versions: [120.0.0.0, 121.0.0.0, 122.0.0.0], platforms: [Windows NT 10.0, Macintosh; Intel Mac OS X 10_15_7] }, { name: Firefox, versions: [120.0, 121.0, 122.0], platforms: [Windows NT 10.0, Macintosh; Intel Mac OS X 10.15] } ] def generate_headers(self, target_url: str) - Dict[str, str]: 生成智能请求头 # 让模型根据目标URL生成合适的请求头 prompt f 请为爬虫访问 {target_url} 生成一个看起来像真实浏览器的HTTP请求头。 要求 1. User-Agent要看起来像真实浏览器 2. 包含Accept、Accept-Language、Accept-Encoding等标准字段 3. Referer设置为合理的来源页面 4. 如果是电商网站可以添加一些购物相关的headers 5. 输出格式为Python字典 目标网站类型{self._guess_site_type(target_url)} # 这里实际应该调用Qwen2.5模型 # headers self._call_qwen(prompt) # 为了演示先返回一个示例 browser random.choice(self.browsers) version random.choice(browser[versions]) platform random.choice(browser[platforms]) return { User-Agent: fMozilla/5.0 ({platform}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Referer: self._generate_referer(target_url), Connection: keep-alive, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: same-origin, Cache-Control: max-age0 } def _guess_site_type(self, url: str) - str: 猜测网站类型 if taobao in url or jd.com in url: return 电商 elif news in url or xinhua in url: return 新闻 elif weibo in url or douyin in url: return 社交媒体 else: return 普通网站 def _generate_referer(self, url: str) - str: 生成合理的Referer # 简单实现如果是商品页Referer设为搜索页 if item in url or product in url: return https://www.google.com/ return https://www.baidu.com/ # 使用示例 header_gen SmartHeaderGenerator() url https://item.jd.com/123456.html headers header_gen.generate_headers(url) response requests.get(url, headersheaders, timeout10)但这只是基础版。更智能的做法是让Qwen2.5分析目标网站的特点动态调整请求头。比如有些网站对移动端访问更友好那我们就生成移动端的请求头。有些API接口需要特定的认证头模型可以根据文档自动生成。3. 动态页面解析应对结构变化网页结构变化是爬虫开发者的噩梦。今天还能正常运行的爬虫明天可能就因为页面改版而失效。传统做法是写一堆XPath或CSS选择器一旦页面结构变了就得手动调整。用Qwen2.5我们可以实现更智能的解析策略。核心思路是让模型理解页面结构然后根据我们的数据需求来提取信息。3.1 基于描述的解析与其写死的选择器不如告诉模型“我要提取商品标题、价格、销量”让模型自己去页面里找。看这个例子from bs4 import BeautifulSoup import json class IntelligentParser: def __init__(self, html_content: str): self.soup BeautifulSoup(html_content, html.parser) # 简化HTML移除脚本和样式 for script in self.soup([script, style]): script.decompose() def extract_with_description(self, extraction_rules: str) - dict: 根据描述提取数据 extraction_rules示例 从商品页面提取以下信息 1. 商品标题通常在h1标签或class包含title的元素中 2. 商品价格可能包含原价和现价找包含price或的元素 3. 商品销量可能显示已售或销量 4. 商品评价数量 5. 商品详情描述 # 获取简化后的页面文本内容用于模型分析 page_text self.soup.get_text()[:5000] # 限制长度 prompt f 以下是网页的部分内容 {page_text} 请根据以下规则提取数据 {extraction_rules} 要求 1. 只提取明确存在的信息如果找不到就留空 2. 输出格式为JSON 3. 对于价格统一转换为浮点数 4. 对于销量提取数字部分 # 调用Qwen2.5模型 # result self._call_qwen(prompt) # 模拟返回结果 result { 商品标题: 示例商品标题, 商品价格: 299.0, 商品销量: 1500, 评价数量: 234, 商品描述: 这是商品描述内容 } return result def find_data_patterns(self) - list: 自动发现页面中的数据模式 # 找所有包含数字的文本可能是价格、销量等 price_patterns [] for text in self.soup.stripped_strings: if in text or ¥ in text or (元 in text and any(c.isdigit() for c in text)): price_patterns.append(text) # 找可能是标题的文本通常比较短且在h1-h6标签中 titles [] for i in range(1, 7): for header in self.soup.find_all(fh{i}): if len(header.get_text(stripTrue)) 100: # 标题通常不会太长 titles.append(header.get_text(stripTrue)) return { 可能的价格: price_patterns[:5], 可能的标题: titles[:5], 链接数量: len(self.soup.find_all(a)), 图片数量: len(self.soup.find_all(img)) } # 使用示例 html html h1华为Mate 60 Pro 智能手机/h1 div classprice6999.00/div div已售1500件/div div评价234条/div /html parser IntelligentParser(html) # 方法1基于描述提取 rules 提取商品信息 1. 商品标题 2. 商品价格 3. 商品销量 4. 评价数量 data parser.extract_with_description(rules) print(json.dumps(data, indent2, ensure_asciiFalse)) # 方法2自动发现数据模式 patterns parser.find_data_patterns() print(\n发现的数据模式) print(json.dumps(patterns, indent2, ensure_asciiFalse))3.2 自适应解析策略对于经常变动的网站我们可以让模型学习页面的变化模式。比如电商网站的商品详情页虽然结构会变但核心信息标题、价格、图片通常还在页面的某些区域。我们可以设计一个自适应系统class AdaptiveParser: def __init__(self): self.parsing_strategies [] self.learned_patterns {} def add_parsing_strategy(self, strategy_type: str, rules: dict): 添加解析策略 self.parsing_strategies.append({ type: strategy_type, rules: rules, success_rate: 0.0, last_used: None }) def parse_with_fallback(self, html: str, target_data: list) - dict: 使用回退策略解析 results {} # 按成功率排序策略 sorted_strategies sorted( self.parsing_strategies, keylambda x: x[success_rate], reverseTrue ) for strategy in sorted_strategies[:3]: # 尝试前3个策略 try: if strategy[type] css_selectors: data self._parse_with_css(html, strategy[rules]) elif strategy[type] xpath: data self._parse_with_xpath(html, strategy[rules]) elif strategy[type] ai_extraction: data self._parse_with_ai(html, target_data) # 验证提取的数据是否完整 if self._validate_data(data, target_data): strategy[success_rate] min(1.0, strategy[success_rate] 0.1) results data break else: strategy[success_rate] max(0.0, strategy[success_rate] - 0.05) except Exception as e: strategy[success_rate] max(0.0, strategy[success_rate] - 0.1) continue # 如果所有策略都失败使用AI解析作为最后手段 if not results: results self._parse_with_ai(html, target_data) # 基于AI解析的结果学习新的解析规则 self._learn_from_ai_result(html, results, target_data) return results def _parse_with_ai(self, html: str, target_data: list) - dict: 使用AI解析 prompt f 请从以下HTML中提取数据 {html[:3000]} # 限制长度 需要提取的信息{, .join(target_data)} 输出格式JSON 要求如果某个信息找不到对应的值为null # 调用Qwen2.5 # return self._call_qwen(prompt) return {status: ai_parsed, data: {}} def _learn_from_ai_result(self, html: str, result: dict, target_data: list): 从AI解析结果中学习规则 # 分析AI是如何找到数据的 # 可以生成新的CSS选择器或XPath规则 # 添加到解析策略中 new_strategy { type: css_selectors, rules: self._generate_rules_from_result(html, result), success_rate: 0.7, # 初始成功率 last_used: 刚刚学习 } self.parsing_strategies.append(new_strategy)4. 数据清洗模板生成从混乱到规整抓下来的数据往往很乱价格可能有6999、6999元、6,999多种格式日期可能是2024-01-01、2024年1月1日、01/01/2024。手动写清洗规则太麻烦了而且每个网站都不一样。Qwen2.5可以帮我们自动生成数据清洗模板。基本思路是给模型一些示例数据让它学习清洗规则。4.1 自动清洗模板生成import re from datetime import datetime from typing import Any, Callable class DataCleaningTemplate: def __init__(self): self.templates {} def create_template_from_samples(self, field_name: str, dirty_samples: list, clean_samples: list None): 从样本数据创建清洗模板 prompt f 我需要清洗{field_name}字段的数据。 原始数据示例 {dirty_samples} {f清洗后的目标格式{clean_samples} if clean_samples else 请分析这些数据的模式并生成清洗规则。} 常见的数据清洗需求 1. 去除多余的空格和换行符 2. 统一货币符号和千位分隔符 3. 统一日期格式 4. 提取数字部分 5. 转换单位如1万 → 10000 请为这个字段生成清洗规则包括 1. 正则表达式如果需要 2. 转换函数逻辑 3. 验证规则 # 调用Qwen2.5生成清洗规则 # rules self._call_qwen(prompt) # 示例规则 if 价格 in field_name or price in field_name.lower(): rules self._create_price_cleaning_rules() elif 日期 in field_name or date in field_name.lower(): rules self._create_date_cleaning_rules() elif 销量 in field_name or sales in field_name.lower(): rules self._create_sales_cleaning_rules() else: rules self._create_general_cleaning_rules() self.templates[field_name] rules return rules def _create_price_cleaning_rules(self) - dict: 创建价格清洗规则 return { description: 清洗价格数据统一为浮点数, steps: [ { name: 去除货币符号, regex: r[¥$€£], replacement: }, { name: 去除千位分隔符, regex: r,, replacement: }, { name: 提取数字部分, regex: r(\d\.?\d*), extract_group: 1 }, { name: 单位转换, conditions: [ { if_contains: 万, multiply_by: 10000 }, { if_contains: 千, multiply_by: 1000 } ] } ], validation: { min_value: 0, max_value: 10000000, is_numeric: True } } def _create_date_cleaning_rules(self) - dict: 创建日期清洗规则 return { description: 清洗日期数据统一为YYYY-MM-DD格式, steps: [ { name: 统一中文日期, regex: r(\d{4})年(\d{1,2})月(\d{1,2})日, replacement: r\1-\2-\3 }, { name: 统一斜杠日期, regex: r(\d{4})/(\d{1,2})/(\d{1,2}), replacement: r\1-\2-\3 }, { name: 补齐月份和日期, function: pad_date_parts } ], validation: { date_format: %Y-%m-%d, min_year: 2000, max_year: 2030 } } def clean_data(self, field_name: str, value: Any) - Any: 使用模板清洗数据 if field_name not in self.templates: # 如果没有模板先创建一个简单的 self.create_template_from_samples(field_name, [str(value)]) rules self.templates[field_name] cleaned_value value for step in rules.get(steps, []): if regex in step: if replacement in step: # 替换操作 cleaned_value re.sub( step[regex], step[replacement], str(cleaned_value) ) elif extract_group in step: # 提取操作 match re.search(step[regex], str(cleaned_value)) if match: cleaned_value match.group(step[extract_group]) if conditions in step: for condition in step[conditions]: if condition[if_contains] in str(cleaned_value): try: cleaned_value float(cleaned_value) * condition[multiply_by] except: pass # 验证数据 if self._validate_data(field_name, cleaned_value, rules.get(validation, {})): return cleaned_value else: # 验证失败返回原始值或None return None def _validate_data(self, field_name: str, value: Any, validation_rules: dict) - bool: 验证清洗后的数据 if not validation_rules: return True try: # 检查数值范围 if min_value in validation_rules and max_value in validation_rules: num_value float(value) if not (validation_rules[min_value] num_value validation_rules[max_value]): return False # 检查日期格式 if date_format in validation_rules: datetime.strptime(str(value), validation_rules[date_format]) return True except: return False # 使用示例 cleaner DataCleaningTemplate() # 创建价格清洗模板 price_samples [6,999.00, 6999元, 约¥7000, 特价5999] cleaner.create_template_from_samples(商品价格, price_samples, [6999.0, 6999.0, 7000.0, 5999.0]) # 清洗数据 dirty_prices [¥7,200.50, 7200元, 七千二, 约7200] for price in dirty_prices: cleaned cleaner.clean_data(商品价格, price) print(f原始{price} - 清洗后{cleaned})4.2 智能类型推断有时候我们不知道数据应该是什么类型可以让模型帮忙推断def infer_data_type_and_clean(self, field_name: str, samples: list) - dict: 推断数据类型并生成清洗规则 prompt f 分析以下数据样本推断字段{field_name}的数据类型和清洗规则 样本数据 {samples} 请分析 1. 这是什么类型的数据价格、日期、数量、文本等 2. 数据有什么共同模式 3. 需要哪些清洗步骤 4. 目标格式应该是什么 输出JSON格式包含 - data_type: 数据类型 - patterns: 发现的模式 - cleaning_steps: 清洗步骤 - target_format: 目标格式 - validation_rules: 验证规则 # 调用Qwen2.5 # analysis self._call_qwen(prompt) # 模拟返回 analysis { data_type: price, patterns: [包含货币符号, 有千位分隔符, 可能有约、特价等前缀], cleaning_steps: [ 去除货币符号和文字前缀, 去除千位分隔符, 提取数字部分, 转换为浮点数 ], target_format: float, validation_rules: { min: 0, max: 1000000, decimal_places: 2 } } return analysis5. 实战案例电商商品数据采集让我们看一个完整的实战案例采集电商网站的商品数据。import time import random import pandas as pd from typing import List, Dict import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class EcommerceCrawler: def __init__(self, use_ai_assist: bool True): self.use_ai use_ai_assist self.header_generator SmartHeaderGenerator() self.parser AdaptiveParser() self.cleaner DataCleaningTemplate() # 初始化解析策略 self._init_parsing_strategies() def _init_parsing_strategies(self): 初始化解析策略 # 策略1CSS选择器针对常见电商网站 self.parser.add_parsing_strategy(css_selectors, { title: [.product-title, h1.product-name, [class*title]], price: [.product-price, .price, [class*price]], sales: [.sales-count, .sold, [class*sale]], comments: [.comment-count, .review, [class*comment]] }) # 策略2XPath备用策略 self.parser.add_parsing_strategy(xpath, { title: //h1[contains(class, title) or contains(class, name)], price: //*[contains(text(), ¥) or contains(text(), )], sales: //*[contains(text(), 已售) or contains(text(), 销量)] }) def crawl_product_page(self, url: str) - Dict: 爬取单个商品页面 logger.info(f开始爬取{url}) try: # 1. 生成智能请求头 headers self.header_generator.generate_headers(url) # 2. 发送请求添加随机延迟模拟真人 time.sleep(random.uniform(1, 3)) # 这里应该是实际的requests调用 # response requests.get(url, headersheaders, timeout10) # html response.text # 模拟HTML响应 html self._mock_html_response() # 3. 解析页面 target_fields [商品标题, 商品价格, 商品销量, 商品评价, 商品描述] if self.use_ai: # 使用AI辅助解析 parsed_data self.parser.parse_with_fallback(html, target_fields) else: # 传统解析 parsed_data self._traditional_parse(html) # 4. 数据清洗 cleaned_data {} for field, value in parsed_data.items(): if value: cleaned_value self.cleaner.clean_data(field, value) cleaned_data[field] cleaned_value # 5. 添加元数据 cleaned_data[来源URL] url cleaned_data[爬取时间] time.strftime(%Y-%m-%d %H:%M:%S) cleaned_data[爬取状态] 成功 logger.info(f爬取成功{url}) return cleaned_data except Exception as e: logger.error(f爬取失败{url}, 错误{str(e)}) return { 来源URL: url, 爬取时间: time.strftime(%Y-%m-%d %H:%M:%S), 爬取状态: f失败{str(e)} } def crawl_product_list(self, list_url: str, max_pages: int 3) - List[Dict]: 爬取商品列表 all_products [] for page in range(1, max_pages 1): logger.info(f爬取第{page}页) # 构造分页URL page_url f{list_url}?page{page} # 爬取列表页获取商品链接 # 这里简化处理实际应该解析列表页获取商品链接 product_urls self._extract_product_urls(page_url) for product_url in product_urls[:5]: # 每页只爬前5个避免太快 product_data self.crawl_product_page(product_url) all_products.append(product_data) # 随机延迟避免请求过快 time.sleep(random.uniform(2, 5)) return all_products def export_to_excel(self, data: List[Dict], filename: str): 导出到Excel df pd.DataFrame(data) # 重新排列列让重要信息在前 preferred_order [商品标题, 商品价格, 商品销量, 商品评价, 商品描述, 来源URL, 爬取时间] existing_columns [col for col in preferred_order if col in df.columns] other_columns [col for col in df.columns if col not in preferred_order] df df[existing_columns other_columns] # 保存到Excel df.to_excel(filename, indexFalse) logger.info(f数据已导出到{filename}) # 生成数据报告 self._generate_report(df) def _generate_report(self, df: pd.DataFrame): 生成数据质量报告 report { 总记录数: len(df), 成功爬取数: len(df[df[爬取状态] 成功]), 失败爬取数: len(df[df[爬取状态] ! 成功]), 价格范围: f{df[商品价格].min():.2f} - {df[商品价格].max():.2f} if 商品价格 in df.columns else 无, 平均销量: df[商品销量].mean() if 商品销量 in df.columns else 无, 字段完整度: {} } # 计算每个字段的完整度 for column in df.columns: if column not in [来源URL, 爬取时间, 爬取状态]: non_null_count df[column].notnull().sum() completeness non_null_count / len(df) * 100 report[字段完整度][column] f{completeness:.1f}% logger.info(数据质量报告) for key, value in report.items(): if isinstance(value, dict): logger.info(f {key}:) for sub_key, sub_value in value.items(): logger.info(f {sub_key}: {sub_value}) else: logger.info(f {key}: {value}) def _mock_html_response(self) - str: 模拟HTML响应用于演示 return html div classproduct-detail h1 classproduct-title华为Mate 60 Pro 智能手机 12GB512GB/h1 div classprice-section span classcurrent-price6999.00/span span classoriginal-price7999.00/span /div div classsales-info span已售1.5万件/span span评价2345条/span /div div classproduct-description 华为Mate 60 Pro搭载麒麟9000S芯片支持卫星通话拥有超可靠玄武架构。 /div /div /html def _extract_product_urls(self, list_url: str) - List[str]: 从列表页提取商品URL模拟 # 实际应该解析列表页 return [ https://item.jd.com/1000001.html, https://item.jd.com/1000002.html, https://item.jd.com/1000003.html ] def _traditional_parse(self, html: str) - Dict: 传统解析方法 from bs4 import BeautifulSoup soup BeautifulSoup(html, html.parser) data {} # 尝试多种选择器 title_selectors [.product-title, h1, [class*title]] for selector in title_selectors: element soup.select_one(selector) if element: data[商品标题] element.get_text(stripTrue) break # 类似地处理其他字段... return data # 使用示例 if __name__ __main__: # 创建爬虫实例 crawler EcommerceCrawler(use_ai_assistTrue) # 爬取商品列表 list_url https://list.jd.com/手机 products crawler.crawl_product_list(list_url, max_pages2) # 导出数据 crawler.export_to_excel(products, 手机商品数据.xlsx) print(f共爬取{len(products)}条商品数据)6. 经验总结与实用建议实际用下来Qwen2.5-32B-Instruct在爬虫开发中确实能帮上大忙但也不是万能的。结合我这段时间的使用经验给大家几点建议关于请求头优化不要完全依赖模型生成最好还是维护一个真实的请求头池。模型生成可以作为补充特别是在遇到新的反爬策略时。建议把模型生成的请求头和实际抓包获取的请求头结合起来用。关于页面解析AI解析适合结构复杂或经常变动的页面但对于结构稳定的网站传统解析方法更快更可靠。我的做法是先用传统方法如果失败了再fallback到AI解析。AI解析的结果可以用来训练新的传统解析规则。关于数据清洗让模型学习清洗规则是个好主意但一定要有人工验证环节。模型可能会“过度清洗”或者误解数据。建议先在小批量数据上测试清洗效果确认无误后再应用到全部数据。性能考虑AI解析比较耗资源如果爬虫规模很大要控制使用频率。可以设置一个阈值比如连续失败3次后才启用AI解析或者只在关键字段上使用AI辅助。错误处理爬虫中的错误处理特别重要。网络超时、页面结构变化、反爬拦截……各种问题都可能出现。建议实现完善的日志记录和重试机制AI可以帮助分析失败原因并调整策略。法律合规这个必须强调爬虫开发一定要遵守法律法规和网站的robots.txt。只爬取公开数据控制请求频率不要对网站造成压力。Qwen2.5可以帮助生成更“友好”的爬虫代码但合规意识还是要靠我们自己。整体来说Qwen2.5-32B-Instruct作为爬虫开发的辅助工具在智能化和适应性方面确实有优势。特别是对于那些页面结构复杂、反爬策略多样的网站AI辅助能显著提高开发效率和爬虫的健壮性。不过它也不是要完全替代传统爬虫技术而是作为一种增强手段。两者结合使用效果最好。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。