一、项目介绍
在当今竞争激烈的就业市场中，及时掌握职位信息和市场动态变得尤为重要。本文将详细介绍如何使用Python开发一个爬虫项目，自动采集BOSS直聘网站的职位数据，并对数据进行处理和分析。
1.1 项目意义
帮助求职者快速获取职位信息
为HR提供招聘市场分析数据
辅助研究人员进行就业市场研究
提供薪资水平参考数据
1.2 技术特点
采用面向对象的编程方式
实现了完整的反爬虫机制
支持数据的自动化处理和合并
提供灵活的配置选项
1.3 数据获取方式
API接口爬取
主要通过BOSS直聘的API接口 https://www.zhipin.com/wapi/zpgeek/search/joblist.json 获取职位列表
使用requests库发送GET请求，携带必要的请求参数和headers
返回数据格式为JSON，包含职位的基本信息
详情页解析
通过职位ID访问详情页 https://www.zhipin.com/job_detail/{job_id}.html
使用BeautifulSoup4解析HTML页面
提取职位描述等详细信息
二、技术栈
2.1 核心技术
Python 3.x：编程语言
requests：网络请求库
BeautifulSoup4：HTML解析
pandas：数据处理
openpyxl：Excel文件操作
2.2 开发环境
操作系统：支持Windows/Linux/MacOS
IDE：推荐PyCharm或VS Code
依赖管理：pip
三、项目实现
3.1 项目结构
bosszhipin_spider/
├── README.md # 项目说明文档
├── zp_spider_main.py # 主爬虫程序
├── merge_job_excel.py # Excel数据合并工具
└── *.xlsx # 爬取的数据文件
3.2 核心功能实现
3.2.1 爬虫初始化配置
def __init__(self):
self.base_url = "https://www.zhipin.com/wapi/zpgeek/search/joblist.json"
self.headers = {
"User-Agent": "Mozilla/5.0 ...",
"Referer": "https://www.zhipin.com/",
"Cookie": "your_cookie_here"
}
self.params = {
"query": "Python高级开发工程师",
"city": "101220100",
"experience": "106",
"scale": "303,304,305",
"page": 1,
"pageSize": 30
}
3.2.2 数据采集流程
获取职位列表
def fetch_data(self, max_pages=3):
for page in range(1, max_pages + 1):
self.params['page'] = page
resp = requests.get(self.base_url,
headers=self.headers,
params=self.params)
# 处理响应数据
job_list = resp.json().get("zpData", {}).get("jobList", [])
for job in job_list:
job_id = job.get("encryptJobId")
job_desc = self.get_job_detail(job_id)
self.process_job_data(job, job_desc)
解析职位详情
def get_job_detail(self, job_id):
url = f"https://www.zhipin.com/job_detail/{job_id}.html"
resp = requests.get(url, headers=self.detail_headers)
soup = BeautifulSoup(resp.text, 'html.parser')
desc_tag = soup.select_one('.job-sec-text')
return desc_tag.text.strip() if desc_tag else ""
3.3 反爬虫策略
3.3.1 请求头处理
headers = {
"User-Agent": "Mozilla/5.0 ...",
"Referer": "https://www.zhipin.com/",
"Cookie": "your_cookie_here"
}
3.3.2 请求频率控制
import random
import time
def request_with_delay(self):
time.sleep(random.uniform(1, 1.5)) # 随机延时
# 执行请求操作
3.4 数据处理
3.4.1 数据结构设计
def process_job_data(self, job, job_desc):
job_data = {
"职位": job.get("jobName"),
"公司": job.get("brandName"),
"薪资": job.get("salaryDesc"),
"地区": job.get("cityName"),
"经验": job.get("jobExperience"),
"学历": job.get("jobDegree"),
"公司规模": job.get("brandScaleName"),
"行业": job.get("brandIndustry"),
"福利标签": ",".join(job.get("welfareList", [])),
"技能标签": ",".join(job.get("skills", [])),
"职位描述": job_desc
}
self.data_list.append(job_data)
3.4.2 数据存储
def save_excel(self):
filename = f"python岗位数据_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
df = pd.DataFrame(self.data_list)
df.to_excel(filename, index=False)
print(f"数据已保存到: {filename}")
四、项目优化
4.1 性能优化
异步请求处理
使用aiohttp实现异步爬取
控制并发数量
优化请求队列
数据处理优化
使用生成器处理大数据
实现增量更新
优化内存使用
4.2 稳定性提升
异常处理机制
def safe_request(self, url, retries=3):
for i in range(retries):
try:
response = requests.get(url, headers=self.headers)
response.raise_for_status()
return response
except Exception as e:
print(f"请求失败 ({i+1}/{retries}): {str(e)}")
if i == retries - 1:
raise
time.sleep(2 ** i) # 指数退避
日志记录
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='spider.log'
)
五、实践经验
5.1 常见问题解决
Cookie失效问题
实现Cookie池
定期更新Cookie
多账号轮换
IP限制处理
使用代理IP池
动态IP切换
请求频率控制
数据质量保证
数据完整性检查
字段格式验证
重复数据处理
5.2 项目扩展
数据分析功能
def analyze_salary(self):
df = pd.DataFrame(self.data_list)
salary_stats = df['薪资'].value_counts()
plt.figure(figsize=(10, 6))
salary_stats.plot(kind='bar')
plt.title('职位薪资分布')
plt.show()
可视化展示
使用matplotlib绘制图表
集成echarts可视化
开发Web展示界面
自动化部署
Docker容器化
定时任务调度
监控告警机制
六、总结
本项目通过Python实现了BS直聘职位数据的自动化采集和处理。在开发过程中，我们重点解决了以下问题：
反爬虫机制的突破
模拟浏览器行为
控制请求频率
使用代理IP
大量数据的高效处理
异步请求优化
数据结构优化
存储方式优化
程序的稳定性保证
完善的异常处理
日志记录机制
数据备份策略
七、参考资料
Python官方文档
Requests库文档
BeautifulSoup4文档
Pandas文档
八、声明
本项目仅供学习交流使用，请勿用于商业用途。使用本项目时请遵守相关法律法规，尊重网站的robots协议。对于因使用本项目造成的任何问题，本项目不承担任何责任。
九最终实现效果
十 完整代码
zp_spider_main.py
import requests
import pandas as pd
import time
import random
from bs4 import BeautifulSoup
import datetime
class BossSpiderAPI:
def __init__(self):
"""
Cookie需要登录后去网页上获取
"""
self.base_url = "https://www.zhipin.com/wapi/zpgeek/search/joblist.json"
self.headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
"Referer": "https://www.zhipin.com/",
"Cookie": "lastCity=101220100; __g=-; Hm_lvt_194df3105ad7148dcf2b98a91b5e727a=1744206399; HMACCOUNT=42C280446392FB0C; wt2=DlNySPpvwfKbhVut0Gr0mgKTqXmBwfqcUqRyST97FmpG1q4fuYfpOEhCiOr1dttN64ZIAMYJ2WlH238k6edxPQw~~; wbg=0; zp_at=5MyB3H6qawmfWoNUp314MzOYgaBoo-yV_45NFKxF4IU~; Hm_lpvt_194df3105ad7148dcf2b98a91b5e727a=1744208479; bst=V2QNMjEeP_0llgXdJtyh0QICmw7D7Rwg~~|QNMjEeP_0llgXdJtyh0QICmw7DvRwQ~~; __c=1744206399; __a=15610779.1712562134.1730179058.1744206399.26.3.12.26; __zp_stoken__=6578fRUnDnsOIwp3DiD42DRYRGxg9LUxJOzJLRTVDSUlFSUVLS0VJPSk9NTLDhl%2FDg8OSX8OcwrzCvUY2RUlFRUlDSURCJkU9wr5FSjA2w4Njw4rDkl%2FDjRjFhcOMHMKww4w3wpXDiRHCqMOIGFUvNyvDiEg9REjDssOLw6XDgcKOw4fDoMOIbcOLw6TCvT1MSMOJLkESaRxpQUxVW2sPWmFcYWVOFlNPVjpIRkpEccOML0sNHBgcGBYTDxMPExYaEQ0WEw8TDxcSDhIOPEXCnsOMWMK8w6DEosO5xKrCnk3CocK7xITCosKtZMKSwqXEgWPCtVhUwrPCrsK2wqbCrsKzXsKvw4tNwr7CpFjCqV%2FCrlFXw4hPHMKAd8OEw4p4WcK%2BVQ4ZwokPFEwVH1nDlw%3D%3D"
}
self.detail_headers = {
"User-Agent": self.headers["User-Agent"],
"Cookie": self.headers["Cookie"]
}
self.params = {
"scene": "1",
"query": "Python高级开发工程师",
"city": "101220100",# 城市：福州
"experience": "106",# 经验：3-5年
"scale": "303,304,305",# 公司规模：100-999人
"page": 1,
"pageSize": 30
}
self.data_list = []
def fetch_data(self, max_pages=3):
for page in range(1, max_pages + 1):
print(f"抓取第 {page} 页...")
self.params['page'] = page
try:
resp = requests.get(self.base_url, headers=self.headers, params=self.params)
resp.raise_for_status()
except requests.RequestException as e:
print(f"请求失败：{e}")
continue
result = resp.json()
job_list = result.get("zpData", {}).get("jobList", [])
if not job_list:
print("没有更多数据了")
break
for job in job_list:
job_id = job.get("encryptJobId")
job_desc = self.get_job_detail(job_id)
item = {
"职位": job.get("jobName"),
"公司": job.get("brandName"),
"薪资": job.get("salaryDesc"),
"地区": job.get("cityName"),
"经验": job.get("jobExperience"),
"学历": job.get("jobDegree"),
"公司规模": job.get("brandScaleName"),
"行业": job.get("brandIndustry"),
"福利标签": ",".join(job.get("welfareList", [])),
"技能标签": ",".join(job.get("skills", [])),
"职位描述": job_desc
}
self.data_list.append(item)
time.sleep(random.uniform(1, 1.5))
def get_job_detail(self, job_id):
url = f"https://www.zhipin.com/job_detail/{job_id}.html"
try:
resp = requests.get(url, headers=self.detail_headers)
if resp.status_code != 200:
return ""
soup = BeautifulSoup(resp.text, 'html.parser')
desc_tag = soup.select_one('.job-sec-text')
return desc_tag.text.strip() if desc_tag else ""
except Exception as e:
print(f"详情页获取失败: {e}")
return ""
def save_excel(self):
filename = f"python岗位数据_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
df = pd.DataFrame(self.data_list)
df.to_excel(filename, index=False)
print(f"保存成功：{filename}，共 {len(df)} 条职位")
def run(self):
self.fetch_data()
self.save_excel()
if __name__ == "__main__":
spider = BossSpiderAPI()
spider.run()
zp_spider_main.py
import os
import pandas as pd
from openpyxl import load_workbook
def merge_excelfiles(dir_path, save_path):
file_list = [os.path.join(dir_path, f) for f in os.listdir(dir_path) if f.endswith(".xlsx")]
df_list = []
headers = None
for file in file_list:
try:
wb = load_workbook(file, data_only=True)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
data = list(ws.values)
if not data:
print(f"⚠️ 跳过空文件或空Sheet: {file} - {sheet_name}")
continue
if headers is None:
headers = list(data[0])
data_rows = data[1:] # 去掉表头
else:
if list(data[0]) == headers:
data_rows = data[1:]
else:
data_rows = data # 有些 sheet 可能没有表头
if not data_rows:
continue
df = pd.DataFrame(data_rows, columns=headers)
df_list.append(df)
except Exception as e:
print(f"❌ 读取文件失败: {file}，错误: {str(e)}")
continue
if df_list:
merge_data = pd.concat(df_list, axis=0)
merge_data.to_excel(save_path, index=False)
print(f"✅ 合并完成，保存至: {save_path}")
else:
print("⚠️ 没有数据可以合并。")
# 示例路径
dir_path = "/Users/melon/Desktop/zj/odoo18/Celery/bosszhipin_spider" # 设置Excel所在目录
save_path = "merged_后端开发.xlsx" # 合并后的保存路径
merge_excelfiles(dir_path, save_path)