Python实战：使用selenium及BeautifulSoup4进行BOOS直聘信息爬取与数据累积【附源码】

运行环境要求

• 操作系统：适用于Windows、macOS、Linux。
• Python版本：Python 3.6及以上。
• 依赖库： - selenium：用于模拟浏览器操作。- webdriver_manager：自动管理驱动程序。- BeautifulSoup4：解析HTML页面。- pandas：数据处理和CSV文件操作。- logging：日志记录。 pip install selenium webdriver_manager beautifulsoup4 pandas

设计思路

本项目旨在通过Selenium模拟用户浏览器行为，获取特定网站（如Boss直聘）上的职位信息，并利用BeautifulSoup解析这些信息。为了实现数据的持续累积而不是每次运行都覆盖原有数据，采用pandas进行数据合并和去重，最终将更新后的数据保存到CSV文件中。

具体实践

1. 初始化Selenium WebDriver：配置ChromeDriver，启动Chrome浏览器实例。 from selenium import webdriverfrom selenium.webdriver.chrome.service import Servicefrom webdriver_manager.chrome import ChromeDriverManageroptions = webdriver.ChromeOptions()try: driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)except Exception as e: logging.error(f"创建WebDriver时出错: {e}") raise
2. 获取并保存Cookies：访问目标网站并手动登录，然后保存Cookies以便后续自动登录使用。 def 获取cookie(driver, url): logging.info("开始获取cookie") driver.get(url) time.sleep(30) # 等待足够的时间手动登录并保存cookies cookies = driver.get_cookies() with open('cookies.json', 'w') as f: json.dump(cookies, f) logging.info("Cookie获取完毕并已保存")
3. 加载Cookies实现自动登录：在后续的会话中加载之前保存的Cookies，实现自动登录。 def 加载cookie(driver, cookie_file='cookies.json'): logging.info("开始加载cookie") with open(cookie_file, 'r') as f: cookies = json.load(f) for cookie in cookies: if 'expiry' in cookie: del cookie['expiry'] # 删除过期时间，避免格式错误 driver.add_cookie(cookie) logging.info("Cookie加载完毕")
4. 爬取职位信息：访问职位列表页面，使用BeautifulSoup解析页面，提取职位相关信息。 from bs4 import BeautifulSoupfrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.common.by import Bydef 获取职位信息(driver, base_url, pages=1): logging.info("开始获取职位信息") 职位信息_list = [] for i in range(1, pages + 1): url = f'{base_url}&page={i}' driver.get(url) WebDriverWait(driver, 10).until(EC.presence_of_all_elements_located((By.CLASS_NAME, "job-card-wrapper"))) soup = BeautifulSoup(driver.page_source, 'html.parser') jobs = soup.find_all('li', class_='job-card-wrapper') # 提取并添加职位信息到列表中... return pd.DataFrame(职位信息_list)
5. 数据去重与累积：读取已有的数据文件（如果存在），将新抓取的数据与旧数据合并，去除重复项，然后保存更新后的数据。 import pandas as pdif os.path.exists(data_file): existing_data = pd.read_csv(data_file)else: existing_data = pd.DataFrame()新职位信息 = 获取职位信息(driver, base_url, pages=5) 更新后的职位信息 = pd.concat([existing_data, 新职位信息], ignore_index=True).drop_duplicates(subset=['职位名称', '公司名称'])更新后的职位信息.to_csv(data_file, index=False, encoding='utf-8-sig')

技术要点

• Selenium的高级应用：包括但不限于Cookies的处理、显式等待（WebDriverWait）等技巧，以确保页面加载完成并成功获取数据。
• BeautifulSoup的灵活运用：精确地定位和提取所需的HTML元素。
• Pandas的数据处理能力：有效地合并、去重和保存数据。

项目复盘

• 挑战： - 页面结构变化导致的数据提取失败。- 网站反爬虫机制的应对。- 数据去重逻辑的设计。
• 解决方案： - 定期检查目标网站的页面结构，及时更新选择器。- 合理设置请求间隔，使用Cookies模拟登录状态，减少被封概率。- 利用pandas强大的数据处理功能，根据特定字段进行去重。

完整源码

import logging
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from bs4 import BeautifulSoup
import pandas as pd
import os
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
import time
import json

# 设置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def 获取cookie(driver, url):
    # 访问指定的URL并等待用户手动登录，然后保存cookies
    logging.info("开始获取cookie")
    driver.get(url)
    time.sleep(30)  # 等待足够的时间手动登录并保存cookies
    cookies = driver.get_cookies()
    with open('cookies.json', 'w') as f:
        json.dump(cookies, f)
    logging.info("Cookie获取完毕并已保存")

