--- title: "04-ann_model_job" created: 2026-04-02 tags: - 项目 aliases: - ann_model_job --- # ann_model_job.py ### `ann_model_job.py` — 公告内单职位解析 由 [[03-ann_model|ann_model]] 分支调用:HTML 转纯文本 → Qwen 大模型提取 → 校验质量 → 与公告元数据(公司名、链接、申请方式、联系人)合并 → 写入标准职位 JSON。 ## 代码 ```python # -*- coding: utf-8 -*- """ 单个职位解析模块 功能:从招聘详情HTML中,使用大模型提取职位描述、任职要求等结构化信息 输出:标准职位JSON,用于入库 """ import json import sys sys.path.append('../') import re from utils import ner_logger, getMD5Str, get_local_ip, QZ_VERISON # 大模型调用接口 from api.doubao_api import call_gpt as doubao_call_gpt from api.qwen_api import call_gpt as qwen_call_gpt # 职位解析Prompt模板 from parsegpt.template import get_template_cjob_annotation # 学历标准化工具 from utils_resume import fix_diploma # HTML纯文本提取工具 from parsegpt.html_to_text import Html2txt # HTML解析 from bs4 import BeautifulSoup # ====================== 核心函数:解析单个招聘职位 ====================== def parse_cjob(_hfile, ann_json_data): """ 从HTML详情页提取单个职位的结构化信息 :param _hfile: 本地HTML文件路径 :param ann_json_data: 上层已解析的公告信息(公司、标题、链接等) :return: (状态, 职位JSON数据) """ _ann_dict = {} try: # 1. 从HTML文件提取纯文本内容 _text = get_cjob_html_content(_hfile) # 2. 构造大模型Prompt,调用千问API解析职位信息 _t_text = get_template_cjob_annotation(_text) (_ok_flag, json_str) = qwen_call_gpt(_t_text, True) if not _ok_flag: return "", f"通过大模型获取公告里面的单个职位职位信息Error:{_ok_flag}\n{json_str}" ner_logger.info(f"通过大模型获取公告里面的单个职位信息json:{json_str}") # 3. 解析大模型返回的JSON json_data = json.loads(json_str, strict=False) # 4. 过滤:职位描述+任职要求过短,视为无效职位 if len(json_data['JobDescribe']) + len(json_data['Jobreq']) < 30: return "", f"职位信息的描述太少:{_hfile}\n{json_str}" # 5. 学历信息标准化(本科/硕士/博士等) fix_diploma_data_map(json_data) # ====================== 复制上层公告的公共字段 ====================== json_data['FileId'] = ann_json_data['FileId'] json_data['JobLink'] = ann_json_data['JobLink'] json_data['DocType'] = 'xiaozhao' # 文档类型:校招 json_data['ComLogo'] = "" # 公司Logo json_data['ComName'] = ann_json_data['ComName'] # 公司名称 json_data['ComShortName'] = ann_json_data['ComName'] json_data['NoticeToJob'] = 1 # 标记为招聘公告 json_data['WxName'] = ann_json_data['WeixinName'] # 公众号名称 json_data['ApplyTypeLink'] = ann_json_data['ApplyTypeLink'] # 网申链接 json_data['GraduationYear'] = ann_json_data['GraduationYear'] # 招聘届别 json_data['AnnouncementLabel'] = ann_json_data['AnnouncementLabel'] json_data['JobTitle'] = ann_json_data['JobTitle'] # 职位标题 json_data['EmailSubject'] = ann_json_data['EmailSubject'] json_data['ApplyTypeText'] = ann_json_data['ApplyTypeText'] json_data['GraduationTimeRequirement'] = ann_json_data['GraduationTimeRequirement'] json_data['mdfile'] = ann_json_data['mdfile'] json_data['HopeWorkType'] = ann_json_data['HopeWorkType'] json_data['PublishTime'] = ann_json_data['PublishTime'] # ====================== 复制联系人信息 ====================== if 'ApplyContacts' in ann_json_data: acontacts = ann_json_data['ApplyContacts'] if len(acontacts) > 0: # 联系人姓名(不含*才有效) if 'Name' in acontacts[0] and "*" not in acontacts[0]['Name']: json_data['ContactPerson'] = acontacts[0]['Name'] # 联系电话 if 'Mobile' in acontacts[0] and "*" not in acontacts[0]['Mobile']: json_data['Phone'] = acontacts[0]['Mobile'] # 邮箱 if 'Email' in acontacts[0] and "*" not in acontacts[0]['Email']: json_data['Email'] = acontacts[0]['Email'] # 如果没有联系人邮箱,从公告公共邮箱补充 if 'Email' in json_data and json_data['Email'] == "" and 'ApplyTypeEmail' in ann_json_data and ann_json_data['ApplyTypeEmail'] != "": json_data['Email'] = ann_json_data['ApplyTypeEmail'] # ====================== 组装最终输出结构 ====================== _ann_dict['cjob'] = json_data # 职位主体信息 # 清理冗余字段(二维码) _n_ann_json_data = ann_json_data if 'ApplyTypeQrcode' in _n_ann_json_data: _n_ann_json_data.pop('ApplyTypeQrcode') _ann_dict['other'] = _n_ann_json_data # 固定前端展示用的字段名映射 _ann_dict['cjob_o_field'] = { 'RecruitProcess': "招聘流程", 'Attention': '注意事项', 'WelfareInfo': '薪酬福利', 'JobDevelopment': '岗位发展', 'ApplyTypeText': '应聘方式' } # JSON解析异常处理 except json.JSONDecodeError as e: import traceback traceback.print_exc() ner_logger.error(f"Error:{json_str}") return "", f"通过大模型获取公告里面的单个职位信息Error:\n{e}" # 返回成功状态 + 职位结构化数据 return "OK", _ann_dict # ====================== 学历标准化 ====================== def fix_diploma_data_map(item): """ 统一学历格式:如“本科及以上”“硕士”等 """ if 'Degree' in item: need_fix = item['Degree'] item['Degree'] = fix_diploma(need_fix) # ====================== 从HTML提取职位纯文本 ====================== def get_cjob_html_content(_htmlfile): """ 读取HTML文件,清洗标签,返回干净的职位文本 """ with open(_htmlfile, "r", encoding="utf-8", errors='ignore') as f1: htmltext = f1.read() # 使用工具类清洗HTML,返回纯文本 text = Html2txt().clean_html(htmltext) return text ``` --- **项目分区导航**:[[03-ann_model|ann_model]] ⬅️ | 04-ann_model_job | ➡️ [[05-cjob_model|cjob_model]]