赞
踩
pip install selenium webdriver_manager beautifulsoup4 pandas
本项目旨在通过Selenium模拟用户浏览器行为,获取特定网站(如Boss直聘)上的职位信息,并利用BeautifulSoup解析这些信息。为了实现数据的持续累积而不是每次运行都覆盖原有数据,采用pandas进行数据合并和去重,最终将更新后的数据保存到CSV文件中。
- from selenium import webdriver
- from selenium.webdriver.chrome.service import Service
- from webdriver_manager.chrome import ChromeDriverManager
-
- options = webdriver.ChromeOptions()
- try:
- driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
- except Exception as e:
- logging.error(f"创建WebDriver时出错: {e}")
- raise
- 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获取完毕并已保存")
- 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加载完毕")
- from bs4 import BeautifulSoup
- from selenium.webdriver.support.ui import WebDriverWait
- from selenium.webdriver.support import expected_conditions as EC
- from selenium.webdriver.common.by import By
-
- def 获取职位信息(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)
- import pandas as pd
-
- if 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')
- 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()
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。