quanzhi_api.py

quanzhi_api.py — 云端数据同步接口

上传公告/职位数据到测试/正式环境；check_cloud 风控拦截（301 过期 / 302 黑名单 / 303 全文重复 / 304 链接重复）；fix_upload_cloud 上传前对日期做二次校验修复。

代码

# -*- 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), ""

项目分区导航：openai4o_api_0 ⬅️ | 08-quanzhi_api | ➡️ qwen_api