2025年10月15日星期三

Boss某聘数据爬取任务脚本

1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/Bg6shY若失效,可用地址

1.购买服务器

阿里云:

服务器购买地址

https://t.aliyun.com/U/Bg6shY

若失效,可用地址

https://www.aliyun.com/daily-act/ecs/activity_selection?source=5176.29345612&userCode=49hts92d

腾讯云:

https://curl.qcloud.com/wJpWmSfU

若失效,可用地址

https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console

华为云

https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=201905

2.部署教程

2024年最新青龙面板跑脚本教程(一)持续更新中

3.代码如下

# -*- coding: utf-8 -*-# 使用Boss直聘获取AI工作岗位数据import requests, pandas as pd, time, random, os, re, jsonfrom lxml import etreefrom fake_useragent import UserAgentfrom urllib.parse import quote
# AI相关关键词KEY_WORDS = ["人工智能""AI""算法""机器学习""深度学习"]
# 数据路径DATA_DIR = "data"company_csv = os.path.join(DATA_DIR, "company_list.csv")output_csv = os.path.join(DATA_DIR, "ai_job_ratio_boss.csv")
# 读取公司列表try:    COMPANY_LIST = pd.read_csv(company_csv)["company_name"].tolist()except Exception as e:    print(f"读取公司列表失败: {e}")    COMPANY_LIST = ["百度""阿里巴巴""腾讯""字节跳动""华为"]  # 默认公司
# 生成随机User-Agentua = UserAgent()
def get_boss_count(company, keyword=None):    """从Boss直聘获取职位数量"""
    # 随机UA和请求头    headers = {        "User-Agent": ua.random,        "Accept""text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",        "Accept-Language""zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",        "Connection""keep-alive",        "Referer""https://www.zhipin.com/",        "Cache-Control""no-cache"    }
    # 构建搜索URL    search_term = company if not keyword else f"{company} {keyword}"    search_term_encoded = quote(search_term)
    # Boss直聘搜索URL (替换为正确格式)    url = f"https://www.zhipin.com/web/geek/search?query={search_term_encoded}"
    try:        # 添加随机延迟        time.sleep(random.uniform(25))
        response = requests.get(url, headers=headers, timeout=15)
        if response.status_code == 200:            # 尝试提取职位数量            html = etree.HTML(response.text)
            # 方法1: 提取搜索结果数量            count_elements = html.xpath('//div[contains(@class,"search-job-result")]//span[@class="search-job-result-count"]/text() | //div[contains(@class,"job-list-header")]//div[contains(@class,"total")]/text()')            for element in count_elements:                match = re.search(r'(\d+)', element)                if match:                    return int(match.group(1))
            # 方法2: 从页面标题提取            title = html.xpath('//title/text()')            if title:                match = re.search(r'(\d+)', title[0])                if match:                    return int(match.group(1))
            # 方法3: 提取任何包含数字和"职位"的文本            all_text = html.xpath('//text()')            for text in all_text:                match = re.search(r'(\d+)\s*[个]?职位', text)                if match:                    return int(match.group(1))
            # 如果找不到明确的数字,尝试计算列表中的职位数            job_items = html.xpath('//div[contains(@class,"job-card-wrapper")]')            if job_items:                return len(job_items)
            # 打印部分HTML以便调试            print(f"无法从Boss直聘提取职位数,页面结构可能已变化: {url}")            print(response.text[:500])            return 0        else:            print(f"请求失败,状态码: {response.status_code}")            return 0    except Exception as e:        print(f"请求Boss直聘数据出错: {e}")        return 0
def get_job_ratio(company):    """获取公司的AI职位占比"""
    # 获取公司总职位数    total = get_boss_count(company)    print(f"{company} 总职位数: {total}")
    # 获取AI相关职位数    ai_jobs = 0    for kw in KEY_WORDS:        job_count = get_boss_count(company, kw)        print(f"{company} + {kw} 职位数: {job_count}")        ai_jobs += job_count        time.sleep(random.uniform(37))  # 添加延迟
    # 计算占比    ratio = ai_jobs / total if total > 0 else 0    return {"company": company, "total": total, "ai_jobs": ai_jobs, "ai_ratio": ratio}
