tencent_nlp.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. import json
  2. from typing import Dict, List, Tuple
  3. from loguru import logger
  4. from tencentcloud.common import credential
  5. from tencentcloud.common.profile.client_profile import ClientProfile
  6. from tencentcloud.common.profile.http_profile import HttpProfile
  7. from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
  8. TencentCloudSDKException,
  9. )
  10. from tencentcloud.nlp.v20190408 import models, nlp_client
  11. from app.core.config import settings
  12. def get_tencent_nlp_client() -> nlp_client.NlpClient:
  13. try:
  14. cred = credential.Credential(
  15. settings.TENCENT_SECRET_ID_V1, settings.TENCENT_SECRET_KEY_V1
  16. )
  17. http_profile = HttpProfile()
  18. http_profile.reqMethod = "GET"
  19. http_profile.endpoint = settings.TENCENT_NLP_ENDPOINT
  20. client_profile = ClientProfile()
  21. client_profile.httpProfile = http_profile
  22. client = nlp_client.NlpClient(cred, "ap-guangzhou", client_profile)
  23. return client
  24. except TencentCloudSDKException as e:
  25. logger.error(e)
  26. class TencentNLP:
  27. def __init__(self):
  28. cred = credential.Credential(
  29. settings.TENCENT_SECRET_ID_V1, settings.TENCENT_SECRET_KEY_V1
  30. )
  31. http_profile = HttpProfile()
  32. http_profile.reqMethod = "GET"
  33. http_profile.endpoint = settings.TENCENT_NLP_ENDPOINT
  34. client_profile = ClientProfile()
  35. client_profile.httpProfile = http_profile
  36. client = nlp_client.NlpClient(cred, "ap-guangzhou", client_profile)
  37. self.client = client
  38. async def get_lexical_analysis_result(self, text: str) -> Tuple[List, List]:
  39. req = models.LexicalAnalysisRequest()
  40. params = {"Text": text}
  41. req.from_json_string(json.dumps(params))
  42. resp = self.client.LexicalAnalysis(req)
  43. return resp.PosTokens, resp.NerTokens
  44. async def get_auto_summarization_result(self, text: str) -> str:
  45. req = models.AutoSummarizationRequest()
  46. params = {"Text": text}
  47. req.from_json_string(json.dumps(params))
  48. resp = self.client.AutoSummarization(req)
  49. return resp.Summary
  50. async def get_text_similarity_result(
  51. self, src_text: str, target_text: List[str]
  52. ) -> List:
  53. req = models.TextSimilarityRequest()
  54. params = {"SrcText": src_text, "TargetText": target_text}
  55. req.from_json_string(json.dumps(params))
  56. resp = self.client.TextSimilarity(req)
  57. return resp.Similarity
  58. async def get_dependency(self, text: str) -> List:
  59. req = models.DependencyParsingRequest()
  60. params = {"Text": text}
  61. req.from_json_string(json.dumps(params))
  62. resp = self.client.DependencyParsing(req)
  63. return resp.DpTokens
  64. async def get_word_similarity(self, src_word: str, target: str) -> float:
  65. try:
  66. req = models.WordSimilarityRequest()
  67. params = {"SrcWord": src_word, "TargetWord": target}
  68. req.from_json_string(json.dumps(params))
  69. resp = self.client.WordSimilarity(req)
  70. except TencentCloudSDKException:
  71. return 0
  72. return resp.Similarity