cjob_model.py

cjob_model.py — 大公司职位详情页解析

独立入口（不经过 ann_model）：detail_selector 精准定位职位正文 → Header 区域提取职位类别/地点/人数 → DocType 规范化（shixi/xiaozhao/shezhao）→ 带 system prompt 的大模型解析。

代码

# -*- coding: utf-8 -*-
"""
企业职位详情解析模块
功能：从职位HTML页面 → 提取正文 → 大模型结构化解析 → 生成标准职位JSON
适用：企业招聘官网、校招/社招/实习岗位统一结构化解析
"""

import json
import sys
sys.path.append('../')
import re

from bs4 import BeautifulSoup

from utils import ner_logger, getMD5Str, get_local_ip, QZ_VERISON
from utils_html import clean_text
from utils_date import get_current_time_string, get_current_data

from parsegpt.template import get_template_cjob, get_context_cjob
from api.openai4o_api import call_gpt_system
from api.doubao_api import call_gpt as doubao_call_gpt
from api.qwen_api import call_gpt as qwen_call_gpt
from api.doubao_api_new import call_gpt as new_call_gpt

from parsegpt.ann_model import fix_diploma_data_map
from parsegpt.ann_md import fix_html_div
from parsegpt.html_to_text import Html2txt

# ====================== 主函数：职位结构化解析 ======================
def parse_cjob(spider_data, _model_file, _info, com_info, _expired_file, _hfile, _stat):
    """
    职位解析主入口
    流程：读取HTML → 提取正文 → 大模型解析 → 字段补全 → 输出JSON模型
    :return: (状态, 错误信息)
    """
    _ann_dict = {}
    _title = _info['announcement_name']
    _full_text = ""

    # 读取职位HTML文件内容
    with open(_hfile, "r", encoding="utf-8") as f:
        _html = f.read()
        # 清洗HTML，提取职位纯文本
        _full_text = get_cjob_html_content(spider_data, com_info, _html)

    # 文本清洗（去空格、特殊符号）
    _text = clean_text(_full_text)

    # 文本过短，直接判定解析失败
    if len(_text) < 100:
        return "", f"通过大模型获取公司职位信息Error:文本长度不足{len(_text)}"

    try:
        # 从页面提取固定字段（职位类别、地点、人数等）
        _a_map = get_hd_element(_html, com_info)

        # 构造大模型Prompt并调用新版豆包API
        _t_text = get_template_cjob(_text)
        _T_text = get_context_cjob(_text)
        (_ok_flag, json_str, tokens) = new_call_gpt(_t_text, _T_text, True)

        if not _ok_flag:
            return "", f"通过大模型获取公司职位信息Error:{_ok_flag}\n{json_str}"

        ner_logger.info(f"通过大模型获取公司职位信息json:{json_str}")

        # 解析大模型返回的JSON
        json_data = json.loads(json_str, strict=False)

        # 补全公共字段（公司、标题、链接、薪资、地点等）
        set_other_info(com_info, _info, json_data, _text, _a_map)

        # 职位描述 + 任职要求过短，视为无效职位
        if len(json_data['JobDescribe']) + len(json_data['Jobreq']) < 30:
            return "", f"职位信息的描述太少:{_hfile}\n{json_str}"

        # 学历标准化（本科/硕士/博士等）
        fix_diploma_data_map(json_data)

        # 组装职位主体
        _ann_dict['cjob'] = json_data

    # 大模型返回非JSON格式时捕获异常
    except json.JSONDecodeError as e:
        import traceback
        traceback.print_exc()
        ner_logger.error(f"Error:{json_str}")
        return "", f"通过大模型获取公司职位信息Error:\n{e}"

    # 附加信息（来源、任务信息）
    _info['mdfile_path'] = _model_file
    _ann_dict['other'] = _info

    # 校验结构完整性
    if len(_ann_dict) != 2:
        return "", f"通过大模型获取公司职位信息Error:节点不足{len(_ann_dict)}"

    # 写入最终模型JSON文件
    with open(_model_file, 'w', encoding='utf-8') as fw:
        json.dump(_ann_dict, fw, ensure_ascii=False, indent=4)
        ner_logger.info(f"处理大公司职位生成模型文件成功,{_model_file}")

    print(tokens)
    return "ok", ""

