baidu_data_proc_api.py

baidu_data_proc_api.py — 百度招聘直连

api_proc 总入口所在文件。直接 POST 百度招聘后台 API 拿分页 JSON，带代理池（getProxy 递归重试取可用代理）绕过 IP 限制。模块概览见 auto_api 主篇。

代码

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'<script[^>]*?>.*?', '', 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)

项目分区导航：auto_api ⬅️ | 01-baidu_data_proc_api | ➡️ isoftstone_data_proc_api