Python网络爬虫技术项目化教程 课件 项目七 应对反爬虫策略 -“突破封锁持续抓取”

项目7应对反爬虫策略–“突破封锁，持续抓取”任务7.1常见反爬虫机制与应对策略7.1.1使用代理IP池避免IP封锁代理IP池核心概念定义：集中管理大量代理IP资源的基础设施，通过动态调度技术为爬虫程序提供动态、高可用的代理服务目的：分散请求来源，绕过IP封锁、地域限制等反爬机制本质：模拟多用户访问行为，将单一IP的高频请求分散到多个代理节点7.1.1使用代理IP池避免IP封锁代理IP池的工作原理1.代理IP获取与存储——资源层获取渠道：公开渠道：爬取西刺代理、快代理等免费代理网站付费服务：对接阿布云、讯代理等商业服务商的API自建节点：借助VPN或AWS、阿里云等云服务器搭建存储信息：地址、端口、协议类型、地域、响应速度等分类维度：地域、协议、匿名级别等7.1.1使用代理IP池避免IP封锁代理IP池的工作原理2.代理IP质量监控与管理——控制层健康检查：定时检测：定期向代理IP发送测试请求多维度评估：可用性、响应速度、匿名性、稳定性动态维护：剔除失效IP：连续多次检测失败或质量低于阈值补充新IP：定期从来源渠道获取新IP替换失效资源权重调整：为低延迟、高成功率的高质量IP赋予更高优先级7.1.1使用代理IP池避免IP封锁代理IP池的工作原理3.代理IP动态调度与使用——服务层请求调度策略：随机选择：分散请求压力地域优先：根据目标网站地域分布选择同地域IP以降低延迟轮询策略：按顺序使用IP避免过度调用负载均衡：根据IP当前负载动态分配请求请求转发流程：从代理池选择可用IP将请求封装到该IP的连接中发送若请求失败自动切换至下一个IP重试标记失败IP为临时不可用User-Agent概念HTTP请求头中的一个字段，用于标识客户端身份信息包含：操作系统（如Windows、macOS、Linux）浏览器类型及版本（如Chrome114、Firefox109）渲染引擎（如WebKit、Blink）设备类型（如PC、Mobile、Tablet）服务器通过User-Agent判断请求是否来自真实用户7.1.2伪装User-Agent模拟真实浏览器行为User-Agent的获取方法方法一：使用浏览器的内置功能Chrome浏览器：在地址栏输入about:version显示浏览器的详细信息，包含UserAgent字符串适用范围：大多数现代浏览器7.1.2伪装User-Agent模拟真实浏览器行为User-Agent的获取方法方法二：使用开发者工具打开浏览器访问任意网页右键点击页面选择"检查"或按F12切换到"Network"选项卡，刷新页面点击任意请求，在"Headers"选项卡中找到UserAgent字段7.1.2伪装User-Agent模拟真实浏览器行为常用的User-Agent示例Chrome浏览器Windows：Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/58.0.3029.110Safari/537.36macOS：Mozilla/5.0(Macintosh;IntelMacOSX10_15_3)AppleWebKit/537.36(KHTML,likeGecko)Chrome/80.0.3987.132Safari/537.36Android：Mozilla/5.0(Linux;Android10;Pixel2)AppleWebKit/537.36(KHTML,likeGecko)Chrome/79.0.3945.93MobileSafari/537.367.1.2伪装User-Agent模拟真实浏览器行为常用的User-Agent示例Firefox浏览器Windows：Mozilla/5.0(WindowsNT10.0;Win64;x64;rv:75.0)Gecko/20100101Firefox/75.0macOS：Mozilla/5.0(Macintosh;IntelMacOSX10.15;rv:75.0)Gecko/20100101Firefox/75.0Safari浏览器macOS：Mozilla/5.0(Macintosh;IntelMacOSX10_15_3)AppleWebKit/605.1.15(KHTML,likeGecko)Version/13.0.5Safari/605.1.15iOS：Mozilla/5.0(iPhone;CPUiPhoneOS13_3likeMacOSX)AppleWebKit/605.1.15(KHTML,likeGecko)Version/13.0.4Mobile/15E148Safari/604.17.1.2伪装User-Agent模拟真实浏览器行为常用的User-Agent示例Edge浏览器Windows：Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/64.0.3282.140Safari/537.36Edge/17.17134Opera浏览器Windows：Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/58.0.3029.110Safari/537.36OPR/45.0.2552.898InternetExplorer浏览器Windows7：Mozilla/5.0(WindowsNT6.1;WOW64;Trident/7.0;rv:11.0)likeGecko7.1.2伪装User-Agent模拟真实浏览器行为任务7.2Python搭建代理IP池7.2.1收集代理从免费代理网站、付费代理服务、爬取公开代理列表、自建代理服务器获取示例代码：importrequestsfrombs4importBeautifulSoupimporttimedeffetch_free_proxies_simple(base_url,max_pages=3):headers={"User-Agent":"Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/Safari/537.36"}all_proxies=[]forpageinrange(1,max_pages+1):#直接构造URLif'?'inbase_url:page_url=f"{base_url}&page={page}&per_page=10"else:page_url=f"{base_url}?page={page}&per_page=10"print(f"正在爬取第{page}页:{page_url}")try:response=requests.get(page_url,headers=headers,timeout=10)response.raise_for_status()soup=BeautifulSoup(response.text,'html.parser')#提取代理