# ====================== 字段补全：公司/标题/链接/时间/类型 ======================
def set_other_info(com_info, _info, json_data, _text, _a_map):
    """
    补全职位结构化字段
    优先级：页面提取 > 配置信息 > 默认值
    """
    # 内容唯一标识
    json_data['FileId'] = getMD5Str(_text)
    # 职位链接
    json_data['JobLink'] = _info['full_url']
    # 职位名称（清洗前缀）
    json_data['JobTitle'] = fix_job_name(_info['announcement_name'])

    # 公司信息
    json_data['ComName'] = com_info['com_name']
    json_data['ComShortName'] = com_info['com_webname']
    json_data['ComLogo'] = com_info['com_logo']
    json_data['DocType'] = _info['job_type']

    # 学历标准化
    fix_diploma_data_map(json_data)

    # 从列表页传递的字段
    if 'hd_dept' in _info and len(_info['hd_dept']) > 1:
        json_data['JobDept'] = _info['hd_dept']
    if 'hd_loc' in _info and len(_info['hd_loc']) > 1:
        json_data['WorkPlace'] = _info['hd_loc']
    if 'publish_time' in _info and len(_info['publish_time']) > 1:
        json_data['PublishTime'] = _info['publish_time']
    if 'hd_job_num' in _info and len(_info['hd_job_num']) > 0:
        json_data['JobNum'] = _info['hd_job_num']
    elif "若干" in "hd_job_num" or "不限" in "hd_job_num":
        json_data['JobNum'] = _a_map["hd_job_num"]

    if 'hd_job_category' in _info and len(_info['hd_job_category']) > 0:
        json_data['JobCategory'] = _info['hd_job_category']
    if 'hd_salary' in _info and len(_info['hd_salary']) > 0:
        json_data['Salary'] = _info['hd_salary']
    if 'hd_hopeworktype' in _info and len(_info['hd_hopeworktype']) > 0:
        json_data['HopeWorkType'] = _info['hd_hopeworktype']

    # 职位类型覆盖：实习/校招/社招
    if 'hd_hopeworktype' in _info and _info['hd_hopeworktype'] == '实习':
        json_data['DocType'] = 'shixi'
    if 'hd_hopeworktype' in _info and _info['hd_hopeworktype'] == '校招':
        json_data['DocType'] = 'xiaozhao'
    if 'hd_hopeworktype' in _info and _info['hd_hopeworktype'] == '社招':
        json_data['DocType'] = 'shezhao'

    # 从详情页HTML提取的覆盖字段
    if 'hd_job_category' in _a_map and len(_a_map['hd_job_category']) > 0:
        json_data['JobCategory'] = _a_map['hd_job_category']
    if 'hd_loc' in _a_map and len(_a_map['hd_loc']) > 0:
        json_data['WorkPlace'] = _a_map['hd_loc']
    if 'hd_job_num' in _a_map and len(_a_map['hd_job_num']) > 0:
        json_data['JobNum'] = _a_map['hd_job_num']
    if 'hd_publish_time' in _a_map and len(_a_map['hd_publish_time']) > 1:
        json_data['PublishTime'] = _a_map['hd_publish_time']

    # 特殊地区替换为公司默认地区
    if json_data['WorkPlace'] in ['全部地区', '全国各地', '']:
        json_data['WorkPlace'] = com_info['hd_all_location']

    # 发布时间为空则自动补当前日期
    if json_data['PublishTime'] == '':
        json_data['PublishTime'] = get_current_data()

    # 职位类别清洗
    try:
        json_data['JobCategory'] = fix_job_category(json_data.get('JobCategory', ""))
    except Exception as e:
        json_data['JobCategory'] = ""

    # 空值安全处理
    if not 'JobDescribe' in json_data:
        json_data['JobDescribe'] = ''
    if not 'Jobreq' in json_data:
        json_data['Jobreq'] = ''
    if not 'JobNum' in json_data:
        json_data['JobNum'] = ''

    # 服务信息（IP/版本/处理时间）
    _info['server_ip'] = get_local_ip()
    _info['qz_version'] = QZ_VERISON
    _info['process_time'] = get_current_time_string()