def main():    # 重试机制    MAX_RETRIES = 3    results = []
    # 随机打乱公司列表    random_companies = COMPANY_LIST.copy()    random.shuffle(random_companies)
    for company in random_companies:        retries = 0        success = False
        while retries < MAX_RETRIES and not success:            try:                # 初始等待                time.sleep(random.uniform(37))
                # 获取数据                result = get_job_ratio(company)                results.append(result)                print(f"成功获取 {company} 数据: 总职位数 {result['total']}, AI职位数 {result['ai_jobs']}, 占比 {result['ai_ratio']:.2%}")                success = True            except Exception as e:                retries += 1                print(f"{company} 获取失败 (尝试 {retries}/{MAX_RETRIES}): {str(e)}")                time.sleep(random.uniform(1015))  # 失败后等待更长时间
        if not success:            # 记录失败的公司            results.append({"company": company, "total": -1"ai_jobs": -1"ai_ratio": -1})            print(f"{company} 彻底失败,标记为 -1")
        # 每处理5家公司保存一次中间结果        if len(results) % 5 == 0 or len(results) == len(random_companies):            interim_df = pd.DataFrame(results)            interim_csv = os.path.join(DATA_DIR, f"ai_job_ratio_boss_interim_{int(time.time())}.csv")            interim_df.to_csv(interim_csv, index=False, encoding="utf_8_sig")            print(f"中间结果已保存至 {interim_csv}")
        # 处理完一家公司后等待较长时间        time.sleep(random.uniform(1020))
    # 保存最终结果    df = pd.DataFrame(results)    df.to_csv(output_csv, index=False, encoding="utf_8_sig")    print(f"数据已保存至 {output_csv}")
    # 输出统计信息    success_count = len(df[df['total'] >= 0])    print(f"成功获取: {success_count}/{len(COMPANY_LIST)} 公司数据")
    if success_count > 0:        # 排除失败的公司        success_df = df[df['total'] >= 0]        if not success_df.empty and not success_df['ai_ratio'].isna().all():            print(f"平均AI职位占比: {success_df['ai_ratio'].mean():.2%}")            max_idx = success_df['ai_ratio'].idxmax()            print(f"AI职位占比最高的公司: {success_df.loc[max_idx]['company']} ({success_df['ai_ratio'].max():.2%})")
if __name__ == "__main__":    # 确保数据目录存在    os.makedirs(DATA_DIR, exist_ok=True)    main()
解析

脚本主要作用

  • 读取公司名单(data/company_list.csv 的 company_name 列,失败则用默认五家公司)。

  • 逐家公司在 Boss直聘 上检索:

    • 获取该公司的总在招职位数

    • 再用 5 个 AI 关键词("人工智能/AI/算法/机器学习/深度学习")分别检索并累加 AI 相关职位数

    • 计算 AI 职位占比 = ai_jobs / total

  • 将结果周期性写入中间文件,并最终落盘到 data/ai_job_ratio_boss.csv,同时打印统计信息(成功数量、平均占比、最高占比公司)。

关键常量/数据

  • KEY_WORDS:AI 相关关键词列表。

  • DATA_DIR / company_csv / output_csv:数据输入输出路径。

  • COMPANY_LIST:公司名称清单(CSV 读取失败时使用内置默认列表)。

  • ua = UserAgent():随机 UA 生成器(反爬伪装用)。

主要方法

get_boss_count(company, keyword=None)

功能:到 Boss 直聘搜索页抓取"职位条数"。
输入:公司名,及可选的关键词(与公司名一起拼为查询词)。
流程

  • 拼接搜索 URL:https://www.zhipin.com/web/geek/search?query=...

  • 随机 UA + 常见浏览器头;随机 sleep(2–5s)。

  • 请求成功后,用多种方式提取数量:

    1. 结果区域中"总数"相关节点(如 search-job-result-count / job-list-header total);

    2. <title> 中的数字;

    3. 页面任意文本中"(\d+)(个)?职位";

    4. 兜底:直接数列表卡片 job-card-wrapper 的数量。

  • 失败或未匹配则返回 0,并打印少量 HTML 片段帮助调试。

说明:页面计数规则可能变化,函数靠多 XPath/正则做"能取到就行"的鲁棒提取,准确性依赖页面结构。

get_job_ratio(company)

功能:计算单个公司的 AI 职位占比。
流程

  • 调 get_boss_count(company) 得到 总职位数 total

  • 遍历 KEY_WORDS,每个关键词调用 get_boss_count(company, kw),将得到的数量相加为 AI 职位数 ai_jobs(中间带 3–7s 随机延迟降频);

  • 返回 {"company", "total", "ai_jobs", "ai_ratio"}

注意:关键词结果直接累加,同一岗位可能被多个关键词命中,存在重复计数的可能,这个口径更偏"相关热度"而非去重后的精确 AI 岗位数。

main()

功能:整体调度、容错与落盘。
要点

  • 将公司列表随机打乱处理(降低被风控概率)。

  • 每家公司最多重试 3 次;失败会延长等待并记录为 -1

  • 每处理 5 家或处理完全部,就保存一次中间结果到 data/ai_job_ratio_boss_interim_<timestamp>.csv

  • 最终结果写入 data/ai_job_ratio_boss.csv

  • 打印汇总:成功公司数、平均占比、占比最高公司。

运行与输出

  • 直接运行 __main__:会确保 data/ 目录存在,然后执行全流程。

  • 输出 CSV 字段:company / total / ai_jobs / ai_ratio



注意

本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。


历史脚本txt文件获取>>
服务器搭建,人工服务咨询>>

没有评论:

发表评论

波士顿(BCG)《2025年全球支付报告》:未来(Anything but)“稳定”

即时支付、数字货币、生成式人工智能以及代理式人工智能——正在迅速普及。软件平台在支付业务收入中的占比不断上升,这要求产品具备更强的功能、更高的速度以及更完善的整合能力。 支付行业正在多个方面经历重大变革。 曾经处于边缘地位的创新技术——即时支付、数字货币、生成式人工智能以及代...