proxies=[]table=soup.find('table')iftable:forrowintable.find_all('tr')[1:]:#跳过表头

cols=row.find_all('td')iflen(cols)>=2:ip=cols[0].text.strip()port=cols[1].text.strip()ifipandport:proxies.append(f"{ip}:{port}")all_proxies.extend(proxies)print(f"第{page}页爬取完成，获取到{len(proxies)}个代理")time.sleep(1)#礼貌性延迟

exceptExceptionase:print(f"爬取第{page}页时出错:{e}")continuereturnall_proxiesurl="https://proxy.scdn.io/"proxies=fetch_free_proxies_simple(url,max_pages=3)print(f"总共爬取到{len(proxies)}个代理")print(proxies)7.2.1收集代理运行结果构建一个真正高效、稳定的代理池，需要注意：IP资源的多样性IP的质量自动化管理合理的负载均衡7.2.2验证代理连通性验证：测试能否成功连接目标服务器示例代码：importrequestsfromipimportproxiesdefis_proxy_working(proxy):url='/ip'proxies={'http':f'http://{proxy}','https':f'https://{proxy}'}try:response=requests.get(url,proxies=proxies,timeout=10)ifresponse.status_code==200:returnTrueexcept:returnFalsereturnFalseworking_proxies=[proxyforproxyinproxiesifis_proxy_working(proxy)]print(working_proxies)7.2.2验证代理速度测试：测量响应时间，筛选高效IP示例代码：importtimeimportrequestsfromworkingimportworking_proxiesdefget_proxy_speed(proxy):url='/ip'proxies={'http':f'http://{proxy}','https':f'https://{proxy}'}start_time=time.time()try:response=requests.get(url,proxies=proxies,timeout=10)ifresponse.status_code==200:returntime.time()-start_timeexcept:returnfloat('inf')returnfloat('inf')proxy_speeds={proxy:get_proxy_speed(proxy)forproxyinworking_proxies}print(proxy_speeds)7.2.2验证代理匿名性验证：区分透明代理、普通匿名代理、高匿代理示例代码：importrequestsfromworkingimportworking_proxiesdefcheck_proxy_anonymity(proxy):url='/headers'proxies={'http':f'http://{proxy}','https':f'https://{proxy}'}try:response=requests.get(url,proxies=proxies,timeout=10)ifresponse.status_code==200:headers=response.json().get('headers',{})#检查代理相关头部

proxy_headers=['Via','X-Forwarded-For','X-Real-IP','Proxy-Connection']forheaderinproxy_headers:ifheaderinheaders:return'anonymous'#普通匿名代理

return'elite'#高匿代理

