某联招聘数据爬取任务脚本

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 -*-import requests, pandas as pd, time, randomfrom lxml import etreefrom fake_useragent import UserAgentimport reimport json
KEY_WORDS = ["人工智能", "AI", "算法", "机器学习", "深度学习"]import osDATA_DIR = "data"company_xlsx = os.path.join(DATA_DIR, "company_list.xlsx")company_csv = os.path.join(DATA_DIR, "company_list.csv")output_csv = os.path.join(DATA_DIR, "ai_job_ratio_zhaopin.csv")try:    COMPANY_LIST = pd.read_excel(company_xlsx)["company_name"].tolist()  # 一列公司全称except Exception:    COMPANY_LIST = pd.read_csv(company_csv)["company_name"].tolist()
# 生成随机User-Agentua = UserAgent()HEADERS = {"User-Agent": ua.random}
# 使用API模式获取职位数def extract_number(text):    if not text:        return 0    # 提取文本中的数字    match = re.search(r'\d+', text)    if match:        return int(match.group())    return 0
def get_job_count_api(url, keyword=None):    """使用API方式获取职位数量，避免网站反爬"""
    # 生成随机User-Agent    headers = {        "User-Agent": ua.random,        "Accept": "application/json, text/javascript, */*; q=0.01",        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",        "Referer": "https://sou.zhaopin.com/",        "Origin": "https://sou.zhaopin.com",        "Connection": "keep-alive",        "X-Requested-With": "XMLHttpRequest"    }
    # 提取公司名和关键词    company = url.split("kw=")[1].split("&")[0]    search_term = f"{company}" if not keyword else f"{company}+{keyword}"
    # 构建API URL (模拟智联招聘AJAX请求)    api_url = f"https://fe-api.zhaopin.com/c/i/search/positions?keyword={search_term}&page=1&pageSize=1"
    try:        # 添加随机延迟        time.sleep(random.uniform(1, 3))
        r = requests.get(api_url, headers=headers, timeout=15)        if r.status_code == 200:            try:                data = r.json()                # 尝试从JSON响应中获取总数                if 'data' in data and 'numFound' in data['data']:                    return int(data['data']['numFound'])
                # 备用方法：尝试从HTML中提取                html = etree.HTML(r.text)                count_text = html.xpath('//text()[contains(., "个职位")]')                if count_text:                    return extract_number(count_text[0])
                return 0            except Exception as e:                print(f"解析响应失败: {e}")                # 如果JSON解析失败，尝试直接解析HTML                return 0        else:            # 如果API调用失败，尝试普通页面请求            normal_url = f"https://sou.zhaopin.com/?kw={search_term}"            r2 = requests.get(normal_url, headers=headers, timeout=15)            html = etree.HTML(r2.text)
            # 尝试不同的XPath表达式            count_expressions = [                '//span[@class="search-result__num__1"]/text()',                '//div[contains(@class, "search-result")]//*[contains(text(), "共")]/text()',                '//*[contains(text(), "个职位")]/text()'            ]
            for expr in count_expressions:                result = html.xpath(expr)                if result:                    return extract_number(result[0])
            return 0    except Exception as e:        print(f"请求失败: {e}")        return 0
def extract_number(text):    if not text:        return 0    # 提取文本中的数字    match = re.search(r'\d+', text)    if match:        return int(match.group())    return 0
def get_job_ratio(comp):    # 智联招聘 URL    url = f"https://sou.zhaopin.com/?kw={comp}&p=1"
    # 使用API方式获取总职位数    total = get_job_count_api(url)    print(f"{comp} 总职位数: {total}")
    # 获取AI相关职位数    ai_jobs = 0    for kw in KEY_WORDS:        url_kw = f"https://sou.zhaopin.com/?kw={comp}+{kw}&p=1"        job_count = get_job_count_api(url_kw, kw)        print(f"{comp} + {kw} 职位数: {job_count}")        ai_jobs += job_count        time.sleep(random.uniform(3, 5))  # 增加间隔，降低被封风险
    ratio = ai_jobs / total if total else 0    return {"company": comp, "total": total, "ai_jobs": ai_jobs, "ai_ratio": ratio}
