record.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141
  1. import time
  2. from uuid import uuid4
  3. from applications.log import logging
  4. class Record(object):
  5. """
  6. 搜索接口处理逻辑
  7. """
  8. def __init__(self, params, mysql_client, config):
  9. self.flow_pool_level = None
  10. self.content_id = None
  11. self.account_name = None
  12. self.contents = None
  13. self.title = None
  14. self.gh_id = None
  15. self.params = params
  16. self.mysql_client = mysql_client
  17. self.article_match_video_table = config.article_match_video_table
  18. self.article_text_table = config.article_text_table
  19. self.trace_id = "search-{}-{}".format(str(uuid4()), str(int(time.time())))
  20. def check_params(self):
  21. """
  22. 检查请求params
  23. :return:
  24. """
  25. try:
  26. self.gh_id = self.params['ghId']
  27. self.title = self.params['title'].split("@@")[-1].replace("'", "")
  28. self.contents = self.params['content'].replace("'", "")
  29. self.account_name = self.params['accountName'].replace("'", "")
  30. self.content_id = self.params['articleId']
  31. self.flow_pool_level = self.params['flowPoolLevelTag']
  32. logging(
  33. code="1001",
  34. info="搜索视频内容接口请求成功, 参数校验成功",
  35. port="title_to_search",
  36. trace_id=self.trace_id,
  37. data=self.params
  38. )
  39. return None
  40. except Exception as e:
  41. result = {
  42. "status": "fail",
  43. "code": 1,
  44. "message": str(e),
  45. "info": "params check error"
  46. }
  47. logging(
  48. code="4001",
  49. info="搜索视频内容接口请求成功, 参数校验失败",
  50. port="title_to_search",
  51. trace_id=self.trace_id,
  52. data=self.params
  53. )
  54. return result
  55. async def input_into_article_match_video_table(self):
  56. """
  57. 把数据插入待处理队列
  58. :return:
  59. """
  60. request_time = int(time.time())
  61. insert_sql = f"""
  62. INSERT INTO {self.article_match_video_table}
  63. (trace_id, content_id, flow_pool_level, gh_id, account_name, request_timestamp)
  64. VALUES
  65. (%s, %s, %s, %s, %s, %s);
  66. """
  67. await self.mysql_client.async_insert(
  68. sql=insert_sql,
  69. params=(
  70. self.trace_id,
  71. self.content_id,
  72. self.flow_pool_level,
  73. self.gh_id,
  74. self.account_name,
  75. request_time
  76. )
  77. )
  78. logging(
  79. code="1001",
  80. info="请求文章存储到 long_articles_match_videos中",
  81. function="Record",
  82. trace_id=self.trace_id
  83. )
  84. async def input_into_article_text_table(self):
  85. """
  86. :return:
  87. """
  88. insert_sql = f"""
  89. INSERT INTO {self.article_text_table} (content_id, article_title, article_text)
  90. values (%s, %s, %s);
  91. """
  92. try:
  93. await self.mysql_client.async_insert(
  94. sql=insert_sql,
  95. params=(
  96. self.content_id,
  97. self.title,
  98. self.contents
  99. )
  100. )
  101. logging(
  102. code="1002",
  103. info="请求文章存储到 long_articles_text中",
  104. function="Record",
  105. trace_id=self.trace_id
  106. )
  107. except Exception as e:
  108. logging(
  109. code="1002",
  110. info="请求文章 id 已经存储在 long_article_text中 {}".format(e),
  111. function="Record",
  112. trace_id=self.trace_id
  113. )
  114. async def deal(self):
  115. """
  116. deal
  117. :return:
  118. """
  119. params_error = self.check_params()
  120. if params_error:
  121. return params_error
  122. else:
  123. # 记录数据
  124. await self.input_into_article_match_video_table()
  125. await self.input_into_article_text_table()
  126. res = {
  127. "status": "success input to article queue",
  128. "code": 0,
  129. "traceId": self.trace_id
  130. }
  131. return res