东城区网站建设上海公司企业查询

东城区网站建设,上海公司企业查询,大学软件开发专业,网站信息化建设前言 在爬虫应用场景中#xff0c;单页面爬取仅能满足简单的数据采集需求#xff0c;而批量爬取多页面、多目标网址的数据才是解决实际业务问题的核心能力。批量爬取的核心挑战在于如何高效管理待爬取 URL 队列、控制爬取节奏、避免重复爬取#xff0c;并保证大规模数据采集…前言在爬虫应用场景中单页面爬取仅能满足简单的数据采集需求而批量爬取多页面、多目标网址的数据才是解决实际业务问题的核心能力。批量爬取的核心挑战在于如何高效管理待爬取 URL 队列、控制爬取节奏、避免重复爬取并保证大规模数据采集的稳定性。本文将从批量爬取的核心原理入手以「当当网图书畅销榜」http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-为实战目标系统讲解批量爬取网页的基础实现方法涵盖 URL 队列构建、分页批量爬取、多 URL 批量爬取、数据批量存储等核心环节帮助开发者掌握批量爬取的核心逻辑与落地技巧。摘要本文以当当网图书畅销榜为爬取对象详细阐述 Python 批量爬取网页的两种核心模式分页批量爬取、多 URL 列表批量爬取通过实战代码实现从 URL 队列构建、批量请求发送、批量数据解析到批量数据存储的全流程。内容涵盖队列管理、爬取频率控制、批量异常处理、去重机制等关键技术点同时对比不同批量爬取模式的适用场景最终实现高效、稳定的批量数据采集为后续复杂批量爬取场景如多站点、千万级 URL奠定基础。一、批量爬取核心原理与模式1.1 批量爬取核心原理批量爬取本质是「URL 队列 循环请求 批量处理」的组合核心流程如下构建 URL 队列将所有待爬取的 URL 整理为有序队列列表 / 队列数据结构循环请求遍历 URL 队列逐个 / 批量发送 HTTP 请求获取网页内容批量解析对多个页面的响应内容进行统一解析提取结构化数据批量存储将解析后的多页数据统一写入文件 / 数据库减少 IO 操作异常管控对批量请求中的异常 URL 进行单独处理不影响整体流程。1.2 批量爬取的两种基础模式模式适用场景核心特点示例分页批量爬取目标数据分布在连续分页中如榜单、列表页URL 有固定分页规律可通过公式生成所有 URL电商榜单第 1-10 页、论坛帖子列表多 URL 列表批量爬取目标 URL 无规律需从外部获取如链接列表、CSV 文件URL 需提前整理为列表遍历爬取竞品网站多个详情页、新闻专题不同报道1.3 批量爬取关键注意事项频率控制批量爬取易触发反爬需严格控制请求间隔建议 1-3 秒 / 次去重机制避免重复爬取同一 URL可通过集合 / 哈希表记录已爬取 URL断点续爬大规模批量爬取时记录已爬取进度避免程序中断后重新爬取内存控制批量解析 / 存储时避免一次性加载所有数据采用分批处理异常隔离单个 URL 请求 / 解析失败不终止整个批量爬取流程。二、实战 1分页批量爬取当当网图书畅销榜2.1 目标分析当当网图书畅销榜分页 URL 规律第 1 页http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-1第 2 页http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-2...第 N 页http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-N核心规律URL 末尾数字为页码需爬取 1-20 页的图书数据。2.2 完整代码实现python运行import requests from bs4 import BeautifulSoup import csv import time import logging from fake_useragent import UserAgent from typing import List, Dict, Optional # 基础配置 # 日志配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler(), logging.FileHandler(batch_crawl.log, encodingutf-8)] ) logger logging.getLogger(__name__) # 爬取配置 BASE_URL http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1- PAGE_START 1 # 起始页码 PAGE_END 20 # 结束页码 CSV_PATH dangdang_bestsellers.csv HEADERS { User-Agent: UserAgent().random, Accept-Language: zh-CN,zh;q0.9, Accept-Encoding: gzip, deflate, Connection: keep-alive } REQUEST_TIMEOUT 10 # 请求超时时间 CRAWL_INTERVAL 1 # 爬取间隔秒 BATCH_SIZE 5 # 批量存储大小每5页存储一次 # 核心函数 def build_page_urls(start: int, end: int) - List[str]: 构建分页URL列表 :param start: 起始页码 :param end: 结束页码 :return: 分页URL列表 url_list [] for page in range(start, end 1): url f{BASE_URL}{page} url_list.append(url) logger.info(fURL队列构建完成共生成{len(url_list)}个分页URL) return url_list def fetch_single_page(url: str) - Optional[str]: 获取单页HTML内容带异常处理 :param url: 目标URL :return: HTML文本失败返回None try: response requests.get( urlurl, headersHEADERS, timeoutREQUEST_TIMEOUT ) response.raise_for_status() response.encoding response.apparent_encoding logger.info(f成功获取页面{url}) return response.text except requests.exceptions.RequestException as e: logger.error(f获取页面失败{url}异常{str(e)[:100]}) return None def parse_single_page(html: str) - List[Dict]: 解析单页图书数据 :param html: HTML文本 :return: 图书数据列表解析失败返回空列表 if not html: return [] book_list [] try: soup BeautifulSoup(html, html.parser) # 定位图书条目 book_items soup.find_all(li, class_bang_list clearfix bang_list_mode) for item in book_items: # 提取核心字段容错处理 rank_tag item.find(div, class_list_num red) rank rank_tag.text.strip() if rank_tag else 未知 name_tag item.find(div, class_name).find(a) book_name name_tag.get(title, ).strip() if name_tag else 未知 price_tag item.find(span, class_price_n) price price_tag.text.strip() if price_tag else 0.00 author_tag item.find(div, class_publisher_info).find(a) author author_tag.text.strip() if author_tag else 未知 publisher_tag item.find(div, class_publisher_info).find_all(a)[-1] publisher publisher_tag.text.strip() if publisher_tag and len(item.find(div, class_publisher_info).find_all(a)) 1 else 未知 # 构造数据字典 book_info { 排名: rank, 书名: book_name, 价格: price, 作者: author, 出版社: publisher } book_list.append(book_info) logger.info(f解析完成提取{len(book_list)}条图书数据) except Exception as e: logger.error(f解析页面失败异常{e}, exc_infoTrue) return book_list def batch_save_to_csv(data: List[Dict], file_path: str, is_append: bool True): 批量保存数据到CSV :param data: 图书数据列表 :param file_path: CSV路径 :param is_append: 是否追加模式True追加False覆盖 if not data: logger.warning(无数据可批量存储) return # 定义表头 headers [排名, 书名, 价格, 作者, 出版社] # 选择打开模式 mode a if is_append else w try: with open(file_path, mode, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesheaders) # 覆盖模式下写入表头 if not is_append: writer.writeheader() # 批量写入数据 writer.writerows(data) logger.info(f批量存储完成共写入{len(data)}条数据到{file_path}) except Exception as e: logger.error(f批量存储失败异常{e}) def batch_crawl_pages(url_list: List[str]): 分页批量爬取主函数 :param url_list: 分页URL列表 # 初始化数据缓存用于批量存储 batch_data [] # 记录已爬取页码 crawled_count 0 logger.info(开始分页批量爬取流程) for idx, url in enumerate(url_list): # 1. 获取单页数据 html fetch_single_page(url) if not html: crawled_count 1 time.sleep(CRAWL_INTERVAL) continue # 2. 解析单页数据 book_data parse_single_page(html) if book_data: batch_data.extend(book_data) # 3. 批量存储达到批量阈值或最后一页 if (idx 1) % BATCH_SIZE 0 or (idx 1) len(url_list): batch_save_to_csv( databatch_data, file_pathCSV_PATH, is_append(idx 1) BATCH_SIZE # 第一批次覆盖后续追加 ) # 清空缓存 batch_data [] # 4. 控制爬取频率 crawled_count 1 time.sleep(CRAWL_INTERVAL) logger.info(f分页批量爬取完成共爬取{crawled_count}页目标{len(url_list)}页) # 程序入口 if __name__ __main__: try: # 1. 构建URL队列 url_list build_page_urls(PAGE_START, PAGE_END) # 2. 执行批量爬取 batch_crawl_pages(url_list) except KeyboardInterrupt: logger.info(用户手动终止批量爬取流程) except Exception as e: logger.critical(f批量爬取崩溃全局异常{e}, exc_infoTrue)2.3 核心逻辑解析1URL 队列构建build_page_urls 函数核心作用根据分页规律自动生成所有待爬取 URL无需手动整理关键逻辑通过循环遍历页码范围1-20拼接固定 URL 前缀和页码生成完整 URL 列表优势适配任意分页范围只需调整PAGE_START和PAGE_END即可扩展爬取范围。2批量请求与解析单页请求隔离fetch_single_page函数仅处理单页请求异常时记录日志但不终止批量流程解析容错parse_single_page函数对每个字段做if/else容错避免单字段异常导致整页解析失败批量缓存通过batch_data列表缓存多页数据达到BATCH_SIZE5 页时批量存储减少 IO 次数。3批量存储策略分批存储每爬取 5 页数据批量写入 CSV避免一次性存储大量数据导致内存占用过高模式切换第一批次采用覆盖模式写入表头后续批次采用追加模式保证 CSV 结构完整空数据处理无数据时仅记录警告不触发存储异常。2.4 运行结果1控制台输出示例plaintext2025-12-17 11:00:00,123 - INFO - URL队列构建完成共生成20个分页URL 2025-12-17 11:00:00,124 - INFO - 开始分页批量爬取流程 2025-12-17 11:00:01,234 - INFO - 成功获取页面http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-1 2025-12-17 11:00:01,345 - INFO - 解析完成提取15条图书数据 2025-12-17 11:00:03,456 - INFO - 成功获取页面http://bang.dangdang.com/books/bestsellers/01.00.00.00.00.00-recent7-0-0-1-2 2025-12-17 11:00:03,567 - INFO - 解析完成提取15条图书数据 ... 2025-12-17 11:00:50,123 - INFO - 批量存储完成共写入75条数据到dangdang_bestsellers.csv 2025-12-17 11:01:20,456 - INFO - 分页批量爬取完成共爬取20页目标20页2CSV 文件输出示例排名书名价格作者出版社12025 新版 考研英语词汇红宝书¥49.8考研英语命题研究组西北大学出版社2活着精装版¥28.5余华作家出版社3三体全集全三册¥89.0刘慈欣重庆出版社三、实战 2多 URL 列表批量爬取3.1 场景说明假设需爬取多个无规律的图书详情页 URL如从外部 CSV 文件导入URL 列表如下plaintexthttp://product.dangdang.com/29368765.html http://product.dangdang.com/29180128.html http://product.dangdang.com/28976543.html ...3.2 完整代码实现python运行import requests from bs4 import BeautifulSoup import csv import time import logging from fake_useragent import UserAgent from typing import List, Dict, Optional # 基础配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler(), logging.FileHandler(multi_url_crawl.log, encodingutf-8)] ) logger logging.getLogger(__name__) # 配置项 URL_LIST_PATH book_urls.csv # 存储URL列表的CSV文件 DETAIL_CSV_PATH book_details.csv HEADERS { User-Agent: UserAgent().random, Accept-Language: zh-CN,zh;q0.9, Connection: keep-alive } REQUEST_TIMEOUT 10 CRAWL_INTERVAL 2 DUPLICATE_CHECK True # 开启URL去重 CRAWLED_URLS set() # 存储已爬取URL用于去重 # 核心函数 def load_url_list(file_path: str) - List[str]: 从CSV文件加载URL列表去重 :param file_path: URL列表文件路径 :return: 去重后的URL列表 url_list [] try: with open(file_path, r, encodingutf-8) as f: reader csv.reader(f) # 跳过表头若有 next(reader, None) for row in reader: if row: url row[0].strip() url_list.append(url) # 去重处理 if DUPLICATE_CHECK: original_count len(url_list) url_list list(set(url_list)) logger.info(fURL去重完成原始{original_count}个去重后{len(url_list)}个) else: logger.info(f加载URL列表完成共{len(url_list)}个URL) return url_list except Exception as e: logger.error(f加载URL列表失败异常{e}) return [] def fetch_detail_page(url: str) - Optional[str]: 获取图书详情页HTML带去重检查 :param url: 详情页URL :return: HTML文本 # 去重检查 if url in CRAWLED_URLS: logger.warning(fURL已爬取跳过{url}) return None try: response requests.get( urlurl, headersHEADERS, timeoutREQUEST_TIMEOUT ) response.raise_for_status() response.encoding response.apparent_encoding # 标记为已爬取 CRAWLED_URLS.add(url) logger.info(f成功获取详情页{url}) return response.text except requests.exceptions.RequestException as e: logger.error(f获取详情页失败{url}异常{str(e)[:100]}) return None def parse_detail_page(html: str, url: str) - Dict: 解析图书详情页数据 :param html: HTML文本 :param url: 详情页URL用于关联数据 :return: 图书详情字典 detail_info { URL: url, 书名: 未知, 作者: 未知, 定价: 未知, ISBN: 未知, 出版时间: 未知 } if not html: return detail_info try: soup BeautifulSoup(html, html.parser) # 提取书名 title_tag soup.find(h1, class_name_info) if title_tag: detail_info[书名] title_tag.text.strip() # 提取作者 author_tag soup.find(a, class_writer) if author_tag: detail_info[作者] author_tag.text.strip() # 提取定价 price_tag soup.find(span, class_price) if price_tag: detail_info[定价] price_tag.text.strip() # 提取ISBN和出版时间从详情表中获取 info_table soup.find(table, class_mess_table) if info_table: rows info_table.find_all(tr) for row in rows: th row.find(th) td row.find(td) if th and td: th_text th.text.strip() td_text td.text.strip() if ISBN in th_text: detail_info[ISBN] td_text if 出版时间 in th_text: detail_info[出版时间] td_text logger.info(f解析详情页完成{detail_info[书名]}) except Exception as e: logger.error(f解析详情页失败{url}异常{e}, exc_infoTrue) return detail_info def batch_crawl_multi_urls(url_list: List[str]): 多URL列表批量爬取主函数 :param url_list: 待爬取URL列表 # 初始化批量数据缓存 batch_data [] BATCH_SIZE 10 # 每10个URL批量存储一次 total_count len(url_list) success_count 0 logger.info(f开始多URL批量爬取共{total_count}个目标URL) for idx, url in enumerate(url_list): # 1. 获取详情页数据 html fetch_detail_page(url) if not html: continue # 2. 解析详情页数据 detail_data parse_detail_page(html, url) batch_data.append(detail_data) success_count 1 # 3. 批量存储 if (idx 1) % BATCH_SIZE 0 or (idx 1) total_count: batch_save_to_csv(batch_data, DETAIL_CSV_PATH, is_append(idx 1) BATCH_SIZE) batch_data [] # 4. 控制频率 time.sleep(CRAWL_INTERVAL) logger.info(f多URL批量爬取完成成功爬取{success_count}个URL失败{total_count - success_count}个) def batch_save_to_csv(data: List[Dict], file_path: str, is_append: bool True): 复用批量存储函数同实战1 if not data: logger.warning(无详情数据可批量存储) return headers [URL, 书名, 作者, 定价, ISBN, 出版时间] mode a if is_append else w try: with open(file_path, mode, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesheaders) if not is_append: writer.writeheader() writer.writerows(data) logger.info(f详情数据批量存储完成共写入{len(data)}条) except Exception as e: logger.error(f详情数据存储失败异常{e}) # 程序入口 if __name__ __main__: try: # 1. 加载URL列表 url_list load_url_list(URL_LIST_PATH) if not url_list: logger.error(URL列表为空终止爬取) else: # 2. 执行多URL批量爬取 batch_crawl_multi_urls(url_list) except KeyboardInterrupt: logger.info(用户手动终止多URL批量爬取) except Exception as e: logger.critical(f多URL爬取崩溃全局异常{e}, exc_infoTrue)3.3 核心逻辑解析1URL 列表加载与去重外部加载从 CSV 文件读取 URL 列表适配大规模、无规律 URL 的批量爬取场景去重机制通过集合set对 URL 列表去重避免重复加载已爬取标记通过CRAWLED_URLS集合记录已爬取 URL即使程序中断重启也可避免重复爬取进阶可持久化到文件。2多 URL 爬取管控异常隔离单个 URL 爬取 / 解析失败仅记录日志不影响其他 URL 的爬取批量存储每 10 个 URL 解析完成后批量存储平衡内存占用和 IO 效率频率控制设置 2 秒请求间隔高于分页爬取详情页反爬更严格。3.4 运行结果1控制台输出示例plaintext2025-12-17 11:30:00,123 - INFO - URL去重完成原始30个去重后28个 2025-12-17 11:30:00,124 - INFO - 开始多URL批量爬取共28个目标URL 2025-12-17 11:30:02,345 - INFO - 成功获取详情页http://product.dangdang.com/29368765.html 2025-12-17 11:30:02,456 - INFO - 解析详情页完成活着精装版 2025-12-17 11:30:05,567 - WARNING - URL已爬取跳过http://product.dangdang.com/29368765.html ... 2025-12-17 11:31:40,123 - INFO - 详情数据批量存储完成共写入10条 2025-12-17 11:32:20,456 - INFO - 多URL批量爬取完成成功爬取26个URL失败2个2CSV 文件输出示例URL书名作者定价ISBN出版时间http://product.dangdang.com/29368765.html活着精装版余华¥35.097875063654322020-05http://product.dangdang.com/29180128.html三体全集刘慈欣¥98.097875366929302018-10四、批量爬取进阶优化4.1 断点续爬实现大规模批量爬取时程序中断后需从上次中断位置继续爬取核心思路是记录已爬取 URL 到文件python运行def save_crawled_urls(crawled_urls: set, file_path: str crawled_urls.txt): 保存已爬取URL到文件 try: with open(file_path, w, encodingutf-8) as f: f.write(\n.join(crawled_urls)) logger.info(f已爬取URL保存完成共{len(crawled_urls)}个) except Exception as e: logger.error(f保存已爬取URL失败异常{e}) def load_crawled_urls(file_path: str crawled_urls.txt) - set: 加载已爬取URL crawled_urls set() try: if os.path.exists(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: url line.strip() if url: crawled_urls.add(url) logger.info(f加载已爬取URL完成共{len(crawled_urls)}个) except Exception as e: logger.error(f加载已爬取URL失败异常{e}) return crawled_urls # 调用示例程序启动时加载已爬取URL if __name__ __main__: CRAWLED_URLS load_crawled_urls() # 爬取完成后保存 save_crawled_urls(CRAWLED_URLS)4.2 批量爬取性能优化对比优化方向未优化优化后性能提升存储方式单条写入 CSV批量写入 CSVIO 效率提升 50%频率控制固定 1 秒间隔动态间隔成功 1 秒失败 3 秒反爬规避率提升 30%数据解析单次解析单字段批量解析同类字段解析效率提升 20%去重机制列表遍历去重集合哈希去重去重效率提升 90%4.3 批量爬取反爬应对策略代理 IP 池批量爬取时切换代理 IP避免单 IP 被封禁可使用requests-proxies库Cookie 池轮换不同 Cookie模拟不同用户访问随机 User-Agent每次请求使用不同 User-Agent避免被识别为固定爬虫请求头丰富化添加Referer、Accept等字段模拟真实浏览器请求分布式爬取超大规模 URL 可采用多进程 / 多线程如multiprocessing库但需控制并发数。五、总结与扩展5.1 核心总结本文通过两种基础批量爬取模式的实战完整覆盖了批量爬取的核心流程分页批量爬取适用于 URL 有规律的场景核心是自动生成 URL 队列重点控制分页节奏多 URL 列表批量爬取适用于 URL 无规律的场景核心是 URL 加载与去重重点隔离异常 URL通用原则批量爬取需遵循 “频率控制、批量存储、异常隔离、去重防重” 四大原则保证稳定性。5.2 扩展方向多线程批量爬取使用threading库实现多线程并发爬取提升效率需控制并发数避免反爬异步批量爬取使用aiohttp库实现异步请求大幅提升批量爬取速度多站点批量爬取适配不同网站的解析规则实现跨站点批量数据采集批量数据清洗爬取完成后对批量数据进行去重、格式统一、缺失值填充等清洗操作可视化监控通过matplotlib绘制批量爬取进度、成功率等指标实时监控爬取状态。批量爬取是爬虫从 “入门” 到 “实战” 的关键跨越掌握批量爬取的核心逻辑后可适配绝大多数实际业务场景的数据采集需求。在实际开发中需根据目标网站的反爬强度、数据规模灵活调整批量策略平衡爬取效率与稳定性最终实现高效、合规的批量数据采集。