1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/Bg6shY若失效,可用地址
阿里云:
服务器购买地址
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.部署教程
3.代码如下
# -*- coding: utf-8 -*-
# 使用Boss直聘获取AI工作岗位数据
import requests, pandas as pd, time, random, os, re, json
from lxml import etree
from fake_useragent import UserAgent
from 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-Agent
ua = 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(2, 5))
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(3, 7)) # 添加延迟
# 计算占比
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(3, 7))
# 获取数据
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(10, 15)) # 失败后等待更长时间
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(10, 20))
# 保存最终结果
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)。请求成功后,用多种方式提取数量:
结果区域中"总数"相关节点(如
search-job-result-count
/job-list-header total
);<title>
中的数字;页面任意文本中"(\d+)(个)?职位";
兜底:直接数列表卡片
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
。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论