isoftstone_data_proc_api.py

isoftstone_data_proc_api.py — 软通动力直连

两套 API 响应格式自动适配（社招 code/data/list / 校招 results/count）；详情页是动态渲染 SPA，用 Playwright 渲染 + ThreadPoolExecutor 避开事件循环冲突。

代码

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
import asyncio
from playwright.sync_api import sync_playwright
import threading
from concurrent.futures import ThreadPoolExecutor

# 请求头配置，模拟浏览器访问软通动力招聘网站
headers = {
    "Accept": "application/json, text/javascript, */*; q=0.01",
    "Accept-Encoding": "gzip, deflate, br, zstd",
    "Accept-Language": "zh-CN,zh;q=0.9",
    "Cache-Control": "no-cache",
    "Connection": "keep-alive",
    "Content-Type": "application/json;charset=UTF-8",
    "Host": "career.isoftstone.com",
    "Origin": "https://career.isoftstone.com",
    "Pragma": "no-cache",
    "Referer": "https://career.isoftstone.com/talent/htmls/shehuizhaopin/index.html",
    "Sec-Ch-Ua": '"Not;A=Brand";v="99", "Google Chrome";v="139", "Chromium";v="139"',
    "Sec-Ch-Ua-Mobile": "?0",
    "Sec-Ch-Ua-Platform": '"Windows"',
    "Sec-Fetch-Dest": "empty",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Site": "same-origin",
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
    "X-Requested-With": "XMLHttpRequest"
}

# 登录Cookie，保持会话状态
cookie_str = ('Hm_lvt_e5e1889ee1cef86df8447e0c983cb5b5=1760151941; '
              'Hm_lvt_c31aaec3450321c4e3d4fd4f7509f181=1760151941; '
              'dreamer-cms-s=c872e558-050d-45d8-bd3d-c97534fc3757')

headers["Cookie"] = cookie_str


# 请求软通动力招聘接口，获取职位JSON数据
def get_isoftstone_job_json(url, recruitType, curPage):
    # 计算分页偏移量
    skip_count = (curPage - 1) * 50
    # 社招/校招使用不同的请求参数
    if recruitType == "1":
        payload = {
            "workCity": "",
            "jobTypeId": "0",
            "keyWord": "",
            "maxcount": 100,
            "page": curPage,
            "recruitType": 1
        }
    elif recruitType == "2":
        payload = {
            "workCity": "",
            "jobTypeId": "0",
            "keyWord": "",
            "skipCount": skip_count,
            "pageCount": 100,
            "recruitType": 2
        }

    # 发送POST请求
    with requests.Session() as s:
        resp = s.post(url, json=payload, headers=headers, timeout=15)
        print("Status:", resp.status_code)
        # 解析返回的JSON数据
        try:
            if resp.status_code == 200:
                json_data = resp.json()
                # 适配 /job/all 接口格式
                if 'code' in json_data:
                    if json_data['code'] == 0:
                        data = json_data['data']['list']
                        total = int(json_data['data']['count'])
                        return True, data, total
                # 适配 /campus/all 接口格式
                elif 'results' in json_data:
                    data = json_data['results']
                    total = int(json_data['count'])
                    return True, data, total
                else:
                    ner_logger.info("Unknown JSON format: %s", json_data)
                    return False, [], 0
            else:
                ner_logger.info("Request failed with status code: %s, response: %s", resp.status_code, resp.text)
                return False, [], 0
        except Exception as e:
            ner_logger.info("JSON decode error: %s, response: %s", str(e), resp.text)
            return False, [], 0


# 使用Playwright获取动态渲染的职位详情页HTML
def get_isoftstone_job_html(url, tmp_file):
    try:
        # 定义浏览器抓取逻辑
        def fetch_page_content():
            with sync_playwright() as p:
                # 无头模式启动浏览器
                browser = p.chromium.launch(headless=True)
                page = browser.new_page()
                
                # 设置请求头
                page.set_extra_http_headers({
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                    "Referer": "https://career.isoftstone.com/talent/htmls/shehuizhaopin/index.html"
                })
                
                # 访问页面并等待网络空闲
                page.goto(url)
                page.wait_for_load_state("networkidle")
                page.wait_for_timeout(5000)
                
                # 获取完整页面内容
                full_text = page.content()
                browser.close()
                return full_text
        
        # 在线程池中运行，避免事件循环冲突
        with ThreadPoolExecutor() as executor:
            future = executor.submit(fetch_page_content)
            full_text = future.result()
        
        # 清除JS脚本内容
        full_text = re.sub(r'<script[^>]*?>.*?', '', full_text, flags=re.DOTALL)
        
        # 写入HTML文件
        with open(tmp_file, "w", encoding="utf-8") as f:
            f.write(full_text)
            
    except Exception as e:
        print(f"请求失败：{e}")
        return None


