data_quality_server/tables/ai/__init__.py

# coding:utf-8
from a2s.a2s_client import a2s_execute
from a2s.tools import json_serialize, json_deserialize
from loguru import logger
from docs.config import FieldServer, RecognitionServer, ProductServer


def fields_ner(text):
    """
    抽取文本字段
    :param text:
    :return:
    """
    try:
        if len(text) > 2000: return {}
        request_body = {"text": text}
        for r in range(FieldServer["retry_times"]):
            bytes_data = json_serialize(request_body)
            result = a2s_execute(FieldServer["a2s_ip"], FieldServer["topic"], FieldServer["timeout"], bytes_data)
            if result is None:
                continue
            response_json = json_deserialize(result)
            if response_json.get("ucode", 0) != 200:
                return {}
            return response_json
    except Exception as e:
        logger.info(str(e))
        return {}


def org_ner(text):
    """
    抽取物品
    :param text:
    :return:
    """
    try:
        companies = []
        if len(text) > 2000:
            return companies
        request_body = {"text": text}
        bytes_data = json_serialize(request_body)
        for r in range(RecognitionServer["retry_times"]):
            result = a2s_execute(RecognitionServer["a2s_ip"], RecognitionServer["topic"], RecognitionServer["timeout"],
                                 bytes_data)
            if result is None:
                continue
            response_json = json_deserialize(result)
            if response_json.get("ucode", 0) != 200:
                return companies
            output = response_json.get("output", [])
            for text in output:
                text_type = text.get("type", "")
                if text_type == "ORG":
                    companies.append(text.get("span", ""))
            return companies
    except Exception as e:
        logger.info(str(e))
        return {}


def product_detail_server(text):
    """
    物品
    :param text:
    :return:
    """
    try:
        ret = a2s_execute(ProductServer["a2s_ip"], ProductServer["topic"], ProductServer["timeout"],
                          bytes_data=json_serialize({"text": text}))
        model_dict = json_deserialize(ret)
        model_dict = model_dict if isinstance(model_dict, dict) else {}
        return model_dict
    except Exception as e:
        print(e)
        return {}


if __name__ == '__main__':
    data = "型号：清廉长沙水350ml"
    # result = org_ner(data)
    result = product_detail_server(data)
    print(result)