# ====================== 从HTML提取职位正文（支持配置选择器） ======================
def get_cjob_html_content(spider_data, com_info, htmltext):
    """
    根据配置的CSS选择器，提取职位正文
    支持：class选择器、正则匹配class
    """
    soup = BeautifulSoup(htmltext, 'html.parser')

    # 清理垃圾标签
    fix_html_div(spider_data, soup, com_info, {})

    # 按配置的选择器提取正文
    class_names = com_info.get("detail_selector")
    for class_name in class_names.split("|"):
        if not class_name:
            continue
        cc = class_name.split(".")
        div_element = soup.find(cc[0], class_=cc[1])
        if div_element:
            html_content = str(div_element)
            return Html2txt().clean_html(html_content)

    # 正则匹配class（备用方案）
    class_name = com_info.get("detail_selector_re")
    if class_name:
        pattern = re.compile(f'div.{class_name}')
        ner_logger.info(f"尝试使用正则查找:{pattern}")
        matched_divs = soup.find_all('div', class_=pattern)
        if matched_divs:
            html_content = str(matched_divs[0])
            return Html2txt().clean_html(html_content)

    # 无匹配则返回整个页面文本
    return Html2txt().clean_html(htmltext)

# ====================== 提取页面固定信息：人数/地点/发布时间/类别 ======================
def get_hd_element(htmltext, com_info):
    """
    根据不同网站模板，提取固定信息
    支持 0001/0002/0003 三种模板
    返回：职位类别、工作地点、招聘人数、发布时间
    """
    soup = BeautifulSoup(htmltext, 'html.parser')
    _map = {}

    # 模板0001：提取职位类别
    if 'detail_hd' in com_info and com_info['detail_hd'] == "0001":
        div_span = soup.find('div', class_='pos-detail-hd__titBar')
        if div_span:
            tit_span = soup.find('span', class_='tit')
            if tit_span:
                label_span = soup.find('span', class_='label')
                if label_span:
                    _text = label_span.get_text()
                    for _t in ['职位类别：', '职位类别']:
                        _text = _text.replace(_t, "")
                    _map['hd_job_category'] = _text

    # 模板0002：提取工作地点
    if 'detail_hd' in com_info and com_info['detail_hd'] == "0002":
        div_span = soup.find('div', class_='pos-detail-hd__infoBar')
        if div_span:
            loc_span = soup.find('span', class_='item-overflow')
            if loc_span:
                _map['hd_loc'] = loc_span.get_text()

    # 模板0003：提取发布时间、招聘人数
    if 'detail_hd' in com_info and com_info['detail_hd'] == "0003":
        div_span = soup.find('div', class_='pos-detail-hd__infoBar')
        if div_span:
            _text = div_span.get_text()
            # 匹配日期 yyyy-mm-dd
            match = re.search(r'\d{4}-\d{2}-\d{2}', _text)
            if match:
                _map['hd_publish_time'] = match.group()
            # 匹配人数：如 5人、若干人
            match = re.search(r'\d{1,3}人', _text)
            if match:
                _map['hd_job_num'] = match.group()
            elif '若干' in _text:
                _map['hd_job_num'] = '若干'

    ner_logger.info(f"获取hd_map成功,{_map}")
    return _map

# ====================== 工具函数：职位名称清洗 ======================
def fix_job_name(_text):
    return _text.replace("职位名称：", "").replace("热招", "")

# ====================== 工具函数：职位类别清洗 ======================
def fix_job_category(_text):
    if _text in ["热招", '社招', '实习', '校招']:
        return ""
    # 过滤 A类/B类 这种格式
    if re.search(r'[A-Z]类', _text):
        return ""
    return _text

项目分区导航：ann_model_job ⬅️ | 05-cjob_model | ➡️ html_to_text