else:return'unknown'except:return'unknown'#批量检测print("开始检测代理匿名性...")results={}fori,proxyinenumerate(working_proxies):print(f"检测{i+1}/{len(working_proxies)}:{proxy}")anonymity=check_proxy_anonymity(proxy)results[proxy]=anonymityprint(f"->{anonymity}")#显示结果print("\n===最终结果===")forproxy,anonymityinresults.items():print(f"{proxy}:{anonymity}")#统计elite=sum(1forvinresults.values()ifv=='elite')anonymous=sum(1forvinresults.values()ifv=='anonymous')print(f"\n高匿代理:{elite}")print(f"普通匿名:{anonymous}")print(f"未知/无效:{len(results)-elite-anonymous}")7.2.3管理代理通过数据库（SQLite、Redis、MySQL）对验证通过的代理进行持久化存储和管理示例代码：importsqlite3fromspeedimportproxy_speedsfromanonymityimportcheck_proxy_anonymitydefstore_proxies(proxies_speeds):#连接SQLite数据库（如果不存在则自动创建）

conn=sqlite3.connect('proxies.db')cursor=conn.cursor()#创建代理IP表

cursor.execute('CREATETABLEIFNOTEXISTSproxies(ipTEXTPRIMARYKEY,speedREAL,anonymityTEXT)')#清空现有数据

cursor.execute('DELETEFROMproxies')#插入新的代理IP数据

forproxy,speedinproxies_speeds.items():anonymity=check_proxy_anonymity(proxy)cursor.execute('INSERTINTOproxies(ip,speed,anonymity)VALUES(?,?,?)',(proxy,speed,anonymity))print(f"存储代理:{proxy},速度:{speed},匿名性:{anonymity}")#提交事务并关闭连接

mit()conn.close()print("所有代理数据已成功存储到数据库")#调用函数存储代理IPstore_proxies(proxy_speeds)7.2.4使用代理通过代理池管理工具实现代理的自动分配、负载均衡和失效剔除示例代码：importrandomimportrequestsimportsqlite3#获取随机代理conn=sqlite3.connect('proxies.db')cursor=conn.cursor()cursor.execute('SELECTipFROMproxiesORDERBYspeedASCLIMIT10')proxies_list=[row[0]forrowincursor.fetchall()]conn.close()ifproxies_list:proxy=random.choice(proxies_list)print(f"使用代理:{proxy}")

proxies={'http':f'http://{proxy}','https':f'https://{proxy}'}

#访问网站

