BOSS直聘职位数据采集：DrissionPage 监听 XHR 包实战

pip install DrissionPage
# 需要本地已安装 Chrome 浏览器

from DrissionPage import ChromiumPage, ChromiumOptions
import csv, time, hashlib

opts = ChromiumOptions()
# opts.headless()  # 调试时不建议开 headless
page = ChromiumPage(addr_or_opts=opts)

def collect_jobs(keyword: str, city_code: str = "101010100", max_page: int = 10):
    # 采集职位列表
    url = f"https://www.zhipin.com/web/geek/job?query={keyword}&city={city_code}"
    page.get(url)
    time.sleep(2)

    seen = set()
    results = []

    for p in range(1, max_page + 1):
        print(f"[*] 采集第 {p} 页...")

        # 监听 XHR 接口
        page.listen.start("zpgeek/search/joblist.json")
        # 触发翻页（点击下一页按钮或模拟 URL 参数变更）
        if p > 1:
            next_btn = page.ele("css:.options-pages .next")
            if next_btn:
                next_btn.click()
            else:
                break

        packet = page.listen.wait(timeout=10)
        if not packet:
            print("[-] 未收到响应，可能已到最后一页")
            break

        data = packet.response.body
        jobs = data.get("zpData", {}).get("jobList", [])
        for job in jobs:
            uid = hashlib.md5(f"{job['jobName']}{job['brandName']}".encode()).hexdigest()
            if uid in seen:
                continue
            seen.add(uid)
            results.append({
                "职位": job.get("jobName"),
                "公司": job.get("brandName"),
                "薪资": job.get("salaryDesc"),
                "城市": job.get("cityName"),
                "经验": job.get("jobExperience"),
                "学历": job.get("jobDegree"),
                "技能": ",".join(job.get("skills", [])),
                "链接": f"https://www.zhipin.com/job_detail/{job.get('encryptJobId')}.html",
            })
        page.listen.stop()
        time.sleep(1.5)

    return results

def save_csv(jobs: list, filename: str = "jobs.csv"):
    if not jobs:
        print("[-] 无数据可保存")
        return
    with open(filename, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=jobs[0].keys())
        writer.writeheader()
        writer.writerows(jobs)
    print(f"[+] 已保存 {len(jobs)} 条数据到 {filename}")

if __name__ == "__main__":
    jobs = collect_jobs("Python开发", city_code="101010100", max_page=5)
    save_csv(jobs, "python_jobs.csv")
    page.quit()

page.scroll.to_bottom()
time.sleep(1)  # 等待懒加载

opts = ChromiumOptions()
opts.set_user_data_path(r"C:\Users\你的用户名\AppData\Local\Google\Chrome\User Data")
opts.set_argument("--profile-directory=boss_profile")  # 独立 Profile

import random
time.sleep(random.uniform(2, 5))

import pandas as pd

df = pd.read_csv("python_jobs.csv")
# 薪资拆分为最小/最大
df[["min_sal", "max_sal"]] = df["薪资"].str.extract(r"(\d+)-(\d+)K")
df["min_sal"] = pd.to_numeric(df["min_sal"])
df["max_sal"] = pd.to_numeric(df["max_sal"])
# 过滤低薪
df = df[df["min_sal"] >= 15]
df.to_csv("filtered_jobs.csv", index=False, encoding="utf-8-sig")
print(df[["职位", "公司", "薪资"]].head(20))