ai_strategy_request_handler.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330
  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. import uuid
  18. class AiStrategyRequestParse(tornado.web.RequestHandler):
  19. def initialize(self, logger):
  20. self.logger = logger
  21. def post(self):
  22. data = self.request.body
  23. data = str(data, 'utf8')
  24. data = json.loads(data, encoding='utf8')
  25. self.logger.info("*************************************** NEW REQUEST ***************************************")
  26. self.logger.info("raw data from request is %s" % data)
  27. create_group_and_creative_response_data = None
  28. # 1、新增广告组操作
  29. try:
  30. if data["operation_type"] == 1:
  31. inst = ParseAddCampaignOrAddGroupRequest(data,self.logger)
  32. inst.get_advertiser_strategy_info()
  33. inst.write_intelligence_strategy_table()
  34. if inst.campaign_id == "" or inst.campaign_id is None:
  35. if inst.add_campaign() == -1:
  36. raise Exception('广告计划创建失败')
  37. inst.get_group_dict()
  38. create_group_and_creative_res = inst.res_data
  39. inst.update_intelligence_strategy_table()
  40. self.logger.info("create group and creative request info: %s" % create_group_and_creative_res)
  41. request = requests.post(create_group_and_creative_url,
  42. headers=headers,
  43. data=json.JSONEncoder().encode(create_group_and_creative_res)
  44. )
  45. create_group_and_creative_response_data = simplejson.loads(request.text)
  46. self.logger.info("请求返回:%s" % create_group_and_creative_response_data)
  47. self.write(json.dumps({'account_id': inst.account_id, 'ai_strategy_id': inst.ai_strategy_uuid,
  48. 'message': create_group_and_creative_response_data}))
  49. self.flush()
  50. else:
  51. self.write(json.dumps({"message": "operation_type 不支持"}))
  52. self.flush()
  53. except Exception:
  54. self.logger.error(traceback.format_exc())
  55. self.write(json.dumps(traceback.format_exc()))
  56. self.flush()
  57. # 该类解析的是 手动请求的新增广告计划或者新增广告组的数据
  58. class ParseAddCampaignOrAddGroupRequest(object):
  59. def __init__(self, request_data, logger):
  60. self.ai_strategy_uuid = str(uuid.uuid4())
  61. self.request_data = request_data
  62. self.logger = logger
  63. self.video = self.request_data['video']
  64. self.account_id = self.request_data['account_id']
  65. self.campaign_info = self.request_data['campaign_info']
  66. self.campaign_id = self.request_data['campaign_info'].get('campaign_id', None)
  67. self.group_info = self.request_data['group_info']
  68. self.creative_info = self.request_data['creative_info']
  69. self.operation_type = self.request_data['operation_type']
  70. self.advertiser_strategy_id = None
  71. self.advertiser_strategy = {}
  72. self.res_data = {}
  73. def get_advertiser_strategy_info(self):
  74. """
  75. 获取指定账户下客户的策略信息
  76. 为后续的写入库表,校验一致性做准备
  77. :return:
  78. """
  79. sql = """
  80. select *
  81. from ctop_ai_kuaishou_advertiser_strategy
  82. where account_id = %d
  83. and status = 1
  84. limit 1
  85. """ % self.account_id
  86. advertiser_strategy_df = pd.read_sql(sql, engine)
  87. self.advertiser_strategy_id = advertiser_strategy_df['id'].values[0]
  88. self.advertiser_strategy = advertiser_strategy_df.T.to_dict()[0]
  89. def write_intelligence_strategy_table(self):
  90. """
  91. 请求信息写入 ctop_ai_kuaishou_intelligence_strategy 表中
  92. :return:
  93. """
  94. intelligence_strategy_dict = \
  95. {
  96. 'ai_strategy_uuid': self.ai_strategy_uuid,
  97. 'advertiser_strategy_id': self.advertiser_strategy_id,
  98. 'account_id': self.account_id,
  99. 'ai_strategy_receive_content': str(self.request_data),
  100. 'ai_strategy_remark': self.request_data['ai_strategy_remark'],
  101. 'create_time': datetime.datetime.now()
  102. }
  103. df = pd.DataFrame.from_dict(intelligence_strategy_dict, orient='index').T
  104. df.to_sql(name="ctop_ai_kuaishou_intelligence_strategy", con=engine, if_exists='append', index=False)
  105. def update_intelligence_strategy_table(self):
  106. """
  107. 拼装之后的json, 更新到 ai_strategy_request_content
  108. :return:
  109. """
  110. db_con = pymysql.connect(host=host,
  111. port=port,
  112. user=user,
  113. password=passwd,
  114. database=db,
  115. charset=charset)
  116. cursor = db_con.cursor()
  117. cursor.execute("""UPDATE ctop_ai_kuaishou_intelligence_strategy
  118. SET ai_strategy_request_content= %s
  119. WHERE ai_strategy_uuid = %s""", (str(self.res_data), self.ai_strategy_uuid))
  120. db_con.commit()
  121. db_con.close()
  122. # 如果手动请求的数据结构中,"campaign" 中的 campaign_id 为空则调用该新增计划方法
  123. def add_campaign(self):
  124. """
  125. 新增广告计划
  126. :return:
  127. """
  128. # 1、校验请求信息中 有没有 与客户策略信息 中存在冲突的地方 -- 计划层级只判断 campaign_type
  129. if self.campaign_info['campaign_type'] != self.advertiser_strategy['campaign_type']:
  130. self.logger.info("客户策略的campaign_type: %d 和 AI策略的campaign_type:%d 不一致!" %
  131. (self.advertiser_strategy['campaign_type'],
  132. self.campaign_info['campaign_type']))
  133. return -1
  134. create_campaign_req_data = {
  135. 'account_id': self.account_id,
  136. 'campaign_name': self.campaign_info['campaign_name'],
  137. 'type': self.campaign_info['campaign_type']
  138. }
  139. request = requests.post(url=create_campaign_url,
  140. headers=headers,
  141. data=json.JSONEncoder().encode(create_campaign_req_data))
  142. res_data = simplejson.loads(request.text)
  143. # res_data = {
  144. # "code": 0,
  145. # "data": {
  146. # "account_id": 23212,
  147. # "campaign_id": 30956133,
  148. # "campaign_create_time": "2021-01-13 13:50:16"
  149. # },
  150. # "message": "SUCCESS"}
  151. if res_data['code'] == 0:
  152. self.campaign_id = res_data['data']['campaign_id']
  153. campaign_info_to_db = {
  154. 'campaign_uuid': str(uuid.uuid4()),
  155. 'account_id': self.account_id,
  156. 'ai_strategy_uuid': self.ai_strategy_uuid,
  157. 'campaign_name': self.campaign_info['campaign_name'],
  158. 'campaign_id': self.campaign_id,
  159. 'campaign_type': self.campaign_info['campaign_type'],
  160. 'operation_type': self.operation_type,
  161. 'campaign_create_time': res_data['data']['campaign_create_time'],
  162. 'create_time': datetime.datetime.now()
  163. }
  164. df = pd.DataFrame.from_dict(campaign_info_to_db, orient='index').T
  165. df.to_sql(name="ctop_ai_kuaishou_campaign_level_operation_record", con=engine, if_exists='append', index=False)
  166. else:
  167. self.logger.info("广告计划创建失败:%s" % res_data['message'])
  168. return -1
  169. def get_group_dict(self):
  170. """
  171. 拼接组和创意层级参数
  172. 写入数据库
  173. 返回 table_id
  174. :return:
  175. """
  176. # 1、TODO 校验请求信息中 有没有 与客户策略信息 中存在冲突的地方 -- 暂时不实现
  177. # 客户策略 和 请求信息 都有值 才会进行校验比对操作 -- 不得高于,不得不等于,不得超过集合
  178. # 因此分为3类 字段
  179. # is_equal_cols = ['template_id','bid_type']
  180. # is_less_than_cols = []
  181. # is_contain_cols = []
  182. #
  183. # for col in is_equal_cols:
  184. # if (self.advertiser_strategy[col] is not None) and \
  185. # (self.group_info.get(col, None) is not None) and \
  186. # (self.group_info.get(col) != self.advertiser_strategy[col]):
  187. # return -1
  188. #
  189. # for col in is_less_than_cols:
  190. # if (self.advertiser_strategy[col] is not None) and \
  191. # (self.group_info.get(col, None) is not None) and \
  192. # (self.group_info.get(col) > self.advertiser_strategy[col]):
  193. # return -1
  194. #
  195. # for col in is_contain_cols:
  196. # if (self.advertiser_strategy[col] is not None) and \
  197. # (self.group_info.get(col, None) is not None) and \
  198. # (self.group_info.get(col) > self.advertiser_strategy[col]):
  199. # return -1
  200. drop_cols = ['id', 'account_id', 'status', 'campaign_type',
  201. 'site_id', 'click_track_url', 'impression_url', 'ad_photo_played_t3s_url',
  202. 'actionbar_click_url',
  203. 'create_time', 'effective_time', 'expiry_time']
  204. cols_to_list = ['app_store', 'scene_id', 'region', 'district_ids', 'ages_range', 'device_brand',
  205. 'device_price', 'business_interest', 'fans_star', 'interest_video', 'app_interest',
  206. 'app_ids', 'population', 'exclude_population', 'paid_audience', 'behavior_interest']
  207. basic_group_info = self.advertiser_strategy.copy()
  208. for col in drop_cols:
  209. del basic_group_info[col]
  210. # 3、从请求的信息中更新计划层级信息
  211. if len(self.group_info) > 0:
  212. for key, value in self.group_info.items():
  213. basic_group_info.update({key: value})
  214. self.res_data['group_list'] = []
  215. group_cnt = 1
  216. for video in self.video:
  217. single_group_info = {'group_uuid': str(uuid.uuid4())}
  218. single_group_info.update(basic_group_info)
  219. single_group_info['unit_name'] = self.group_info['unit_name'] + '-' + str(group_cnt)
  220. # 组信息写入数据表中
  221. single_group_info_to_db = single_group_info.copy()
  222. single_group_info_to_db['ai_strategy_uuid'] = self.ai_strategy_uuid
  223. single_group_info_to_db['account_id'] = self.account_id
  224. single_group_info_to_db['campaign_id'] = self.campaign_id
  225. single_group_info_to_db['create_time'] = datetime.datetime.now()
  226. single_group_info_to_db['operation_type'] = self.operation_type
  227. df = pd.DataFrame.from_dict(single_group_info_to_db, orient='index').T
  228. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  229. df.to_sql(name="ctop_ai_kuaishou_unit_level_operation_record",
  230. con=engine,
  231. if_exists='append',
  232. index=False)
  233. # 发送的请求,这几个字段需要转为 [] 数组
  234. for col in cols_to_list:
  235. single_group_info[col] = eval(single_group_info[col]) if single_group_info[col] is not None else None
  236. single_group_info['creative_list'] = []
  237. group_cnt += 1
  238. creative_cnt = 1
  239. for image in video['imageList']:
  240. single_creative_info = {
  241. 'creative_uuid': str(uuid.uuid4()),
  242. 'is_sticky': 0,
  243. 'creative_name': self.creative_info['creative_name'] + '_' + str(creative_cnt),
  244. 'photo_id': video['photo_id'],
  245. 'image_md5': image, # TODO 需要和蒙蒙确认字段名称
  246. 'image_tokens': self.creative_info.get('image_tokens'),
  247. 'action_bar_text': self.creative_info['action_bar_text'],
  248. 'description': self.creative_info['description'],
  249. 'short_slogan': self.creative_info.get('short_slogan'),
  250. 'sticker_title': self.creative_info.get('sticker_title'),
  251. 'overlay_type': self.creative_info.get('overlay_type'),
  252. 'expose_tag': self.creative_info.get('expose_tag'),
  253. 'new_expose_tag': self.creative_info.get('new_expose_tag'),
  254. 'site_id': self.creative_info.get('site_id'),
  255. 'click_track_url': self.advertiser_strategy.get('click_track_url'),
  256. 'impression_url': self.advertiser_strategy.get('impression_url'),
  257. 'ad_photo_played_t3s_url': self.advertiser_strategy.get('ad_photo_played_t3s_url'),
  258. 'actionbar_click_url': self.advertiser_strategy.get('actionbar_click_url'),
  259. 'creative_category': self.creative_info.get('creative_category'),
  260. 'creative_tag': self.creative_info.get('creative_tag')}
  261. single_group_info['creative_list'].append(single_creative_info)
  262. # 创意层级写入数据表中
  263. single_creative_info_to_db = single_creative_info.copy()
  264. single_creative_info_to_db['ai_strategy_uuid'] = self.ai_strategy_uuid
  265. single_creative_info_to_db['account_id'] = self.account_id
  266. single_creative_info_to_db['campaign_id'] = self.campaign_id
  267. single_creative_info_to_db['create_time'] = datetime.datetime.now()
  268. single_creative_info_to_db['operation_type'] = self.operation_type
  269. cols_to_str = ['creative_tag']
  270. for col in cols_to_str:
  271. single_creative_info_to_db[col] = str(single_creative_info_to_db[col]) \
  272. if single_creative_info_to_db[col] is not None else None
  273. df = pd.DataFrame.from_dict(single_creative_info_to_db, orient='index').T
  274. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  275. df.to_sql(name="ctop_ai_kuaishou_creative_level_operation_record",
  276. con=engine,
  277. if_exists='append',
  278. index=False)
  279. creative_cnt += 1
  280. # 组和创意信息添加到最终返回数据 group_list 中
  281. self.res_data['group_list'].append(single_group_info)
  282. self.res_data['account_id'] = self.account_id
  283. self.res_data['campaign_id'] = self.campaign_id
  284. # TODO 修改操作
  285. class ParseModifyRequest(object):
  286. def __init__(self):
  287. pass
  288. # TODO 关停操作
  289. class ParseShutDownRequest(object):
  290. def __init__(self):
  291. pass