def main():    # 增加错误重试机制    MAX_RETRIES = 3    res = []
    # 使用随机顺序处理公司名单，降低被识别风险    random_companies = COMPANY_LIST.copy()    random.shuffle(random_companies)
    for c in random_companies:        retries = 0        success = False
        while retries < MAX_RETRIES and not success:            try:                # 随机等待一段时间再开始下一个公司的抓取                if retries == 0:                    time.sleep(random.uniform(2, 5)) 
                result = get_job_ratio(c)                res.append(result)                print(f"成功获取 {c} 数据: 总职位数 {result['total']}, AI职位数 {result['ai_jobs']}, 占比 {result['ai_ratio']:.2%}")                success = True            except Exception as e:                retries += 1                print(f"{c} 获取失败 (尝试 {retries}/{MAX_RETRIES}): {str(e)}")                time.sleep(random.uniform(5, 10))  # 失败后等待更长时间                # 更换随机User-Agent                global HEADERS                HEADERS = {"User-Agent": ua.random}
        if not success:            # 记录失败的公司            res.append({"company": c, "total": -1, "ai_jobs": -1, "ai_ratio": -1})            print(f"{c} 彻底失败，标记为 -1")
        # 每完成一定数量的公司，保存一次中间结果，避免全部失败        if len(res) % 5 == 0 or len(res) == len(random_companies):            interim_df = pd.DataFrame(res)            interim_csv = os.path.join(DATA_DIR, f"ai_job_ratio_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(8, 15))
    # 输出结果    df = pd.DataFrame(res)    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]        print(f"平均AI职位占比: {success_df['ai_ratio'].mean():.2%}")        if len(success_df) > 0:            max_ratio_idx = success_df['ai_ratio'].idxmax() if not success_df['ai_ratio'].isna().all() else 0            print(f"AI职位占比最高的公司: {success_df.loc[max_ratio_idx]['company']} ({success_df['ai_ratio'].max():.2%})")
if __name__ == "__main__":    main()

解析

脚本主要作用

• 目标：基于智联招聘检索接口/页面，统计一批公司的总在招职位数与AI相关职位数（"人工智能/AI/算法/机器学习/深度学习"五类关键词），计算AI职位占比并输出到 data/ai_job_ratio_zhaopin.csv。

• 数据源：优先从 data/company_list.xlsx 读取公司名（列名 company_name）；若失败则读 data/company_list.csv。

• 反爬应对：随机 User-Agent、随机等待、失败重试；优先走 fe-api.zhaopin.com 的"类 API"接口，失败再回退到普通搜索页做 HTML 解析。

主要方法

extract_number(text)

• 功能：从一段文本里抽取第一个整数（如"共123个职位"→123）。

• 用途：做多种页面/文本解析时的统一"取数字"工具。

get_job_count_api(url, keyword=None)

• 功能：查询某公司（可携带关键词）的职位数量。

• 做法：

1. 伪装请求头（随机 UA、Referer、X-Requested-With 等）；

2. 从传入的 URL 中取出 kw= 参数作为公司名，拼出"类 API"地址：

   https://fe-api.zhaopin.com/c/i/search/positions?keyword=公司或公司+关键词&page=1&pageSize=1
   

3. 优先解析 JSON：若响应含 data.numFound，直接返回为总数；

4. 兜底解析：若 JSON 取不到，再尝试把响应当 HTML，使用 XPath/文本包含"个职位"等提示词结合 extract_number 取数；

5. 再兜底：若 API 失败则访问普通搜索页 https://sou.zhaopin.com/?kw=...，尝试多组 XPath 匹配提取职位总数；

6. 过程内置随机 sleep 以降频。

• 返回：尽量返回一个整数职位数，失败则 0。

get_job_ratio(comp)

• 功能：计算某公司 AI 岗位占比。

• 流程：

1. 调 get_job_count_api 拿总职位数；

2. 依次用五个 AI 关键词调用 get_job_count_api 求和得 AI职位数（关键词之间相加，可能有重叠，这里无去重逻辑）；

3. 计算 ai_ratio = ai_jobs / total（total 为 0 时占比 0）。

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

main()

• 功能：整体调度、容错和落盘。

• 要点：

• 将公司列表随机打乱；

• 每家公司最多重试 3 次（失败加长等待并更换随机 UA）；

• 每处理 5 家（或完成全部）就保存一次中间结果到 data/ai_job_ratio_interim_<timestamp>.csv；

• 最终将所有结果写到 data/ai_job_ratio_zhaopin.csv；

• 最后打印统计：成功公司数、平均占比、最高占比公司。

注意：

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

历史脚本txt文件获取>>

服务器搭建，人工服务咨询>>