1. 政府采购网站数据采集的核心挑战政府采购网站的数据采集和普通电商网站完全不同最大的区别在于政府采购网站的反爬机制往往更加严格。我做过几十个政府采购网站的爬虫项目发现它们通常会有以下几个特点高频访问限制政府采购网站对短时间内的大量请求非常敏感一旦检测到异常流量会立即封禁IP。比如search.ccgp.gov.cn这个网站如果连续请求超过5次/分钟大概率会触发封禁。动态参数验证很多政府采购网站会在请求中加入动态生成的token或者时间戳参数如果请求中缺少这些参数即使IP没有被封也会返回错误数据。复杂页面结构政府采购网站的HTML结构往往比较复杂一个采购公告可能分散在多个div层中而且class命名没有规律可循。验证码机制当系统检测到可疑行为时会弹出图形验证码或者滑动验证码这对自动化采集是个不小的挑战。针对这些特点我们需要设计一个健壮的采集系统。下面这个基础框架是我在实际项目中验证过的import requests from bs4 import BeautifulSoup import time import random class GovPurchaseSpider: def __init__(self): self.base_url https://search.ccgp.gov.cn self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } def get_page(self, url, paramsNone, retry3): for i in range(retry): try: resp requests.get(url, headersself.headers, paramsparams, timeout10) if resp.status_code 200: return resp.text else: time.sleep(random.uniform(1, 3)) except Exception as e: print(f请求失败: {e}) time.sleep(5) return None2. Requests库的高级使用技巧很多新手在使用Requests库时只会用最简单的get方法这在实际项目中是远远不够的。下面分享几个我在政府采购网站采集中的实战经验会话保持非常重要。政府采购网站很多都需要登录后才能查看完整信息使用Session对象可以自动处理cookiessession requests.Session() login_url https://example.com/login login_data {username: yourname, password: yourpass} session.post(login_url, datalogin_data) # 登录 # 后续请求会自动携带cookies profile_page session.get(https://example.com/profile)请求头伪装是另一个关键点。政府采购网站会检测User-Agent、Referer等头信息。我通常会准备一组常见的浏览器头信息随机使用user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ] headers { User-Agent: random.choice(user_agents), Referer: https://www.ccgp.gov.cn/, Accept-Language: zh-CN,zh;q0.9 }异常处理也需要特别注意。政府采购网站经常会出现连接超时、响应过慢等问题完善的异常处理可以大大提高采集稳定性try: response requests.get(url, headersheaders, timeout15) response.raise_for_status() # 检查HTTP状态码 if 验证码 in response.text: raise Exception(触发验证码) except requests.exceptions.Timeout: print(请求超时正在重试...) except requests.exceptions.RequestException as e: print(f请求异常: {e})3. BeautifulSoup解析复杂页面结构政府采购网站的页面结构往往非常复杂用对解析方法可以事半功倍。我总结了几种常见情况的处理方案多层嵌套结构在采购公告中很常见。比如下面这个典型结构div classcontent div classvT_detail div classvT_detail_main ul li采购项目编号XXXXX/li li采购项目名称XXXXX/li /ul /div /div /div对应的解析代码应该是soup BeautifulSoup(html, lxml) main_div soup.find(div, class_vT_detail_main) if main_div: items main_div.find_all(li) for item in items: text item.get_text(stripTrue) if 采购项目编号 in text: project_id text.split()[-1]表格数据是另一个常见形式。政府采购网站喜欢用表格展示关键信息table soup.find(table, {id: resultTable}) if table: rows table.find_all(tr)[1:] # 跳过表头 for row in rows: cols row.find_all(td) if len(cols) 3: project_name cols[0].get_text(stripTrue) budget cols[1].get_text(stripTrue) deadline cols[2].get_text(stripTrue)动态加载内容越来越普遍。有些政府采购网站会使用AJAX加载数据这时候需要先分析XHR请求import json # 先找到包含动态数据的script标签 script soup.find(script, textre.compile(window.__DATA__)) if script: # 提取JSON数据 json_str script.text.split()[1].strip().rstrip(;) data json.loads(json_str) items data[list]4. 代理IP系统的工程化实现政府采购网站对IP的限制非常严格一个稳定的代理IP系统是项目成功的关键。我设计过不少代理IP方案下面分享一个经过验证的实现代理IP池的基本架构应该包含以下几个模块IP获取模块 - 从代理服务商API获取IP验证模块 - 检查IP是否可用存储模块 - 使用Redis存储可用IP调度模块 - 分配IP给各个爬虫任务下面是一个简化版的实现import redis import requests class ProxyPool: def __init__(self): self.redis redis.StrictRedis(hostlocalhost, port6379, db0) self.test_url http://httpbin.org/ip def add_proxy(self, proxy): 添加新代理到池子 if self._check_proxy(proxy): self.redis.sadd(valid_proxies, proxy) return True return False def _check_proxy(self, proxy): 验证代理是否可用 proxies { http: fhttp://{proxy}, https: fhttp://{proxy} } try: resp requests.get(self.test_url, proxiesproxies, timeout10) if resp.status_code 200: return True except: return False def get_random_proxy(self): 随机获取一个可用代理 proxy self.redis.srandmember(valid_proxies) return proxy.decode() if proxy else None def remove_proxy(self, proxy): 移除失效代理 self.redis.srem(valid_proxies, proxy)代理使用策略也很重要。在政府采购网站采集时我建议采用以下策略按请求切换IP每个请求使用不同的IP降低被封风险失败重试机制当请求失败时自动更换IP重试IP冷却机制使用过的IP暂时搁置过段时间再使用实现代码示例def make_request(url, proxy_pool, max_retry3): for i in range(max_retry): proxy proxy_pool.get_random_proxy() if not proxy: raise Exception(没有可用代理) proxies { http: fhttp://{proxy}, https: fhttp://{proxy} } try: resp requests.get(url, proxiesproxies, timeout15) if resp.status_code 200: return resp.text else: proxy_pool.remove_proxy(proxy) except Exception as e: print(f使用代理 {proxy} 请求失败: {e}) proxy_pool.remove_proxy(proxy) raise Exception(超过最大重试次数)5. 工程化部署与优化建议当采集系统需要长期运行时工程化部署就变得非常重要。下面分享几个实战中的经验分布式架构可以显著提升采集效率。我常用的方案是使用Redis作为任务队列多个爬虫节点从队列获取任务结果统一存储到MySQL或MongoDB基础实现代码import redis from multiprocessing import Process class CrawlerWorker(Process): def __init__(self, proxy_pool): super().__init__() self.redis redis.StrictRedis() self.proxy_pool proxy_pool def run(self): while True: task self.redis.lpop(task_queue) if not task: time.sleep(5) continue url task.decode() html make_request(url, self.proxy_pool) data parse_page(html) save_to_db(data) def start_workers(num4): proxy_pool ProxyPool() workers [CrawlerWorker(proxy_pool) for _ in range(num)] for w in workers: w.start() for w in workers: w.join()日志系统是另一个关键组件。完善的日志可以帮助快速定位问题import logging from logging.handlers import RotatingFileHandler def setup_logger(name): logger logging.getLogger(name) logger.setLevel(logging.INFO) # 文件日志自动轮转 file_handler RotatingFileHandler( crawler.log, maxBytes10*1024*1024, backupCount5 ) file_formatter logging.Formatter( %(asctime)s - %(levelname)s - %(message)s ) file_handler.setFormatter(file_formatter) # 控制台日志 console_handler logging.StreamHandler() console_formatter logging.Formatter(%(levelname)s: %(message)s) console_handler.setFormatter(console_formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger性能优化方面我有几个实用建议连接复用使用requests.Session可以复用TCP连接减少握手开销异步IO对于IO密集型任务可以使用aiohttp替代requests缓存机制对已经采集过的页面进行缓存避免重复采集智能休眠根据网站响应时间动态调整请求间隔异步采集示例import aiohttp import asyncio async def async_fetch(url, session, proxy): try: async with session.get(url, proxyproxy, timeout15) as response: if response.status 200: return await response.text() except Exception as e: print(f异步请求失败: {e}) return None async def async_crawler(urls): connector aiohttp.TCPConnector(limit10) # 限制并发数 async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for url in urls: proxy proxy_pool.get_random_proxy() task async_fetch(url, session, proxy) tasks.append(task) return await asyncio.gather(*tasks)
政府采购网站数据采集实战：Requests+BeautifulSoup与代理IP的工程化应用
发布时间：2026/7/15 2:20:08
1. 政府采购网站数据采集的核心挑战政府采购网站的数据采集和普通电商网站完全不同最大的区别在于政府采购网站的反爬机制往往更加严格。我做过几十个政府采购网站的爬虫项目发现它们通常会有以下几个特点高频访问限制政府采购网站对短时间内的大量请求非常敏感一旦检测到异常流量会立即封禁IP。比如search.ccgp.gov.cn这个网站如果连续请求超过5次/分钟大概率会触发封禁。动态参数验证很多政府采购网站会在请求中加入动态生成的token或者时间戳参数如果请求中缺少这些参数即使IP没有被封也会返回错误数据。复杂页面结构政府采购网站的HTML结构往往比较复杂一个采购公告可能分散在多个div层中而且class命名没有规律可循。验证码机制当系统检测到可疑行为时会弹出图形验证码或者滑动验证码这对自动化采集是个不小的挑战。针对这些特点我们需要设计一个健壮的采集系统。下面这个基础框架是我在实际项目中验证过的import requests from bs4 import BeautifulSoup import time import random class GovPurchaseSpider: def __init__(self): self.base_url https://search.ccgp.gov.cn self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } def get_page(self, url, paramsNone, retry3): for i in range(retry): try: resp requests.get(url, headersself.headers, paramsparams, timeout10) if resp.status_code 200: return resp.text else: time.sleep(random.uniform(1, 3)) except Exception as e: print(f请求失败: {e}) time.sleep(5) return None2. Requests库的高级使用技巧很多新手在使用Requests库时只会用最简单的get方法这在实际项目中是远远不够的。下面分享几个我在政府采购网站采集中的实战经验会话保持非常重要。政府采购网站很多都需要登录后才能查看完整信息使用Session对象可以自动处理cookiessession requests.Session() login_url https://example.com/login login_data {username: yourname, password: yourpass} session.post(login_url, datalogin_data) # 登录 # 后续请求会自动携带cookies profile_page session.get(https://example.com/profile)请求头伪装是另一个关键点。政府采购网站会检测User-Agent、Referer等头信息。我通常会准备一组常见的浏览器头信息随机使用user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ] headers { User-Agent: random.choice(user_agents), Referer: https://www.ccgp.gov.cn/, Accept-Language: zh-CN,zh;q0.9 }异常处理也需要特别注意。政府采购网站经常会出现连接超时、响应过慢等问题完善的异常处理可以大大提高采集稳定性try: response requests.get(url, headersheaders, timeout15) response.raise_for_status() # 检查HTTP状态码 if 验证码 in response.text: raise Exception(触发验证码) except requests.exceptions.Timeout: print(请求超时正在重试...) except requests.exceptions.RequestException as e: print(f请求异常: {e})3. BeautifulSoup解析复杂页面结构政府采购网站的页面结构往往非常复杂用对解析方法可以事半功倍。我总结了几种常见情况的处理方案多层嵌套结构在采购公告中很常见。比如下面这个典型结构div classcontent div classvT_detail div classvT_detail_main ul li采购项目编号XXXXX/li li采购项目名称XXXXX/li /ul /div /div /div对应的解析代码应该是soup BeautifulSoup(html, lxml) main_div soup.find(div, class_vT_detail_main) if main_div: items main_div.find_all(li) for item in items: text item.get_text(stripTrue) if 采购项目编号 in text: project_id text.split()[-1]表格数据是另一个常见形式。政府采购网站喜欢用表格展示关键信息table soup.find(table, {id: resultTable}) if table: rows table.find_all(tr)[1:] # 跳过表头 for row in rows: cols row.find_all(td) if len(cols) 3: project_name cols[0].get_text(stripTrue) budget cols[1].get_text(stripTrue) deadline cols[2].get_text(stripTrue)动态加载内容越来越普遍。有些政府采购网站会使用AJAX加载数据这时候需要先分析XHR请求import json # 先找到包含动态数据的script标签 script soup.find(script, textre.compile(window.__DATA__)) if script: # 提取JSON数据 json_str script.text.split()[1].strip().rstrip(;) data json.loads(json_str) items data[list]4. 代理IP系统的工程化实现政府采购网站对IP的限制非常严格一个稳定的代理IP系统是项目成功的关键。我设计过不少代理IP方案下面分享一个经过验证的实现代理IP池的基本架构应该包含以下几个模块IP获取模块 - 从代理服务商API获取IP验证模块 - 检查IP是否可用存储模块 - 使用Redis存储可用IP调度模块 - 分配IP给各个爬虫任务下面是一个简化版的实现import redis import requests class ProxyPool: def __init__(self): self.redis redis.StrictRedis(hostlocalhost, port6379, db0) self.test_url http://httpbin.org/ip def add_proxy(self, proxy): 添加新代理到池子 if self._check_proxy(proxy): self.redis.sadd(valid_proxies, proxy) return True return False def _check_proxy(self, proxy): 验证代理是否可用 proxies { http: fhttp://{proxy}, https: fhttp://{proxy} } try: resp requests.get(self.test_url, proxiesproxies, timeout10) if resp.status_code 200: return True except: return False def get_random_proxy(self): 随机获取一个可用代理 proxy self.redis.srandmember(valid_proxies) return proxy.decode() if proxy else None def remove_proxy(self, proxy): 移除失效代理 self.redis.srem(valid_proxies, proxy)代理使用策略也很重要。在政府采购网站采集时我建议采用以下策略按请求切换IP每个请求使用不同的IP降低被封风险失败重试机制当请求失败时自动更换IP重试IP冷却机制使用过的IP暂时搁置过段时间再使用实现代码示例def make_request(url, proxy_pool, max_retry3): for i in range(max_retry): proxy proxy_pool.get_random_proxy() if not proxy: raise Exception(没有可用代理) proxies { http: fhttp://{proxy}, https: fhttp://{proxy} } try: resp requests.get(url, proxiesproxies, timeout15) if resp.status_code 200: return resp.text else: proxy_pool.remove_proxy(proxy) except Exception as e: print(f使用代理 {proxy} 请求失败: {e}) proxy_pool.remove_proxy(proxy) raise Exception(超过最大重试次数)5. 工程化部署与优化建议当采集系统需要长期运行时工程化部署就变得非常重要。下面分享几个实战中的经验分布式架构可以显著提升采集效率。我常用的方案是使用Redis作为任务队列多个爬虫节点从队列获取任务结果统一存储到MySQL或MongoDB基础实现代码import redis from multiprocessing import Process class CrawlerWorker(Process): def __init__(self, proxy_pool): super().__init__() self.redis redis.StrictRedis() self.proxy_pool proxy_pool def run(self): while True: task self.redis.lpop(task_queue) if not task: time.sleep(5) continue url task.decode() html make_request(url, self.proxy_pool) data parse_page(html) save_to_db(data) def start_workers(num4): proxy_pool ProxyPool() workers [CrawlerWorker(proxy_pool) for _ in range(num)] for w in workers: w.start() for w in workers: w.join()日志系统是另一个关键组件。完善的日志可以帮助快速定位问题import logging from logging.handlers import RotatingFileHandler def setup_logger(name): logger logging.getLogger(name) logger.setLevel(logging.INFO) # 文件日志自动轮转 file_handler RotatingFileHandler( crawler.log, maxBytes10*1024*1024, backupCount5 ) file_formatter logging.Formatter( %(asctime)s - %(levelname)s - %(message)s ) file_handler.setFormatter(file_formatter) # 控制台日志 console_handler logging.StreamHandler() console_formatter logging.Formatter(%(levelname)s: %(message)s) console_handler.setFormatter(console_formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger性能优化方面我有几个实用建议连接复用使用requests.Session可以复用TCP连接减少握手开销异步IO对于IO密集型任务可以使用aiohttp替代requests缓存机制对已经采集过的页面进行缓存避免重复采集智能休眠根据网站响应时间动态调整请求间隔异步采集示例import aiohttp import asyncio async def async_fetch(url, session, proxy): try: async with session.get(url, proxyproxy, timeout15) as response: if response.status 200: return await response.text() except Exception as e: print(f异步请求失败: {e}) return None async def async_crawler(urls): connector aiohttp.TCPConnector(limit10) # 限制并发数 async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for url in urls: proxy proxy_pool.get_random_proxy() task async_fetch(url, session, proxy) tasks.append(task) return await asyncio.gather(*tasks)
