Python 动态渲染页面爬取
动态渲染页面爬取最常见的就是面向接口的爬虫，在采集过程中需要首先分析出其接口地址，本文将带来 2 个基于接口的爬虫。
怎么用
目标站点【一派话题广场】分析
本篇博客的第一个采集目标站点是：https://Base64加密站点/matrix/pods，少数派网站的一个子级栏目。
目标站点采用 base64加密 c3NwYWkuY29t
目标数据所在界面如下图所示：
通过开发者工具，不断下拉加载页面，得到的接口请求规则如下：
其中参数除 offset 变化外，其余无变化，其中 limit 参数应该为每个数据量，基于此逻辑，请求接口可以通过代码进行批量生成，实测过程发现数据量也不大，只有 6 页。
下述代码采用了后进先出队列 LifoQueue，没有特殊原因，单纯给大家展示一下用法。
# 初始化一个队列 q = LifoQueue(maxsize=0) # 批量生成请求地址链接 for page in range(1, 7): # https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=0&created_at=0 q.put('https://Base64加密站点/api/v1/bullet/search/page/get?type=0&limit=10&offset={}&created_at=0'.format((page-1)*10))请求地址批量生成完毕之后，可以开始获取接口返回的数据了，具体代码如下，核心关注 main 部分内容。
# 请求头函数，可以参考之前的系列文章 def get_headers(): uas = [ "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)", "Mozilla/5.0 (compatible; Baiduspider-render/2.0; +http://www.baidu.com/search/spider.html)" ] ua = random.choice(uas) headers = { "user-agent": ua } return headers # 储存数据 def save(text): # 文件名使用时间戳命名 with open(f'{time.time()}.json', 'a+', encoding='utf-8') as f: f.write(text) print(text, "--- 保存成功") if __name__ == "__main__": # 判断队列是否为空，空则停止循环 while q.qsize() > 0: # 获取一个 URL url = q.get() # 通知任务已经完成 q.task_done() # 获取相应数据 res = requests.get(url=url, headers=get_headers(), timeout=10) # 调用保存函数 save(res.text) q.join() print("所有任务都已完成")因为上述案例实在太简单，连多线程都不用，所以我基于【话题广场】关键字，查询是否还有其它可用于 学习目的 相关站点，结果还真被我发现一个。
话题广场 - 集思录
集思录，一个以数据为本的投资社区，https://Python脱敏处理/topic/。
目标站点使用 base64 编码：d3d3Lmppc2lsdS5jbg==
每一个话题下面，都有很多问题，完美符合生产者消费者模型。
编码逻辑分析
本案例中生产函数用于产生列表页，消费函数用于抓取内页 标题，作者 ，链接。
由于在列表页之前，还存在一个层级-【热门话题】，所以需要提前准备好待抓取队列。
# 热门话题列表页待抓取链接 hot_subjects = Queue(maxsize=0) for i in range(1, 11): url = f'https://Python脱敏处理/topic/square/id-hot__feature_id-__page-{i}' hot_subjects.put(url)接下来生产函数用于产生列表页数据，到提前初始化好的 q_data_ids 队列中，其中 get_headers 函数，参考前文即可。
def get_headers(): uas = [ "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)", "Mozilla/5.0 (compatible; Baiduspider-render/2.0; +http://www.baidu.com/search/spider.html)" ] ua = random.choice(uas) headers = { "user-agent": ua } return headers # 初始化一个队列 q_data_ids = Queue(maxsize=0) # 生产函数，用于产生话题列表页 URL def producer(): while hot_subjects.qsize() > 0: # 取得一个分类页请求地址 list_url = hot_subjects.get() hot_subjects.task_done() print("正在解析：", list_url) # 获取话题列表页相应数据 res = requests.get(list_url, headers=get_headers(), timeout=3) # 解析，获取话题详情页进入的关键参数 `data-id`，代码后有对于该关键字的说明 element = etree.HTML(res.text) data_ids = element.xpath('//a[@class="aw-topic-name"]/@data-id') for data_id in data_ids: q_data_ids.put(data_id)上述代码最重要的是捕获 data-id，即热门话题的 ID 队列，该 ID 值会用在消费者函数中，用于请求详情页数据。
接下来就是消费者函数的实现，该逻辑主要用于抓取下图所示数据。
以上数据在测试过程中发现如下接口格式，其中 data_id 在生产者中产生，start_page 由循环迭代生成。
https://Python脱敏处理/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}上述地址即为最后目标数据所在地址，我们需要拼凑出标准地址，然后再对 start_page 页码，进行迭代，遍历获取全部数据。
学习时请注意下述代码注释，技术层级难度不大，重点为实现逻辑。
# 消费者函数 def consumer(): # 死循环读取 data_id 值，用于拼凑话题详情页数据 while True: # 取一个分类ID data_id = q_data_ids.get() q_data_ids.task_done() if data_id is None: break # start_page 初始值设置为 1，即从第一个开始读取数据 start_page = 1 # URL 拼接 url = f'https://Python脱敏处理/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}' res = requests.get(url=url, headers=get_headers(), timeout=5) text = res.text # 通过判断 text 是否为空，确定是否继续解析数据 while len(text) > 0: url = f'https://Python脱敏处理/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}' res = requests.get(url=url, headers=get_headers(), timeout=5) # print(res.url) text = res.text start_page += 1 # 如果 text 不为空，则解析数据，并存储数据 if len(text)>0: element = etree.HTML(res.text) titles = element.xpath('//h4/a/text()') urls = element.xpath('//h4/a/@href') names = element.xpath('//a[@class="aw-user-name"]/text()') data = zip(titles,names,urls) save_list = [f"{item[0]},{item[1]},{item[2]}\n" for item in data] long_str = "".join(save_list) with open("./data.csv","a+",encoding="utf-8") as f: f.write(long_str)该程序实现的步骤与逻辑相对会绕一些，故通过下图你可以再复盘进行理解，按照步骤 1,2,3 进行学习。
函数运行可是采用多线程实现，具体如下：
# 开启2个生产者线程 for p_in in range(1, 3): p = threading.Thread(target=producer) p.start() # 开启2个消费者线程 for p_in in range(1, 2): p = threading.Thread(target=consumer) p.start()