--- title: "01-baidu_data_proc_api" created: 2026-04-02 tags: - 项目 aliases: - baidu_data_proc_api --- # baidu_data_proc_api.py ### `baidu_data_proc_api.py` — 百度招聘直连 `api_proc` 总入口所在文件。直接 POST 百度招聘后台 API 拿分页 JSON，带代理池（`getProxy` 递归重试取可用代理）绕过 IP 限制。模块概览见 [[00-auto_api|auto_api 主篇]]。 ## 代码 ```python import time import hashlib import os import requests from urllib.parse import urlencode import sys sys.path.append('../') import json from utils import ner_logger import re from auto_api.isoftstone_data_proc_api import api_proc_isoftstone # 请求头配置，模拟浏览器访问百度招聘 headers = { "Accept": "application/json, text/plain, */*", "Content-Type": "application/x-www-form-urlencoded;charset=utf-8", "Origin": "https://talent.baidu.com", "Referer": "https://talent.baidu.com/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.3 Safari/605.1.15", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh-Hans;q=0.9", "Connection": "keep-alive", } # 登录Cookie，保持会话状态 cookie_str = ('RT="z=1&dm=baidu.com&si=6ece4843-c53f-43cb-9190-cdeb1e60fc10&ss=mgkm3gti&sl=2&tt=m7' '&bcn=https%3A%2F%2Ffclog.baidu.com%2Flog%2Fweirwood%3Ftype%3Dperf&ld=3o9k&ul=3pos&hd=3pp1"; ' 'Hm_lpvt_50e85ccdd6c1e538eb1290bc92327926=1760086734; ' 'Hm_lvt_50e85ccdd6c1e538eb1290bc92327926=1760083124; HMACCOUNT=BABEAB1BF5E31B7B; ' 'H_WISE_SIDS=60279_63144_63325_64314_64650_64695_64814_64817_64866_64840_64909_64913_64965_64988_65005_65003_65120_65141_65140_65137_65190_65203_65246_65255_65143_65273_65315_65322_65367; ' 'BAIDUID=DF45EE91831D9BA2A80B24F2DCE23BB1:FG=1; ' 'BIDUPSID=DF45EE91831D9BA2D6517973C86A9556; H_PS_PSSID=60272_63140_63325_64651_64702_64813_64815_64840_64873_64904_64923_64986_65120_65141_65140_65138_65187_65203_65216_65249_65255_65144_65277_65309_65327_65373_65367; PSTM=1758515092') headers["Cookie"] = cookie_str # 获取代理IP配置 def myProxy(): # 设置请求代理 _ps = getProxy() proxies = { "http": f"http://{_ps}", } ner_logger.info("getProxy:", proxies) return proxies # 请求百度招聘接口，获取职位JSON数据 def get_baidu_job_json(url,recruitType,projectType,curPage): # 构造请求参数 payload_dict = { "IME类型": "application/x-www-form-urlencoded;charset=utf-8", "recruitType": recruitType, # 招聘类型：社招/校招/实习 "pageSize": 20, # 每页20条 "keyWord": "", # 搜索关键词 "curPage": curPage, # 当前页码 "projectType": projectType, # 项目类型 } payload = urlencode(payload_dict) # 发送POST请求 with requests.Session() as s: resp = s.post(url, data=payload, headers=headers, timeout=15, verify=False,proxies=myProxy()) print("Status:", resp.status_code) # 解析返回的JSON数据 try: json = resp.json() if json['status'] == 'ok': data = json['data']['list'] total = int(json['data']['total']) return True,data,total except Exception: ner_logger.info("baidu Text response:", resp.text) return False,-1 # 下载职位详情页HTML并保存 def get_baidu_job_html(url,tmp_file): # 请求头配置 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Referer": "https://talent.baidu.com/" } try: # 发送GET请求获取页面 response = requests.get(url, headers=headers, timeout=10, verify=False,proxies=myProxy()) response.raise_for_status() response.encoding = response.apparent_encoding # 清除JS脚本内容 full_text = re.sub(r']*?>.*?', '', response.text, flags=re.DOTALL) # 写入HTML文件 with open(tmp_file, "w", encoding="utf-8") as f: f.write(full_text) time.sleep(10) except requests.exceptions.RequestException as e: print(f"请求失败：{e}") return None # 将百度原始职位数据转换为统一格式 def transform_job_json(item,recruitType,job_type,channel,target_url,tmp_file,json_file): """ 将源JSON转换为目标JSON格式 """ # 字段映射关系 field_mapping = { "announcement_name": "name", "publish_time": "publishDate", "hd_dept": "bgShortName", "hd_loc": "workPlace", "hd_job_num": "recruitNum", "hd_job_category": "postType" } # 固定字段 fixed_fields = { "link": target_url, "full_url": target_url, "last_url": target_url, "file_path": tmp_file, "parent_url": "https://talent.baidu.com/static/index.html", "channel": channel, "job_type": job_type } target_json = {} # 映射字段 for target_field, source_field in field_mapping.items(): target_json[target_field] = item.get(source_field, "") # 添加固定字段 target_json.update(fixed_fields) # 保存JSON文件 with open(json_file, 'w', encoding='utf-8') as f: json.dump(target_json, f, ensure_ascii=False, indent=4) time.sleep(10) # API处理分发入口 def api_proc(spider_com,_key, com_info,k,url,_stat): if _key == "com_90001": api_proc_baidu(spider_com,_key, com_info,k,url,_stat) if _key == "com_90002": api_proc_isoftstone(spider_com,_key, com_info,k,url,_stat) if _key == "com_90003": from auto_api.jd_data_proc_api import api_proc_jd api_proc_jd(spider_com,_key, com_info,k,url,_stat) if _key == "com_90004": from auto_api.kingdee_data_proc_api import api_proc_kingdee api_proc_kingdee(spider_com,_key, com_info,k,url,_stat) if _key == "com_90005": from auto_api.picc_data_proc_api import api_proc_picc api_proc_picc(spider_com,_key, com_info,k,url,_stat) ner_logger.info(f"api_proc_baidu: {_key}, {com_info}, {k}, {url}") return True # 百度招聘专用API爬取逻辑 def api_proc_baidu(spider_com,_key, com_info,k,url,_stat): # 根据任务类型设置招聘分类 recruitType = "SOCIAL" job_type= "shezhao" projectType = "" if k.startswith("shezhao"): recruitType = "SOCIAL" job_type= "shezhao" elif k.startswith("xiaozhao"): recruitType = "GRADUATE" job_type= "xiaozhao" projectType ="3" elif k.startswith("shixi"): recruitType = "INTERN" job_type= "shixi" if not 'method' in _stat: _stat["method"] = "cp_full" # 获取临时文件目录 key_tmp_dir = spider_com.get_key_dir(_key) total_page = 0 # 循环翻页爬取 for curPage in range(1,100): flag, json_data,totalcount = get_baidu_job_json(url,recruitType,projectType,curPage) if flag: if total_page == 0: total_page = int(totalcount / 20) + 1 # 保存列表页JSON _hash = hashlib.md5(url.encode("utf-8")).hexdigest() tmp_fname = f'{key_tmp_dir}/index_{_hash}_{curPage}.json' with open(tmp_fname,'w',encoding='utf-8') as f: json.dump(json_data,f,ensure_ascii=False,indent=4) # 终止条件 if curPage > total_page: break if curPage > 5 and _stat['method'] != "cp_full": break # 遍历职位，生成详情页 for item in json_data: jobId = item.get("jobId") _fullurl = f"https://talent.baidu.com/jobs/detail/{recruitType}/{jobId}" # 生成文件路径 _hash = hashlib.md5(_fullurl.encode("utf-8")).hexdigest() tmp_file = os.path.join(key_tmp_dir,f"detail_{_hash}.html") tmp_json_file = os.path.join(key_tmp_dir,f"detail_{_hash}.json") # 文件已存在则跳过，仅更新时间 if os.path.exists(tmp_file) and os.path.exists(tmp_json_file): try: current_time = time.time() os.utime(tmp_file, (current_time, current_time)) os.utime(tmp_json_file, (current_time, current_time)) ner_logger.info(f"文件 {tmp_json_file} 的修改时间已更新") except Exception as e: ner_logger.error(f"更新文件时间出错：{str(e)}") continue # 转换数据格式并下载详情页 transform_job_json(item,recruitType,job_type,_key,_fullurl,tmp_file,tmp_json_file) get_baidu_job_html(_fullurl,tmp_file) time.sleep(30) time.sleep(30) return True # 获取代理IP（重试3次） def getProxy(tryTimes = 0): if tryTimes >= 3: return '' params = {"channel":'yupao', "env": 1} proxy = '' try: import requests import time ret = requests.post('http://121.36.63.42:6868/getproxy', params=params) if ret.status_code == 200: rescontent = json.loads(ret.content.decode()) if rescontent.get("code") == 200: proxy = rescontent.get('data').get('proxy') else: proxy = '' if proxy == '': tryTimes = tryTimes + 1 time.sleep(1) return getProxy(tryTimes) ret.close() ner_logger.info(f"获取代理成功:{proxy}") return proxy except: ner_logger.info(f"获取代理失败") time.sleep(1) tryTimes = tryTimes + 1 return getProxy(tryTimes) ``` --- **项目分区导航**：[[00-auto_api|auto_api]] ⬅️ | 01-baidu_data_proc_api | ➡️ [[02-isoftstone_data_proc_api|isoftstone_data_proc_api]]