ann_model_job.py
ann_model_job.py — 公告内单职位解析
由 ann_model 分支调用：HTML 转纯文本 → Qwen 大模型提取 → 校验质量 → 与公告元数据（公司名、链接、申请方式、联系人）合并 → 写入标准职位 JSON。
代码
# -*- 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
项目分区导航：ann_model ⬅️ | 04-ann_model_job | ➡️ cjob_model
💬 评论