response=requests.get('/ip',proxies=proxies,timeout=10)print(f"状态码:{response.status_code}")print("响应内容:")print(response.text)else:print("没有可用的代理")任务7.3处理简单验证码7.3.1安装和配置TesseractTesseract安装步骤1.进入下载页面：选择最新版本，如tesseract-ocr-w64-setup-0240606.exe7.3.1安装和配置TesseractTesseract安装步骤2.语言选择界面：选择"English"单击"OK"7.3.1安装和配置TesseractTesseract安装步骤3.安装界面：单击"Next"继续7.3.1安装和配置TesseractTesseract安装步骤4.许可协议界面：单击"IAgree"同意安装协议7.3.1安装和配置TesseractTesseract安装步骤5.选择用户界面：选择"Installforanyoneusingthiscomputer"7.3.1安装和配置TesseractTesseract安装步骤6.选择组件界面：找到"Additionallanguagedata(download)"7.3.1安装和配置TesseractTesseract安装步骤7.选择中文语言：Chinese(Simplified)-简体中文；Chinese(SimplifiedVertical)-竖排简体中文Chinese(Traditional)-繁体中文；Chinese(TraditionalVertical)-竖排繁体中文7.3.1安装和配置TesseractTesseract安装步骤8.选择安装路径：可选择保持默认或自定义，如安装到D盘7.3.1安装和配置TesseractTesseract安装步骤9.选择开始菜单文件夹：保持默认，单击"Install"7.3.1安装和配置TesseractTesseract安装步骤10.安装完成：等待安装完成后单击"Next"，最后单击"Finish"7.3.1安装和配置Tesseract环境配置1.配置Path环境变量：打开系统环境变量配置窗口：此电脑→属性→高级系统设置→环境变量在"系统变量"中找到"Path"，点击"编辑"新增：D:\ProgramFiles\Tesseract-OCR点击"确定"保存7.3.1安装和配置Tesseract环境配置2.新建系统变量：新建系统变量TESSDATA_PREFIX变量值：D:\ProgramFiles\Tesseract-OCR\tessdata点击"确定"保存7.3.1安装和配置Tesseract验证安装命令：tesseract–version运行结果：输出显示Tesseract版本信息7.3.1安装和配置Tesseract安装Pillow和pytesseract命令：pipinstallpillowpipinstallpytesseract7.3.2使用Tesseract识别验证码验证码图片示例：简单验证码识别代码：fromPILimportImageimportpytesseract#导入一个png图片img=Image.open('png/modified_image_path.png')print(img)#识别读取图片的字母和数字text=pytesseract.image_to_string(img)print(text)7.3.2使用Tesseract识别验证码网站中常见的验证码类型1.文字验证码特点：由一串随机生成的字母、数字或字符组成难度：可能添加干扰线示例：7.3.2使用Tesseract识别验证码网站中常见的验证码类型2.图形验证码特点：通过展示图像要求用户根据图像内容完成验证类型：选择包含某物体的图片、识别扭曲文字等示例：7.3.2使用Tesseract识别验证码网站中常见的验证码类型3.滑动验证码特点：分为背景图和前景图，拖动前景图到背景图缺口处机制：通过鼠标拖动下方的箭头带动前景图移动示例：任务7.4利用aiohttp框架实现高效数据抓取7.4.1同步编程程序按照顺序依次执行每一个任务，在执行完一个任务后，才会开始执行下一个任务。示例代码：importdatetimeimporttimeimportrequests#同步请求defmain():start=time.time()foriinrange(5):res=requests.get('/delay/2')print(f'当前时间：{datetime.datetime.now()},status_code={res.status_code}')print(f'requests同步耗时：{time.time()-start}')if__name__=='__main__':main()7.4.2异步编程1.协程（Coroutine）：异步任务的基本单元可以暂停执行并在稍后恢复的函数使用asyncdef定义协程函数内部用await来暂停并等待其他协程的结果2.事件循环（EventLoop）：异步编程的核心循环结构，负责管理任务的调度不断监听并处理事件在Python中使用asyncio模块提供的事件循环7.4.3aiohttp框架aiohttp核心优势：基于asyncio实现，支持高并发请求降低I/O等待时间，提升爬取效率适合大规模数据采集场景安装aiohttp命令：pipinstallaiohttppipinstallasyncio7.4.3aiohttp框架使用aiohttp进行异步HTTP请求示例代码：importasyncioimportdatetimeimporttimeimportaiohttpasyncdefasync_http():asyncwithaiohttp.ClientSession()assession:res=awaitsession.get('/delay/2')print(f'当前时间：{datetime.datetime.now()},status_code={res.status}')asyncdefmain():tasks=[asyncio.create_task(async_http())for_inrange(5)]awaitasyncio.gather(*tasks)if__name__=="__main__":start=time.time()asyncio.run(main())#运行主协程

print(f'aiohttp异步耗时：{time.time()-start}')7.4.3aiohttp框架异步爬虫实战：批量抓取网页数据importasyncioimportaiohttpfrombs4importBeautifulSoupasyncdeffetch(url):asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url)asresponse:html=awaitresponse.text()soup=BeautifulSoup(html,'html.parser')title=soup.title.stringprint(f"URL:{url},Title:{title}")asyncdefmain():urls=["","",""]awaitasyncio.gather(*(fetch(url)forurlinurls))asyncio.run(main())7.4.4aiohttp进阶应用与优化设置请求超时时间importasyncioimportaiohttpasyncdefmain():timeout=aiohttp.ClientTimeout(total=2)asyncwithaiohttp.ClientSession(timeout=timeout)assession:asyncwithsession.get('/get')asresponse:print('status:',response.status)if__name__=='__main__':asyncio.run(main())7.4.4aiohttp进阶应用与优化处理异常importaiohttpimportasyncioasyncdeffetch(url):asyncwithaiohttp.ClientSession()assession:try:asyncwithsession.get(url)asresponse:response.raise_for_status()#如果响应状态码不是200，将抛出异常

