--- title: "05-cjob_model" created: 2026-04-02 tags: - 项目 aliases: - cjob_model --- # cjob_model.py ### `cjob_model.py` — 大公司职位详情页解析 独立入口（不经过 [[03-ann_model|ann_model]]）：`detail_selector` 精准定位职位正文 → Header 区域提取职位类别/地点/人数 → DocType 规范化（shixi/xiaozhao/shezhao）→ 带 system prompt 的大模型解析。 ## 代码 ```python # -*- 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 ``` --- **项目分区导航**：[[04-ann_model_job|ann_model_job]] ⬅️ | 05-cjob_model | ➡️ [[06-html_to_text|html_to_text]]