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=2019052.部署教程
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"}# 构建搜索URLsearch_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 0else:print(f"请求失败,状态码: {response.status_code}")return 0except Exception as e:print(f"请求Boss直聘数据出错: {e}")return 0def get_job_ratio(company):"""获取公司的AI职位占比"""# 获取公司总职位数total = get_boss_count(company)print(f"{company} 总职位数: {total}")# 获取AI相关职位数ai_jobs = 0for kw in KEY_WORDS:job_count = get_boss_count(company, kw)print(f"{company} + {kw} 职位数: {job_count}")ai_jobs += job_counttime.sleep(random.uniform(3, 7)) # 添加延迟# 计算占比ratio = ai_jobs / total if total > 0 else 0return {"company": company, "total": total, "ai_jobs": ai_jobs, "ai_ratio": ratio}def main():# 重试机制MAX_RETRIES = 3results = []# 随机打乱公司列表random_companies = COMPANY_LIST.copy()random.shuffle(random_companies)for company in random_companies:retries = 0success = Falsewhile 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 = Trueexcept Exception as e:retries += 1print(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。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论