returnawaitresponse.text()exceptaiohttp.ClientErrorase:print(f'Anerroroccurred:{e}')returnNoneasyncdefmain():urls=["",""]tasks=[fetch(url)forurlinurls]awaitasyncio.gather(*tasks)if__name__=='__main__':asyncio.run(main())7.4.4aiohttp进阶应用与优化限制并发请求数量importasynciofromdatetimeimportdatetimeimportaiohttpsemaphore=asyncio.Semaphore(2)#设置最大并发数为2asyncdefget_api():asyncwithsemaphore:print(f'scrapting...{datetime.now()}')asyncwithsession.get('')asresponse:awaitasyncio.sleep(2)asyncdefmain():globalsessionsession=aiohttp.ClientSession()tasks=[asyncio.ensure_future(get_api())for_inrange(100)]awaitasyncio.gather(*tasks)awaitsession.close()if__name__=="__main__":asyncio.run(main())项目实战

爬取Scrape书籍网站【任务描述】使用Python的aiohttp框架开发异步爬虫，爬取Scrape书籍网站的书籍信息，实现高效数据采集、解析与存储。【任务实施】1.页面分析网站特点：包含数千本图书信息网站数据是JavaScript渲染而得数据可以通过Ajax接口获取接口没有设置反爬措施和加密参数页面加载方式全部为Ajax加载【任务实施】1.页面分析列表页Ajax请求接口格式：https://spa5.scrape.center/api/book/?limit=18&offset=0limit参数：每页包含多少本书offset参数：每一页的偏移量计算公式：offset=limit*(1)示例：第一页offset值为0，第2页offset值为18，以此类推【任务实施】1.页面分析列表页数据特征results字段：包含当前页里18本图书的信息每本书数据包含id字段：图书本身的ID，用于进一步请求详情页【任务实施】1.页面分析详情页Ajax请求接口格式：https://spa5.scrape.center/api/book/{id}id参数：详情页对应图书的ID可以从列表页Ajax接口的返回结果中获取【任务实施】2.实现思路异步爬虫设计原则维护一个动态变化的爬取队列每产生一个新的task，就将其放入爬取队列中专门的爬虫消费者从此队列中获取task并执行在最大并发量的前提下充分利用等待时间【任务实施】2.实现思路爬取逻辑拆分第一阶段：异步爬取所有列表页将所有列表页的爬取任务集合在一起声明为由task组成的列表进行异步爬取第二阶段：爬取所有详情页解析列表页获取所有图书id将所有图书id组合为详情页的爬取任务集合声明为task组成的列表进行异步爬取同时以异步方式存储到文件中【任务实施】3.基本配置importasyncioimportaiohttpimportlogginglogging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s')INDEX_URL='https://spa5.scrape.center/api/book/?limit=18&offset={offset}'DETAIL_URL='https://spa5.scrape.center/api/book/{id}'PAGE_SIZE=18PAGE_NUMBER=100CONCURRENCY=5【任务实施】4.爬取列表页Semaphore=asyncio.Semaphore(CONCURRENCY)session=Noneasyncdefscrape_api(url):asyncwithsemaphore:try:('scraping%s',url)asyncwithsession.get(url)asresponse:returnawaitresponse.json()exceptaiohttp.ClientError:logging.error('erroroccurredwhilescraping%s',url,exc_info=True)asyncdefscrape_index(page):url=INDEX_URL.format(offset=PAGE_SIZE*(1))returnawaitscrape_api(url)importjsonasyncdefmain():globalsessionsession=aiohttp.ClientSession()

scrape_index_tasks=[asyncio.ensure_future(scrape_index(page))forpageinrange(1,PAGE_NUMBER+1)]

results=awaitasyncio.gather(*scrape_index_tasks)('results%s',json.dumps(results,ensure_ascii=False,indent=2))if__name__=='__main__':asyncio.run(main())【任务实施】4.爬取列表页运行结果：开始异步爬取，并发量为5，爬取速度会进一步加快【任务实施】5.爬取详情页importasyncioimportaiohttpimportaiofilesimportloggingimportjsonlogging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s')INDEX_URL='https://spa5.scrape.center/api/book/?limit=18&offset={offset}'DETAIL_URL='https://spa5.scrape.center/api/book/{id}'PAGE_SIZE=18PAGE_NUMBER=100CONCURRENCY=5FILE_NAME='books.json'#移除全局semaphore和session的声明file_lock=Nonesemaphore=Nonesession=Noneasyncdefscrape_api(url):asyncwithsemaphore:#这里使用main()中创建的semaphoretry:('scraping%s',url)asyncwithsession.get(url)asresponse: