--- title: "08-quanzhi_api" created: 2026-04-02 tags: - 项目 aliases: - quanzhi_api --- # quanzhi_api.py ### `quanzhi_api.py` — 云端数据同步接口 上传公告/职位数据到测试/正式环境；`check_cloud` 风控拦截（301 过期 / 302 黑名单 / 303 全文重复 / 304 链接重复）；`fix_upload_cloud` 上传前对日期做二次校验修复。 ## 代码 ```python # -*- coding: utf-8 -*- """ @Desc : 公告/招聘数据 上传云端接口工具 @Date : 2025年1月17日10:13:06 @Author : leo 功能： 1. 上传公告数据到测试/正式环境 2. 上传大公司职位数据到云端 3. 云端重复公告校验 4. 自动格式化日期、修复上传参数 """ import json import requests import sys sys.path.append('../') # 添加上级目录，导入自定义工具 from utils import ner_logger # 日志工具 from utils_date import fix_data_format, fix_data_format_large # 日期格式化工具 # ======================== 1. 云端接口地址配置 ======================== # lc=1：不验证JobLink；ck=1：不校验重复 # 测试环境 - 添加公告 debug_cloud_url = "http://192.168.1.17:6868/ann_add?lc=1&ck=1" # 正式环境 - 添加公告 prod_cloud_url = "http://121.36.63.42:6868/ann_add?lc=1&ck=1" # 测试环境 - 检查重复公告 debug_check_url = "http://192.168.1.17:6868/checkrepeat" # 正式环境 - 检查重复公告 prod_check_url = "http://121.36.63.42:6868/checkrepeat" # 大公司职位 - 测试环境上传接口 debug_job_cloud_url = "http://192.168.1.17:6868/jobparse" # 大公司职位 - 正式环境上传接口 prod_job_cloud_url = "http://121.36.63.42:6868/jobparse" # ======================== 2. 请求头配置 ======================== # 请求头：指定传输数据格式为JSON headers = { "Content-Type": "application/json" } # ======================== 3. 接口返回码说明 ======================== """ 返回值说明： 成功: {'code': 200, 'msg': 'ok', 'data': {'jobId': "67890c78f663ba81f1c854b9"}} 失败： {'code':201,'msg':'参数错误',{"JobTitle":'no'}} # JobTitle 没值 {'code':204,'msg':'公司不存在'} {'code': 300, 'msg': '公告公司名称校验失败'} # 需要记录，人工修正公司名后重新入库 {'code': 301, 'msg': '公告已过期'} {'code': 302, 'msg': '公告职位名称包含黑名单'} {'code': 303, 'msg': '公告全文重复或太相似'} {'code':304, 'msg': '公告Joblink连接重复'} {'code':305, 'msg': '公告被人工修改过，不做更新处理'} """ # ======================== 4. 公告上传主函数 ======================== def upload_cloud(json_file_path, _dist='dev'): """ 上传【公告数据】到云端 :param json_file_path: 本地JSON文件路径 :param _dist: 环境标识，dev=测试环境，prod=正式环境 :return: (是否成功, 响应内容, 响应码) """ # 选择上传地址：默认测试环境，prod切换正式环境 cloud_url = debug_cloud_url if _dist == 'prod': cloud_url = prod_cloud_url ner_logger.info(f"开始上传云端 - {json_file_path} / {cloud_url}") # 读取本地JSON文件 with open(json_file_path, 'r', encoding='utf-8') as f: data = json.load(f) try: # 数据预处理：格式化日期、修正字段 fix_upload_cloud(data) # 发送POST请求上传JSON数据 response = requests.post(cloud_url, headers=headers, json=data) # HTTP请求成功 if response.status_code == 200: response_data = response.text json_data = json.loads(response_data) # 200：上传成功 if str(json_data['code']) in ['200']: ner_logger.info(f"上传云端成功 - {json_data}") return True, response_data, "200" # 301~305：业务层面成功（过期/重复/黑名单等，不算失败） elif str(json_data['code']) in ['301', '302', '303', '304', '305']: ner_logger.info(f"上传云端成功（业务状态） - {json_data}") return True, response_data, str(json_data['code']) # 其他码：上传失败 else: ner_logger.info(f"上传云端失败 - {json_data}") return False, response_data, "" # 捕获上传过程异常 except Exception as e: ner_logger.info(f"上传云端失败 - {json_file_path}，异常：{str(e)}") return False, str(e), "" # 异常兜底 return False, "未知错误（upload_cloud第57行附近）", "" # ======================== 5. 重复公告校验函数 ======================== def check_cloud(_durl, _companyname, _fulltext, _jie, _dist='prod'): """ 云端校验：检查公告是否重复 :param _durl: 公告链接（必填） :param _companyname: 公司名称（必填） :param _fulltext: 公告全文（必填） :param _jie: 毕业年限最大值（必填） :param _dist: 环境 :return: (是否通过校验, 错误信息) """ # 构造校验参数 data = { "l": _durl, "c": _companyname, "t": _fulltext, "g": _jie, } # 选择校验接口地址 cloud_url = debug_check_url if _dist == 'prod': cloud_url = prod_check_url try: # 发送校验请求 response = requests.post(cloud_url, data=data) if response.status_code == 200: response_data = response.json() # code=200 表示无重复，可以上传 if "code" in response_data and response_data["code"] == 200: ner_logger.info(f"云端校验成功，可以上传 - {response_data}") return True, "" else: ner_logger.info(f"云端校验失败（重复/相似） - {response_data}") return False, response_data # 网络异常等，默认允许上传 except Exception as e: ner_logger.info(f"云端校验异常，自动跳过 - {e}") return True, "" # ======================== 6. 上传前数据修复函数 ======================== def fix_upload_cloud(data): """ 上传前自动修复数据：日期格式化、字段补全 支持两种数据结构：ann（公告）、cjob（企业职位） """ json_data = None # 判断是公告数据还是职位数据 if 'ann' in data: json_data = data['ann'] elif 'cjob' in data: json_data = data['cjob'] # 职位数据：使用公司全称作为简称 json_data['ComShortName'] = json_data['ComName'] # 格式化过期日期 if 'CutDate' in json_data: json_data['CutDate'] = fix_data_format_large(json_data['CutDate']) else: # 无有效数据，直接返回 return # ======================== 日期统一格式化 ======================== # 发布开始日期 if 'OnlineStartDate' in json_data: json_data['OnlineStartDate'] = fix_data_format(json_data['OnlineStartDate']) # 发布结束日期 if 'OnlineEndDate' in json_data: json_data['OnlineEndDate'] = fix_data_format(json_data['OnlineEndDate']) # 发布时间 if 'PublishTime' in json_data: json_data['PublishTime'] = fix_data_format(json_data['PublishTime']) # ======================== 7. 大公司职位上传函数 ======================== def upload_cloud_job(json_file_path, _dist='dev', _retry='0'): """ 上传【大公司职位数据】到云端 :param json_file_path: 职位JSON文件路径 :param _dist: 环境 dev/prod :param _retry: 是否重入 1=重入 0=正常 :return: (是否成功, 响应内容, HTTP状态码) """ # 重入标识转换 retry = 0 if _retry == '1': retry = 1 # 选择职位上传接口地址 cloud_url = debug_job_cloud_url if _dist == 'prod': cloud_url = prod_job_cloud_url ner_logger.info(f"开始上传职位到云端 - {json_file_path} / {cloud_url}") try: # 读取本地职位JSON文件 with open(json_file_path, 'r', encoding='utf-8') as f: data = json.load(f) # 数据预处理：日期格式化、字段修正 fix_upload_cloud(data) # ======================== 构造上传表单参数 ======================== # comFrom：固定渠道来源 100000 # fileName：使用FileId作为文件名 # content：整个JSON数据转字符串上传 # retry：1=重入（允许更新），0=不重入 form_data = { "comFrom": "100000", "fileName": data['cjob']['FileId'], "content": json.dumps(data), "retry": retry, } ner_logger.info(f"上传参数 - {form_data}") print("请求接口地址：", cloud_url) # 发送职位上传请求 response = requests.post(cloud_url, data=form_data) # HTTP请求成功 if response.status_code == 200: response_data = response.text json_data = json.loads(response_data) # code=200 上传成功 if str(json_data['code']) in ['200']: return True, response.text, response.status_code # 上传失败 ner_logger.info(f"职位上传云端失败 - {json_file_path}, 响应：{response.text}") return False, f"未知错误（upload_cloud_job第165行）{response.text}", "" # 捕获所有异常并打印堆栈 except Exception as e: import traceback traceback.print_exc() ner_logger.info(f"职位上传异常 - {json_file_path}, 异常：{str(e)}") return False, str(e), "" ``` --- **项目分区导航**：[[07-openai4o_api_0|openai4o_api_0]] ⬅️ | 08-quanzhi_api | ➡️ [[09-qwen_api|qwen_api]]