ai_strategy_request_handler.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164
  1. from typing import Optional, Awaitable
  2. import pandas as pd
  3. import numpy as np
  4. import datetime
  5. import pymysql
  6. from sqlalchemy import create_engine
  7. import simplejson
  8. import requests
  9. import tornado.web
  10. import pandas as pd
  11. import json
  12. import numpy as np
  13. import datetime
  14. from sqlalchemy import create_engine
  15. from db_config import *
  16. import traceback
  17. class AiStrategyRequestParse(tornado.web.RequestHandler):
  18. def initialize(self, logger):
  19. self.logger = logger
  20. def post(self):
  21. data = self.request.body
  22. data = str(data, 'utf8')
  23. self.logger.info("*************************************** NEW REQUEST ***************************************")
  24. self.logger.info("raw data from request is %s" % data)
  25. # 1、新增广告组操作
  26. if data["operation_type"] == "add":
  27. inst = ParseModifyRequest(data)
  28. # 该类解析的是 手动请求的新增广告计划或者新增广告组的数据
  29. class ParseAddCampaignOrAddGroupRequest(object):
  30. def __init__(self, request_data):
  31. self.request_data = request_data
  32. self.video = self.request_data['video']
  33. self.account_id = self.request_data['account_id']
  34. self.campaign_info = self.request_data['campaign_info']
  35. self.group_info = self.request_data['group_info']
  36. self.operation_type = self.request_data['operation_type']
  37. self.advertiser_strategy_id = None
  38. self.ai_strategy_id = None
  39. self.advertiser_strategy = {}
  40. self.res_data = {}
  41. def get_advertiser_strategy_info(self):
  42. """
  43. 获取指定账户下客户的策略信息
  44. 为后续的写入库表,校验一致性做准备
  45. :return:
  46. """
  47. sql = """
  48. select *
  49. from ctop_ai_kuaishou_advertiser_strategy
  50. where account_id = %d
  51. and status = 1
  52. """ % self.account_id
  53. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  54. advertiser_strategy_df = pd.read_sql(sql, engine)
  55. self.advertiser_strategy = advertiser_strategy_df.T.to_dict()[0]
  56. self.advertiser_strategy_id = self.advertiser_strategy['id']
  57. def write_intelligence_strategy_table(self):
  58. """
  59. 请求信息写入 ctop_ai_kuaishou_intelligence_strategy 表中
  60. :return:
  61. """
  62. intelligence_strategy_dict = \
  63. {'advertiser_strategy_id': self.advertiser_strategy_id,
  64. 'account_id': self.account_id,
  65. 'ai_strategy_request_content': self.request_data,
  66. 'create_time': datetime.datetime.now()}
  67. df = pd.DataFrame.from_dict(intelligence_strategy_dict, orient='index').T
  68. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  69. df.to_sql(name="ctop_ai_kuaishou_intelligence_strategy", con=engine, if_exists='append', index=False)
  70. # 获取新插入的 intelligence_strategy 的唯一标识ID
  71. # TODO 这个方法的缺点是不适合高并发。如果同时插入的时候返回的值可能不准确。
  72. sql = """
  73. select max(id) from ctop_ai_kuaishou_intelligence_strategy where account_id = %d
  74. """ % self.account_id
  75. df = pd.read_sql(sql,engine)
  76. self.ai_strategy_id = df['id'].values[0]
  77. # 如果手动请求的数据结构中,"campaign" 中的 campaign_id 为空则调用该新增计划方法
  78. def add_campaign(self):
  79. """
  80. 新增广告计划
  81. :return:
  82. """
  83. # 1、校验请求信息中 有没有 与客户策略信息 中存在冲突的地方 -- 计划层级只判断 campaign_type
  84. if self.campaign_info['campaign_type'] != self.advertiser_strategy['campaign_type']:
  85. # TODO 写入日志信息
  86. return -1
  87. # 3、从请求的信息中更新计划层级信息
  88. # if len(self.request_data['campaign']) > 0:
  89. # for key, value in self.request_data['campaign'].items():
  90. # self.campaign_dict.update({key: value})
  91. create_campaign_req_data = {'account_id': self.account_id,
  92. 'campaign_name': self.campaign_info['campaign_name'],
  93. 'type': self.campaign_info['campaign_type']}
  94. url = "http://192.168.1.8:8080/jeecg-boot//jeecg-boot/kuaishou/create/campaignCreate"
  95. request = requests.post(url, create_campaign_req_data)
  96. res_data = simplejson.loads(request.text)
  97. if res_data['code'] == 0:
  98. campaign_info_to_db = {'account_id': self.account_id,
  99. 'ai_strategy_id': 1,
  100. 'campaign_name': self.campaign_info['campaign_name'],
  101. 'campaign_id': res_data['data']['campaign_id'],
  102. 'campaign_type': self.campaign_info['campaign_type'],
  103. 'operation_type': self.operation_type,
  104. 'campaign_create_time': res_data['data']['campaign_create_time'],
  105. 'create_time': datetime.datetime.now()}
  106. df = pd.DataFrame.from_dict(campaign_info_to_db, orient='index').T
  107. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  108. df.to_sql(name="ctop_ai_kuaishou_campaign_level_operation_record", con=engine, if_exists='append', index=False)
  109. else:
  110. # TODO 写入计划创建失败的原因
  111. return -1
  112. # 写入数据库 ctop_ai_kuaishou_campaign_level_operation_record
  113. # 4、将计划层级的信息添加到 返回数据中
  114. self.res_data['campaign'] = self.campaign_dict
  115. def get_group_dict(self):
  116. """
  117. 拼接组和创意层级参数
  118. :return:
  119. """
  120. def update_intelligence_strategy_table(self):
  121. """
  122. 拼接完成的信息更新到 ctop_ai_kuaishou_intelligence_strategy 表中
  123. :return:
  124. """
  125. pass
  126. # TODO 修改操作
  127. class ParseModifyRequest(object):
  128. def __init__(self):
  129. pass
  130. # TODO 关停操作
  131. class ParseShutDownRequest(object):
  132. def __init__(self):
  133. pass