Python爬虫实战：二手平台数据采集与反爬策略

1. 爬虫项目概述与法律边界在当今数据驱动的商业环境中获取公开市场数据对于价格监测、竞品分析和市场研究具有重要意义。本次项目以某二手交易平台以下简称目标平台的商品数据采集为例演示如何通过Python技术栈实现合规的网页数据采集。需要特别强调的是所有爬虫操作必须严格遵循以下原则遵守目标平台的robots.txt协议通常位于网站根目录下如https://example.com/robots.txt请求频率控制在人类操作范围内建议单次采集间隔不低于3-5秒仅采集公开可见数据不尝试突破任何访问限制采集的数据仅用于个人学习和技术研究不得用于商业用途重要提示在开始任何爬虫项目前建议使用requests库检查目标网站的robots.txt文件。例如import requests robots requests.get(https://目标网站.com/robots.txt).text print(robots)2. 技术选型与工具准备2.1 核心工具链配置本项目采用Python 3.8环境主要依赖以下工具包pip install requests2.31.0 # 网络请求 pip install pandas2.0.3 # 数据处理 pip install hashlib # 加密算法内置库 pip install fake-useragent # 随机User-Agent生成2.2 开发辅助工具浏览器开发者工具Chrome/Firefox快捷键F12打开Network面板用于监控HTTP请求Sources面板用于调试JavaScriptJSON格式化工具推荐VS Code插件或在线工具帮助解析复杂的API响应结构时间戳转换工具用于验证请求参数中的时间戳3. 逆向工程全流程解析3.1 接口定位与数据分析通过浏览器开发者工具的Network面板我们可以观察到目标平台采用典型的前后端分离架构。商品数据通过XHR接口返回关键发现包括搜索接口URL模式https://api.example.com/search?q关键词page页码响应数据结构示例{ code: 200, data: { total: 1200, items: [ { id: 123456, title: 二手iPhone 12, price: ¥1999, seller: 用户***, location: 北京 } ] } }3.2 加密参数逆向实战目标平台的防爬机制主要依赖动态签名sign参数通过逆向分析发现其生成逻辑如下签名原始字符串构成token timestamp fixed_salt json_payloadMD5加密实现代码import hashlib def generate_sign(token: str, timestamp: str, payload: str) - str: salt 固定盐值 # 通过逆向获得 raw f{token}{timestamp}{salt}{payload} return hashlib.md5(raw.encode()).hexdigest()参数获取技巧token可从首次请求的Cookie中提取timestamp使用当前毫秒级时间戳payload为JSON序列化的查询参数3.3 请求头精细化配置模拟真实浏览器的请求头能显著降低被封风险推荐配置headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://目标网站.com/search, Accept: application/json, X-Requested-With: XMLHttpRequest, Cookie: 获取到的有效cookie }4. 完整爬虫实现4.1 核心采集逻辑import time import json import pandas as pd from typing import List, Dict class ItemSpider: def __init__(self, keyword: str, max_pages: int 10): self.base_url https://api.example.com/search self.keyword keyword self.max_pages max_pages self.session requests.Session() self.session.headers.update(self._gen_headers()) def crawl(self) - List[Dict]: all_items [] for page in range(1, self.max_pages 1): items self._fetch_page(page) if not items: break all_items.extend(items) time.sleep(5) # 遵守爬虫礼仪 return all_items def _fetch_page(self, page: int) - List[Dict]: params self._build_params(page) try: resp self.session.get(self.base_url, paramsparams) resp.raise_for_status() return self._parse_items(resp.json()) except Exception as e: print(f第{page}页采集失败: {str(e)}) return [] def _parse_items(self, data: Dict) - List[Dict]: 解析接口返回的JSON数据 return [ { title: item.get(title), price: float(item.get(price, 0).replace(¥, )), location: item.get(location), timestamp: pd.Timestamp.now() } for item in data.get(data, {}).get(items, []) ]4.2 数据存储方案提供三种存储方式供选择CSV存储适合小型数据集def save_to_csv(items: List[Dict], path: str): pd.DataFrame(items).to_csv(path, indexFalse)JSON存储保留原始结构def save_to_json(items: List[Dict], path: str): with open(path, w, encodingutf-8) as f: json.dump(items, f, ensure_asciiFalse)数据库存储推荐MySQLimport sqlalchemy def save_to_mysql(items: List[Dict]): engine sqlalchemy.create_engine(mysql://user:passlocalhost/db) pd.DataFrame(items).to_sql(items, engine, if_existsappend)5. 高级反反爬策略5.1 IP轮换方案from itertools import cycle proxies cycle([ http://proxy1.example.com:8080, http://proxy2.example.com:8080 ]) def get_with_proxy(url): proxy next(proxies) try: return requests.get(url, proxies{http: proxy}, timeout10) except: return get_with_proxy(url) # 自动切换下一个代理5.2 浏览器自动化方案当遇到复杂JavaScript渲染时可使用Seleniumfrom selenium.webdriver import Chrome from selenium.webdriver.chrome.options import Options opts Options() opts.add_argument(--headless) # 无界面模式 driver Chrome(optionsopts) driver.get(https://目标网站.com) search driver.find_element(name, q) search.send_keys(手机) search.submit() # 解析渲染后的DOM items driver.find_elements(css selector, .item)6. 常见问题排查指南6.1 请求被拒绝403错误可能原因及解决方案签名过期检查时间戳是否在有效期内通常±5分钟User-Agent被识别使用fake-useragent随机生成from fake_useragent import UserAgent ua UserAgent() headers {User-Agent: ua.random}6.2 数据解析失败调试建议保存原始响应到文件with open(debug.json, w) as f: json.dump(response.json(), f)使用JSONPath快速定位数据from jsonpath_ng import parse expr parse($.data.items[*].title) titles [match.value for match in expr.find(response.json())]6.3 频率限制应对阶梯式退避策略实现import random from time import sleep def safe_request(url): retries 3 for attempt in range(retries): try: return requests.get(url) except Exception as e: wait (2 ** attempt) random.random() print(f请求失败等待{wait:.2f}秒后重试) sleep(wait) raise Exception(超过最大重试次数)7. 项目优化方向分布式采集使用Scrapy-Redis实现多机协同增量采集记录最后采集时间只获取新数据数据清洗使用pandas处理价格异常值df pd.DataFrame(items) df df[df[price].between(100, 10000)] # 过滤异常价格可视化监控使用PrometheusGrafana监控采集状态在实际操作中发现目标平台大约每2周会更新一次签名算法。保持对接口变化的监控非常重要可以通过定期运行测试用例来验证采集逻辑的有效性。建议将核心采集逻辑封装为独立模块方便后续维护升级。