# 将软通动力原始数据转换为统一标准JSON
def transform_job_json(item, recruitType, job_type, channel, target_url, tmp_file, json_file):
    # 社招字段映射
    field_mapping = {
        "announcement_name": "job_name",
        "publish_time": "public_time",
        "hd_dept": "",
        "hd_loc": "work_city",
        "hd_job_num": "count",
        "hd_job_category": ""
    }
    
    # 校招字段映射
    if "name" in item and "address_detail" in item:
        field_mapping = {
            "announcement_name": "name",
            "publish_time": "publish_date",
            "hd_dept": "",
            "hd_loc": "address_detail",
            "hd_job_num": "",
            "hd_job_category": ""
        }
    
    # 固定公共字段
    fixed_fields = {
        "link": target_url,
        "full_url": target_url,
        "last_url": target_url,
        "file_path": tmp_file,
        "parent_url": "https://career.isoftstone.com/talent/htmls/shehuizhaopin/index.html",
        "channel": channel,
        "job_type": job_type
    }
    
    target_json = {}
    # 字段映射赋值
    for target_field, source_field in field_mapping.items():
        if source_field:
            value = item.get(source_field, "")
            if target_field == "hd_job_num" and isinstance(value, int):
                value = str(value)
            target_json[target_field] = value
        else:
            target_json[target_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(1)


# 软通动力招聘主爬取逻辑
def api_proc_isoftstone(spider_com, _key, com_info, k, url, _stat):
    # 未传入URL则使用默认接口地址
    if not url or url == "":
        url = "https://career.isoftstone.com/job/all"

    # 根据任务类型判断社招/校招
    recruitType = "2"
    job_type = "shezhao"
    
    if k.startswith("shezhao"):
        recruitType = "2"
        job_type = "shezhao"
    elif k.startswith("xiaozhao"):
        recruitType = "1"
        job_type = "xiaozhao"
    
    ner_logger.info("开始处理isoftstone数据, k: %s, url: %s, job_type: %s", k, url, job_type)
    # 获取临时文件目录
    key_tmp_dir = spider_com.get_key_dir(_key)
    ner_logger.info("临时目录: %s", key_tmp_dir)
    
    total_page = 0
    
    # 循环翻页爬取
    for curPage in range(1, 100):
        flag, json_data, totalcount = get_isoftstone_job_json(url, recruitType, curPage)
        if flag:
            # 计算总页数
            if total_page == 0:
                total_page = int(totalcount / 100) + 1
            ner_logger.info("总页数: %s", total_page)
                
            # 终止条件：已到最后一页
            if curPage >= total_page:
                ner_logger.info("已达到总页数，结束分页爬取")
                break
            # 非全量模式只爬5页
            if curPage > 5 and _stat.get('method', '') != "cp_full":
                ner_logger.info("已爬取5页且不是完整模式，结束分页爬取")
                break
                
            # 遍历职位数据
            for i, item in enumerate(json_data):
                job_id = item.get("id")
                # 拼接详情页URL
                if recruitType == "2":
                    _fullurl = f"https://career.isoftstone.com/talent/htmls/shezhaozhiweixiangqing/index.html?id={job_id}&recruitType={recruitType}"
                else:
                    _fullurl = f"https://career.isoftstone.com/talent/htmls/xiaozhaozhiweixiangqing/index.html?id={job_id}&recruitType={recruitType}"
                
                # 生成文件路径
                _hash = hashlib.md5(_fullurl.encode("utf-8")).hexdigest()
                tmp_file = os.path.join(key_tmp_dir, f"detail_{job_id}_{_hash}.html")
                tmp_json_file = os.path.join(key_tmp_dir, f"detail_{job_id}_{_hash}.json")
                
                ner_logger.info("正在处理第 %s 页第 %s 个职位", curPage, i+1)
                
                # 文件已存在则跳过，仅更新时间
                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))
                    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)
                time.sleep(1)
                get_isoftstone_job_html(_fullurl, tmp_file)
                time.sleep(1)
        else:
            ner_logger.info("Failed to fetch data for page %s", curPage)
            if curPage == 1:
                ner_logger.error("第一页数据获取失败，终止处理")
                return False
        time.sleep(1)
        
    ner_logger.info("isoftstone数据处理完成")
    return True

项目分区导航：baidu_data_proc_api ⬅️ | 02-isoftstone_data_proc_api | ➡️ jd_data_proc_api