文章目录
引言
在当今就业比较困难，很多人对于要投递的岗位相关行业信息不了解，如果有招聘网站职位信息的可视化分析，那就可以直观地了解这个岗位信息，而数据分析的前提就是数据获取。本文将介绍如何使用Python语言和Selenium库来爬取招聘网站上的职位信息，并将其存储为CSV文件。我们将以“前端工程师”这一职位为例，展示如何实现这一过程。
效果示例：
环境准备
在开始之前，确保你的开发环境中安装了以下组件：
- Python 3.x
- selenium~=4.24.0
- Chrome WebDriver
- pandas~=2.2.3
- csv库
我这里使用的浏览器是Google Chrome浏览器
读者请根据自己的浏览器情况安装符合自己的浏览器驱动。
网页分析
网站URL：https://www.zhipin.com/web/geek/job?
参数：
query=前端工程师：搜索的内容
city=101280600：城市代码：深圳
下面的列表信息即是要爬取的信息。
先鼠标右键“检查”进入开发者模式，再找到相应的HTML信息，之后就可以分析更多信息的元素位置了。
代码解析
1. 导入必要的库
import csv
import json
import os
import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
这段代码导入了爬虫程序所需的所有外部库。
2. 定义爬虫类
class spider(object):
def __init__(self, type, page):
self.type = type
self.page = page
self.spiderUrl = "https://www.zhipin.com/web/geek/job?query=%s&city=101280600&page=%s"
spider类是爬虫的核心，它接受职位类型和起始页面作为参数，并初始化爬取的URL模板。
ps:page=是页数
3. 启动浏览器
def startBrowser(self):
service = Service('./chromedriver.exe')
options = webdriver.ChromeOptions()
options.add_experimental_option('excludeSwitches', ['enable-automation'])
browser = webdriver.Chrome(service=service, options=options)
return browser
startBrowser方法用于启动Chrome浏览器，这是Selenium进行网页操作的基础。
4. 主要爬取逻辑
def main(self, page):
# 省略部分代码...
job_List = browser.find_elements(by=By.XPATH, value='//div[@class="search-job-result"]/ul[@class="job-list-box"]/li')
for index, job in enumerate(job_List):
try:
# 提取职位信息的代码...
except Exception as e:
print(e)
pass
self.page += 1
browser.quit()
self.main(self.page)
main方法是爬取逻辑的核心，它循环访问每一页的职位列表，提取每个职位的详细信息，并递归地爬取下一页。（所有的代码放在文章末尾）
5. 提取职位信息
在main方法中，我们使用XPATH来定位页面元素，并提取职位的标题、地址、薪资等信息。
6. 保存数据到CSV
def save_to_csv(self, rowData):
with open('./temp.csv', 'a', newline='', encoding='utf-8') as wf:
writer = csv.writer(wf)
writer.writerow(rowData)
save_to_csv方法将提取的职位信息以行的形式追加到CSV文件中。
7. 初始化CSV文件
def init(self):
if not os.path.exists('./temp.csv'):
with open('./temp.csv', 'w', encoding='utf-8') as wf:
writer = csv.writer(wf)
writer.writerow(["title", "address", "type", ...])
init方法用于在程序开始时创建CSV文件，并定义列名。
8. 清理和整理CSV数据
def clear_csv(self):
df = pd.read_csv('./temp.csv')
df.dropna(inplace=True)
df.drop_duplicates(inplace=True)
df['salaryMonth'] = df['salaryMonth'].map(lambda x: x.replace('薪', ''))
return df.values
clear_csv方法用于清理CSV文件中的数据，去除空值和重复项，并统一薪资字段的格式。
9. 全部代码
import csv
import json
import os
import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
class spider(object):
def __init__(self, type, page):
self.type = type
self.page = page
self.spiderUrl = "https://www.zhipin.com/web/geek/job?query=%s&city=101280600&page=%s"
def startBrowser(self):
service = Service('./chromedriver.exe')
options = webdriver.ChromeOptions()
options.add_experimental_option('excludeSwitches', ['enable-automation'])
# options.add_experimental_option("debuggerAddress", "127.0.0.1:9222")
browser = webdriver.Chrome(service=service, options=options)
return browser
def main(self, page):
if self.page > 2:
return True
browser = self.startBrowser()
print("正在爬取页面路径" + self.spiderUrl % (self.type, self.page))
browser.get(self.spiderUrl % (self.type, self.page))
time.sleep(30)
browser.execute_script("""var totalHeight = 0;
var distance = 100;
var interval = setInterval(function(){
var scrollHeight = document.body.scrollHeight;
window.scrollBy(0, distance);
totalHeight += distance;
if(totalHeight >= scrollHeight){
clearInterval(interval);
}
}, 100);
""")
time.sleep(10)
# print(browser.page_source)
job_List = browser.find_elements(by=By.XPATH,
value='//div[@class="search-job-result"]/ul[@class="job-list-box"]/li')
print(len(job_List))
for index, job in enumerate(job_List):
try:
jobData = []
print('正在爬取第%s条职位信息' % (index + 1))
# print(job.text)
# title = job.find_element(by=By.XPATH, value='').text
# title
title = job.find_element(by=By.XPATH,
value=".//div[@class='job-title clearfix']/span[@class='job-name']").text
print(title)
# addresses
addresses = job.find_element(by=By.XPATH,
value='.//div[@class="job-card-body clearfix"]/a/div/span[2]').text.split(
'·')
address = addresses[0]
# # dist
if len(addresses) != 1:
dist = addresses[1]
else:
dist = ""
print(address, dist)
# type
type = self.type
print(type)
tag_list = job.find_elements(by=By.XPATH,
value='.//div[1]/a/div[2]/ul/li')
if len(tag_list) == 2:
workExperience = tag_list[0].text
educational = tag_list[1].text
else:
workExperience = tag_list[1].text
educational = tag_list[2].text
print(educational, workExperience)
hr_full = job.find_element(by=By.XPATH, value='.//div[@class="info-public"]').text.split()[0]
# hrWork
hrWork = job.find_element(by=By.XPATH, value='.//div[@class="info-public"]/em').text
# hrName
hrName = hr_full.replace(hrWork, "")
print(hrName, hrWork)
# workTag
workTag = job.find_elements(by=By.XPATH, value='.//div[2]/ul/li')
workTag = json.dumps(list(map(lambda x: x.text, workTag)))
print(workTag)
# pratice
pratice = 0
# salary
salaries = job.find_element(by=By.XPATH, value='.//div[1]/a/div[2]/span').text
print(salaries)
if salaries.find("K"):
salaries = salaries.split('-')
if len(salaries) == 1:
salary = list(map(lambda x: int(x) * 1000, salaries[0].replace('K', '').split('-')))
salaryMonth = '0薪'
else:
salary = list(map(lambda x: int(x) * 1000, salaries[0].replace('K', '').split('-')))
salaryMonth = salaries[1]
else:
salary = list(map(lambda x: int(x), salaries.replace('元/天', '').split('-')))
salaryMonth = '0薪'
pratice = 1
print(salary, salaryMonth, pratice)
# companyTitle
companyTitle = job.find_element(by=By.XPATH, value='.//div[1]/div/div[2]/h3/a').text
print(companyTitle)
# companyAvatar
companyAvatar = job.find_element(by=By.XPATH,
value='.//div[1]/div/div[1]/a/img').get_attribute(
'src')
print(companyAvatar)
companyInfo = job.find_elements(by=By.XPATH,
value='.//div[@class="job-card-right"]/div[@class="company-info"]/ul[@class="company-tag-list"]/li')
# 打印公司信息的数量
print(len(companyInfo))
if len(companyInfo) == 3:
# companyNature
companyNature = companyInfo[0].text
# companyStatue
companyStatue = companyInfo[1].text
# companyPeople
companyPeople = companyInfo[2].text
if companyPeople != '10000人以上':
companyPeople = list(map(lambda x: int(x), companyInfo[2].text.replace('人', '').split('-')))
else:
companyPeople = [0, 10000]
else:
# companyNature
companyNature = companyInfo[0].text
# companyStatue
companyStatue = "未融资"
# companyPeople
companyPeople = companyInfo[1].text
if companyPeople != '10000人以上':
companyPeople = list(map(lambda x: int(x), companyInfo[1].text.replace('人', '').split('-')))
else:
companyPeople = [0, 10000]
print(companyNature, companyStatue, companyPeople)
# companyTag
companyTag = job.find_element(by=By.XPATH,
value='.//div[@class="job-card-footer clearfix"]/div[@class="info-desc"]').text
if not companyTag:
companyTag = "无"
else:
companyTag = ', '.join(companyTag.split(','))
print(companyTag)
# detailUrl
detailUrl = job.find_element(by=By.XPATH,
value='.//div[@class="job-card-body clearfix"]/a').get_attribute(
'href')
print(detailUrl)
# companyUrl
companyUrl = job.find_element(by=By.XPATH,
value='.//div[1]/div/div[2]/h3/a').get_attribute(
'href')
print(companyUrl)
print(title, address, dist, type, educational, workExperience, workTag, salary, salaryMonth, companyTag,
hrWork, hrName, pratice, companyTitle, companyAvatar, companyNature, companyStatue, companyPeople,
detailUrl, companyUrl)
jobData.append(title)
jobData.append(address)
jobData.append(type)
jobData.append(educational)
jobData.append(workExperience)
jobData.append(workTag)
jobData.append(salary)
jobData.append(salaryMonth)
jobData.append(companyTag)
jobData.append(hrWork)
jobData.append(hrName)
jobData.append(pratice)
jobData.append(companyTitle)
jobData.append(companyAvatar)
jobData.append(companyNature)
jobData.append(companyStatue)
jobData.append(companyPeople)
jobData.append(detailUrl)
jobData.append(companyUrl)
jobData.append(dist)
self.save_to_csv(jobData)
except Exception as e:
print(e)
pass
self.page += 1
browser.quit()
self.main(self.page)
def clear_csv(self):
df = pd.read_csv('./temp.csv')
df.dropna(inplace=True)
df.drop_duplicates(inplace=True)
df['salaryMonth'] = df['salaryMonth'].map(lambda x: x.replace('薪', ''))
print(f'总数量为{df.shape[0]}')
return df.values
def save_to_csv(self, rowData):
with open('./temp.csv', 'a', newline='', encoding='utf-8') as wf:
writer = csv.writer(wf)
writer.writerow(rowData)
def init(self):
if not os.path.exists('./temp.csv'):
with open('./temp.csv', 'w', encoding='utf-8') as wf:
writer = csv.writer(wf)
writer.writerow(["title",
"address",
"type",
"educational",
"workExperience",
"workTag",
"salary",
"salaryMonth",
"companyTag",
"hrWork",
"hrName",
"pratice",
"companyTitle",
"companyAvatar",
"companyNature", "companyStatue", "companyPeople", "detailUrl",
"companyUrl",
"dist"])
if __name__ == '__main__':
spiderOpj = spider("前端工程师", 1)
spiderOpj.init()
spiderOpj.main(1)
spiderOpj.clear_csv()
结语
通过上述步骤，可以自动爬取招聘网站上的职位信息，并将其整理成结构化的数据。这不仅节省了大量的手动查找和整理时间，还可以为后续的数据分析和决策提供支持。
但是这段代码只是实现了基本的爬虫功能，其实还有改进的空间，特别是在异常处理、代码重复、性能优化和代码安全性方面。