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 -*-# 使用拉勾网获取AI工作岗位数据import requests, pandas as pd, time, random, os, re, jsonfrom fake_useragent import UserAgent# 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_lagou.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_lagou_count(company, keyword=None):"""从拉勾网获取职位数量（使用其JSON API）"""# 构建搜索关键词search_term = company if not keyword else f"{company} {keyword}"# 构建拉勾网API请求头（模拟其前端AJAX请求）headers = {"User-Agent": ua.random,"Accept": "application/json, text/javascript, */*; q=0.01","Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7","Content-Type": "application/x-www-form-urlencoded; charset=UTF-8","Origin": "https://www.lagou.com","Referer": "https://www.lagou.com/jobs/list_" + requests.utils.quote(search_term),"X-Requested-With": "XMLHttpRequest"}# 构建POST请求参数data = {"first": "true","pn": "1","kd": search_term,"sid": ""}# 拉勾网搜索APIurl = "https://www.lagou.com/jobs/positionAjax.json"# 拉勾网的反爬机制较强，需要先访问Referer页面获取cookiesession = requests.Session()try:# 先访问搜索页获取cookiereferer_url = f"https://www.lagou.com/jobs/list_{requests.utils.quote(search_term)}"session.get(referer_url, headers={"User-Agent": headers["User-Agent"]}, timeout=15)# 添加随机延迟time.sleep(random.uniform(2, 5))# 然后请求APIresponse = session.post(url, headers=headers, data=data, timeout=15)if response.status_code == 200:try:# 尝试解析JSONresp_json = response.json()# 拉勾API返回的是一个较复杂的JSON结构if "content" in resp_json and "positionResult" in resp_json["content"]:result = resp_json["content"]["positionResult"]if "totalCount" in result:return int(result["totalCount"])# 如果无法解析结构，记录以便调试print(f"拉勾网返回结构异常: {json.dumps(resp_json)[:200]}...")return 0except json.JSONDecodeError:# 如果不是JSON，可能是被反爬拦截了print("拉勾网返回非JSON响应，可能被反爬拦截")print(response.text[:500])return 0else:print(f"请求失败，状态码: {response.status_code}")return 0except Exception as e:print(f"请求拉勾网数据出错: {e}")return 0def get_job_ratio(company):"""获取公司的AI职位占比"""# 获取公司总职位数total = get_lagou_count(company)print(f"{company} 总职位数: {total}")# 获取AI相关职位数ai_jobs = 0for kw in KEY_WORDS:job_count = get_lagou_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_lagou_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(15, 25)) # 拉勾网反爬较强，等待更久# 保存最终结果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读取公司名单（失败则用内置 5 家大厂）。对每个公司在 拉勾网 检索：
获取该公司的总在招职位数；
用 5 个 AI 关键词（"人工智能/AI/算法/机器学习/深度学习"）分别检索并累加 AI 相关职位数；
计算 AI 职位占比 = ai_jobs / total。
将结果周期性写入中间文件，最终保存到
data/ai_job_ratio_lagou.csv，并打印统计概览（成功公司数、平均占比、最高占比公司）。
关键配置/数据
KEY_WORDS：AI 关键词列表。DATA_DIR / company_csv / output_csv：输入输出路径。COMPANY_LIST：公司清单（CSV 读取失败时用默认列表）。ua = UserAgent()：随机 UA，配合随机延时，降低反爬触发。
主要方法说明
get_lagou_count(company, keyword=None)
用途：查询某公司（可叠加关键词）的职位总数（走拉勾 JSON 接口）。
做法：
组合检索词
search_term（公司名或"公司+关键词"）。先访问搜索页（拿 Cookie），再
POST https://www.lagou.com/jobs/positionAjax.json。解析返回 JSON 的
content.positionResult.totalCount作为职位数。若返回非 JSON/被反爬拦截，打印片段做诊断并返回 0。
加了随机 UA 与 2–5 秒延时，尽量模拟正常流量。
get_job_ratio(company)
用途：计算单个公司的 AI 职位占比。
流程：
调
get_lagou_count(company)得到 总职位数total；遍历
KEY_WORDS，对每个关键词调用get_lagou_count(company, kw)，将结果累加为 AI 职位数ai_jobs（中间 3–7 秒延时）；返回
{"company","total","ai_jobs","ai_ratio"}。
注：关键词直接累加，可能把同一职位重复计入（更像"相关热度"的估算口径）。
main()
用途：整体调度、容错与落盘。
要点：
公司顺序打乱（降低风控概率）。
单公司最多重试 3 次；失败记为
-1。每处理 5 家或全部完成，写一次中间结果
ai_job_ratio_lagou_interim_<timestamp>.csv。最终写
ai_job_ratio_lagou.csv，并打印：成功公司数；
平均 AI 占比；
AI 占比最高公司。
运行与输出
直接执行脚本会确保
data/存在并跑完整流程。输出 CSV 字段：
company / total / ai_jobs / ai_ratio。
注意：
本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。
没有评论:
发表评论