def 加载cookie(driver, cookie_file='cookies.json'):
    # 从文件中加载cookies并添加到driver
    logging.info("开始加载cookie")
    with open(cookie_file, 'r') as f:
        cookies = json.load(f)
        for cookie in cookies:
            if 'expiry' in cookie:
                del cookie['expiry']  # 删除过期时间，避免格式错误
            driver.add_cookie(cookie)
    logging.info("Cookie加载完毕")

def 安全获取文本(job, selector, default='未知'):
    """尝试从job对象中获取指定选择器的文本，如果失败则返回默认值"""
    try:
        return job.find(selector[0], class_=selector[1]).text.strip()
    except AttributeError:
        return default

def 获取职位信息(driver, base_url, pages=1):
    logging.info("开始获取职位信息")
    职位信息_list = []  
    for i in range(1, pages + 1):
        url = f'{base_url}&page={i}'
        
        加载成功 = False
        for 尝试次数 in range(3):  
            try:
                driver.get(url)
                WebDriverWait(driver, 10).until(EC.presence_of_all_elements_located((By.CLASS_NAME, "job-card-wrapper")))
                加载成功 = True
                break  
            except Exception as e:
                logging.error(f"尝试 {尝试次数+1} 次访问 {url} 失败: {e}")
                time.sleep(5)  
        
        if not 加载成功:
            logging.error(f"无法加载页面: {url}")
            continue  
        
        soup = BeautifulSoup(driver.page_source, 'html.parser')
        jobs = soup.find_all('li', class_='job-card-wrapper')
        for job in jobs:
            # 获取每个职位的信息，并添加到列表中
            职位名称 = 安全获取文本(job, ('span', 'job-name'))
            工作地点 = 安全获取文本(job, ('span', 'job-area'))
            薪资 = 安全获取文本(job, ('span', 'salary'), '面议')
            标签列表 = [li.text for li in job.find('ul', class_='tag-list').find_all('li')]
            经验要求 = 标签列表[0] if len(标签列表) > 0 else '未知'
            教育要求 = 标签列表[1] if len(标签列表) > 1 else '未知'
            联系人 = 安全获取文本(job, ('div', 'info-public'))
            公司名称 = 安全获取文本(job, ('h3', 'company-name'))
            公司标签列表 = [li.text for li in job.find('ul', class_='company-tag-list').find_all('li')]
            公司类型 = 公司标签列表[0] if len(公司标签列表) > 0 else '未知'
            公司规模 = 公司标签列表[1] if len(公司标签列表) > 1 else '未知'
            详情 = ','.join([li.text for li in job.find('div', class_='job-card-footer').find('ul', class_='tag-list').find_all('li')])
            职位详情链接 = "https://www.zhipin.com" + job.find('a')['href']
             # 直接将字典添加到列表中
            职位信息_list.append({
                '职位名称': 职位名称, 
                '工作地点': 工作地点, 
                '薪资': 薪资, 
                '经验要求': 经验要求, 
                '教育要求': 教育要求, 
                '联系人': 联系人, 
                '公司名称': 公司名称, 
                '公司类型': 公司类型, 
                '公司规模': 公司规模, 
                '详情': 详情,
                '职位详情链接': 职位详情链接
            })
              
    # 循环结束后，使用收集到的职位信息列表创建DataFrame
    职位信息 = pd.DataFrame(职位信息_list)
    logging.info(f"获取到 {len(职位信息)} 个职位信息")
    return 职位信息    

if __name__ == '__main__':
    base_url = 'https://www.zhipin.com/web/geek/job?query=&city=101240100'
    data_file = '职位信息.csv'  # 数据文件路径
    
    # 初始化webdriver，并设置Chrome Driver
    options = webdriver.ChromeOptions()
    try:
        driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
        driver.set_window_size(1920, 1080)  # 调整为合适的大小
    except Exception as e:
        logging.error(f"创建WebDriver时出错: {e}")
        raise

    # 第一次运行时获取cookie，之后可以注释掉这行代码
    # 获取cookie(driver, 'https://www.zhipin.com')
    
     # 加载已有数据（如果文件存在）
    if os.path.exists(data_file):
        existing_data = pd.read_csv(data_file)
    else:
        existing_data = pd.DataFrame()

    try:
        # 加载cookie，必须先访问网站才能设置cookie
        driver.get("https://www.zhipin.com")  
        加载cookie(driver)
        
        # 获取职位信息，可以根据需要调整页数
        新职位信息 = 获取职位信息(driver, base_url, pages=10)  
        
        # 合并新旧数据，并去除重复项
        更新后的职位信息 = pd.concat([existing_data, 新职位信息], ignore_index=True).drop_duplicates(subset=['职位名称', '公司名称'])
        
    except Exception as e:
        logging.error(f"获取职位信息时出错: {e}")
        driver.quit()
        raise

    # 输出或保存职位信息到CSV文件
    print(更新后的职位信息)
    更新后的职位信息.to_csv(data_file, index=False, encoding='utf-8-sig')

    # 关闭driver
    driver.quit()