BS直聘职位数据采集与分析（爬虫）-爬虫-CSS教程网

@[## 一、项目介绍

在当今竞争激烈的就业市场中，及时掌握职位信息和市场动态变得尤为重要。本文将详细介绍如何使用Python开发一个爬虫项目，自动采集BOSS直聘网站的职位数据，并对数据进行处理和分析。

1.1 项目意义

帮助求职者快速获取职位信息
为HR提供招聘市场分析数据
辅助研究人员进行就业市场研究
提供薪资水平参考数据

1.2 技术特点

采用面向对象的编程方式
实现了完整的反爬虫机制
支持数据的自动化处理和合并
提供灵活的配置选项

1.3 数据获取方式

API接口爬取
- 主要通过BOSS直聘的API接口 https://www.zhipin.***/wapi/zpgeek/search/joblist.json 获取职位列表
- 使用requests库发送GET请求，携带必要的请求参数和headers
- 返回数据格式为JSON，包含职位的基本信息
详情页解析
- 通过职位ID访问详情页 https://www.zhipin.***/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.***/wapi/zpgeek/search/joblist.json"
    self.headers = {
        "User-Agent": "Mozilla/5.0 ...",
        "Referer": "https://www.zhipin.***/",
        "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.***/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.***/",
    "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.basi***onfig(
    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.***/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.***/",
            "Cookie": "lastCity=101220100; __g=-; Hm_lvt_194df3105ad7148dcf2b98a91b5e727a=1744206399; HMA***OUNT=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%2FDjRjFh***HMKww4w3wpXDiRHCqMOIGFUvNyvDiEg9REjDssOLw6XDgcKOw4fDoMOIbcOLw6TCvT1MSMOJLkESaRxpQUxVW2sPWmFcYWVOFlNPVjpIRkpEc***L0sNHBgcGBYTDxMPExYaEQ0WEw8TDxcSDhIOPEX***sOMWMK8w6DEosO5xKr***k3CocK7xITCosKtZMKSwqXEgWPCtVhUwrPCrsK2wqbCrsKzXsKvw4tNwr7CpFjCqV%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.***/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)

转载请说明出处内容投诉
CSS教程网 » BS直聘职位数据采集与分析（爬虫）

sycup

分享到：

1.1 项目意义

1.2 技术特点

1.3 数据获取方式

二、技术栈

2.1 核心技术

2.2 开发环境

三、项目实现

3.1 项目结构

3.2 核心功能实现

3.2.1 爬虫初始化配置

3.2.2 数据采集流程

3.3 反爬虫策略

3.3.1 请求头处理

3.3.2 请求频率控制

3.4 数据处理

3.4.1 数据结构设计

3.4.2 数据存储

四、项目优化

4.1 性能优化

4.2 稳定性提升

五、实践经验

5.1 常见问题解决

5.2 项目扩展

六、总结

七、参考资料

八、声明

九最终实现效果

十 完整代码

zp_spider_main.py

zp_spider_main.py

sycup

发表评论

一个令你着迷的主题！

十完整代码