Streamlit+Selenium快速构建一个网络爬虫应用
项目需要从网上爬取数据，用了八爪鱼来进行测试，可以通过自定义任务，不需要编程即可实现对于数据的爬取，但是缺点是免费版本自定义任务有数量限制，另外在采集过程的控制上还不够便利，对于熟悉Python编程的人来说，可以选择用Selenium这个自动化测试的工具库来构建一个爬虫工具，然后通过Streamlit来快速搭建一个界面优美的Web应用。以下是如何用这两个工具构建自己的爬虫工具的介绍。
网页分析
首先要分析要爬取的网站的特点，因为我是要爬取各个省纪委网站发布的通报案例，因此需要对这些网站的页面进行分析。以山西省纪委的网站为例，http://www.sxdi.gov.cn/xxgk/jdpg/qb/wfzybxgdjswt/index.html，在页面中可以看到有列举每个案例的标题，发布时间。在页面底部有跳转不同页面的链接。点击某个案例的标题，就跳转到具体案例的内容页面。因此对于这个网站，首先我们要提取所有的案例的标题和发布时间，通过分析页面可知，页面的<ul class="yw-con">这个列表标签里面包含了很多的<li>列表项，每一项对应一个案例标题和时间，在<li>列表项里面的<a>标签是案例内容的链接。在页面底部的下一页的按钮，其对应的标签是<a class="layui-laypage-next">，分析了这些信息之后，我们就可以针对这些特征来提取相应的内容了。对于其他纪委网站的网页，其特征也是大同小异，都可以通过这些方法来进行分析提取特征。
下面我们可以定义一个Json格式的配置文件，把各个纪委网站要提取的内容的特征写到配置文件里面，如下是山西纪委和中纪委这两个网站的配置：
{
"中纪委-违反中央八项规定精神的通报": {
"start_page": "https://www.ccdi.gov.cn/jdjbnew/wfbxgd/index.html",
"next_page_click_class": "a.next",
"list_class": "ul.list_news_dl2",
"list_element_tag": "li",
"content_class": "div.content"
},
"山西纪委-违反中央八项规定精神的通报": {
"start_page": "http://www.sxdi.gov.cn/xxgk/jdpg/qb/wfzybxgdjswt/index.html",
"next_page_click_class": "a.layui-laypage-next",
"last_page_class": "a.layui-laypage-next.layui-disabled",
"list_class": "ul.yw-con",
"list_element_tag": "li",
"content_class": "div.nrs-con",
"same_page": true
}
}解释一下这个配置文件，其中start_page是这个网站的首页地址，next_page_click_class是下一页这个按钮的css类，last_page_class这个是当点击到最后一页时，下一页这个按钮的css类。不同的网站设计不一样，例如中纪委的网站当点击到最后一页时，下一页这个按钮消失，但是对于山西纪委的网站，当点击到最后一页时，下一页这个按钮会应用不同的css类，变成灰色。list_class是标题列表项的标签css类，list_element_tag是每一个列表项的标签，content_class是每个案例页面里面显示案件内容这个信息所对应的css类，same_page为true表示网站点击下一页的网页url不变。
Selenium安装与使用
Selenium是一个自动化Web浏览的工具，通常用于编写自动化测试脚本或者网络爬虫。其主要的核心组件是WebDriver，通过对应的驱动程序将Selenium的命令转换为浏览器的操作。考虑到很多网站都有反爬虫机制，Chrome的driver在应对这方面比较好，所以我采用的是ChromeDriver，注意下载Chromedriver的版本要和本地环境的Chrome浏览器的版本相对应。对于Ubuntu系统，需要把下载的ChromeDriver拷贝到/usr/local/bin目录下。
下面编写一个Python脚本来通过Selenium来爬取数据，代码如下：
from selenium.webdriver import Chrome
from selenium.webdriver import ChromeOptions
from selenium.webdriver.common.by import By
from selenium.common.exceptions import TimeoutException, NoSuchElementException, StaleElementReferenceException
import time
import tqdm
import pandas as pd
import argparse
from datetime import datetime
import json
parser = argparse.ArgumentParser(description='Process arguments.')
parser.add_argument('output', default="crawler_case.xlsx", help='输出结果的文件名，xlsx格式')
parser.add_argument('website', default="中纪委-违反中央八项规定精神的通报", help='要爬取的网站名')
parser.add_argument('--print_progress', type=bool, default=False, help='是否需要打印进度')
parser.add_argument('--records', type=int, default=0, help='抓取多少条数据，0表示全部')
parser.add_argument('--from_date', type=str, help='发布时间的开始日期, 格式为YYYY-MM-DD')
parser.add_argument('--to_date', type=str, help='发布时间的结束日期, 格式为YYYY-MM-DD')
args = parser.parse_args()
if args.from_date is not None:
fromdate = datetime.strptime(args.from_date, "%Y-%m-%d")
else:
fromdate = None
if args.to_date is not None:
todate = datetime.strptime(args.to_date, "%Y-%m-%d")
else:
todate = None
options = ChromeOptions()
options.add_argument("--headless")
options.add_argument("--no-sandbox")
options.add_argument("--disable-blink-features")
options.add_argument("--disable-blink-features=AutomationControlled")
driver = Chrome(options=options)
driver.set_page_load_timeout(600)
driver.implicitly_wait(10)
page_browser = Chrome(options=options)
page_browser.set_page_load_timeout(10)
with open('crawler.json', 'r') as f:
all_config = json.load(f)
config = all_config[args.website]
count = 0
results = []
flag = True
driver.get(config['start_page'])
while flag:
retryCount = 0
sleepTime = 1
while retryCount<5:
list_elements = driver.find_element(By.CSS_SELECTOR, config['list_class'])
time.sleep(sleepTime)
try:
if config['list_element_tag'] is not None:
elements = list_elements.find_elements(By.TAG_NAME, config['list_element_tag'])
else:
elements = list_elements.find_elements(By.CSS_SELECTOR, config['list_element_class'])
break
except StaleElementReferenceException:
retryCount += 1
sleepTime *= 2
for i in tqdm.trange(len(elements)):
item = elements[i]
split_strs = item.text.split('\n')
if len(split_strs) < 2:
continue
try:
title = split_strs[0]
publish_time = split_strs[1]
publish_date = datetime.strptime(publish_time, "%Y-%m-%d")
except ValueError:
title = split_strs[1]
publish_time = split_strs[0]
publish_date = datetime.strptime(publish_time, "%Y-%m-%d")
count += 1
if fromdate is not None:
if publish_date < fromdate:
flag = False
break
if todate is not None:
if publish_date > todate:
continue
link = item.find_element(By.TAG_NAME, 'a').get_attribute("href")
try:
page_browser.get(link)
article = page_browser.find_element(By.CSS_SELECTOR, config['content_class']).text
results.append([title, publish_time, article])
time.sleep(0.5)
except TimeoutException:
count -= 1
except NoSuchElementException:
count -= 1
if args.records > 0 and count >= args.records:
flag = False
break
if len(results) > 0:
df = pd.DataFrame(results, columns=['标题', '时间', '正文'])
df.to_excel(args.output, header=True, index=False)
if args.print_progress:
print("Dataframe saved: " + args.output)
if not flag:
break
try:
next_page = driver.find_element(By.CSS_SELECTOR, config['next_page_click_class'])
except NoSuchElementException:
next_page = None
try:
last_page = driver.find_element(By.CSS_SELECTOR, config['last_page_class'])
except NoSuchElementException:
last_page = None
except KeyError:
last_page = None
if last_page:
flag = False
elif next_page is None:
flag = False
else:
initial_url = driver.current_url
next_page.click()
time.sleep(1)
new_url = driver.current_url
if "same_page" not in config and initial_url == new_url:
flag = False
if args.print_progress:
print("Crawler finished.")在这个脚本里，可以通过传入参数来控制抓取多少条记录，以及设定发布时间的范围来抓取记录。
Streamlit编写Web应用
Streamlit 是一个开源的 Python 库，用于快速创建和分享交互式的 Web 应用程序，特别是在数据科学和机器学习领域。它允许开发者使用简单的 Python 代码来构建应用，而无需具备复杂的前端开发技能。
以下代码构建一个Web应用
from io import BytesIO
import streamlit as st
import pandas as pd
import numpy as np
import subprocess
import re
import datetime
from datetime import date
import json
st.write('### 纪委通报案例采集与整理')
tab1 = st.tabs(["案例爬取"])
#读取爬虫网站的配置
with open('crawler.json', 'r') as f:
config = json.load(f)
with tab1:
options = list(config.keys())
selected_option = st.selectbox(
"**请选择一个网站**", # 标签文本
options, # 选项列表
index=0 # 默认选中项的索引
)
records_number = st.number_input("**请输入要抓取的数据条数，0表示全部**", min_value=0)
today = date.today()
prev_year = today.year - 20
prev = datetime.date(prev_year, 1, 1)
d = st.date_input(
"**选择要提取的案例发布时间的范围**",
(prev, today),
prev,
today,
format="YYYY.MM.DD"
)
button1 = st.button('爬取通报案例')
placeholder1 = st.empty()
placeholder2 = st.empty()
if button1:
with subprocess.Popen(
['python', 'crawler.py', 'temp.xlsx', selected_option,
'--print_progress', 'True', '--records', str(records_number),
'--from_date', d[0].strftime("%Y-%m-%d"), '--to_date', d[1].strftime("%Y-%m-%d")],
stdout=subprocess.PIPE, text=True) as p:
placeholder1.markdown('处理中... 请等待!')
for line in p.stdout:
if not line.startswith('INFO') and not line.startswith('WARN'):
# 更新进度条的文本信息
if line.startswith('Dataframe saved: '):
df = pd.read_excel('temp.xlsx', header=0)
placeholder2.write(df)
if line.startswith('Crawler finished'):
placeholder1.markdown('处理完成!')
# 将DataFrame转换为Excel格式的字节流
output = BytesIO()
with pd.ExcelWriter(output, engine='xlsxwriter') as writer:
df.to_excel(writer, index=False, sheet_name='Sheet1')
output.seek(0)
st.download_button(
label="下载数据",
data = output.getvalue(),
file_name = "download.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
) 在以上代码中，通过Subprocess来调用之前编写的爬虫程序，通过获取程序打印的信息来了解执行的进度，并且定期把爬取的记录显示在页面上。
运行效果演示
通过Streamlit+Selenium实现的一个网络爬虫应用