env_regulation_api/app/services/tencent_nlp.py

import json

from loguru import logger
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
    TencentCloudSDKException,
)
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.nlp.v20190408 import models, nlp_client

from app.core.config import settings


def get_tencent_nlp_client() -> nlp_client.NlpClient:
    try:
        cred = credential.Credential(
            settings.TENCENT_SECRET_ID_V1, settings.TENCENT_SECRET_KEY_V1
        )
        http_profile = HttpProfile()
        http_profile.reqMethod = "GET"
        http_profile.endpoint = settings.TENCENT_NLP_ENDPOINT

        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
        client = nlp_client.NlpClient(cred, "ap-guangzhou", client_profile)

        return client
    except TencentCloudSDKException as e:
        logger.error(e)


class TencentNLP:
    def __init__(self):
        cred = credential.Credential(
            settings.TENCENT_SECRET_ID_V1, settings.TENCENT_SECRET_KEY_V1
        )
        http_profile = HttpProfile()
        http_profile.reqMethod = "GET"
        http_profile.endpoint = settings.TENCENT_NLP_ENDPOINT

        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
        client = nlp_client.NlpClient(cred, "ap-guangzhou", client_profile)
        self.client = client

    async def get_lexical_analysis_result(self, text: str) -> tuple[list, list]:
        req = models.LexicalAnalysisRequest()
        params = {"Text": text}
        req.from_json_string(json.dumps(params))
        resp = self.client.LexicalAnalysis(req)

        return resp.PosTokens, resp.NerTokens

    async def get_auto_summarization_result(self, text: str) -> str:
        req = models.AutoSummarizationRequest()
        params = {"Text": text}
        req.from_json_string(json.dumps(params))
        resp = self.client.AutoSummarization(req)

        return resp.Summary

    async def get_text_similarity_result(
            self, src_text: str, target_text: list[str]
    ) -> list:
        req = models.TextSimilarityRequest()
        params = {"SrcText": src_text, "TargetText": target_text}
        req.from_json_string(json.dumps(params))
        resp = self.client.TextSimilarity(req)

        return resp.Similarity

    async def get_dependency(self, text: str) -> list:
        req = models.DependencyParsingRequest()
        params = {"Text": text}
        req.from_json_string(json.dumps(params))
        resp = self.client.DependencyParsing(req)

        return resp.DpTokens

    async def get_word_similarity(self, src_word: str, target: str) -> float:
        try:
            req = models.WordSimilarityRequest()
            params = {"SrcWord": src_word, "TargetWord": target}
            req.from_json_string(json.dumps(params))
            resp = self.client.WordSimilarity(req)
        except TencentCloudSDKException:
            return 0

        return resp.Similarity