from typing import Optional, Awaitable import pandas as pd import numpy as np import datetime import pymysql from sqlalchemy import create_engine import simplejson import requests import tornado.web import pandas as pd import json import numpy as np import datetime from sqlalchemy import create_engine from db_config import * import traceback class AiStrategyRequestParse(tornado.web.RequestHandler): def initialize(self, logger): self.logger = logger def post(self): data = self.request.body data = str(data, 'utf8') self.logger.info("*************************************** NEW REQUEST ***************************************") self.logger.info("raw data from request is %s" % data) # 1、新增广告组操作 if data["operation_type"] == "add": inst = ParseModifyRequest(data) # 该类解析的是 手动请求的新增广告计划或者新增广告组的数据 class ParseAddCampaignOrAddGroupRequest(object): def __init__(self, request_data): self.request_data = request_data self.video = self.request_data['video'] self.account_id = self.request_data['account_id'] self.campaign_info = self.request_data['campaign_info'] self.group_info = self.request_data['group_info'] self.operation_type = self.request_data['operation_type'] self.advertiser_strategy_id = None self.ai_strategy_id = None self.advertiser_strategy = {} self.res_data = {} def get_advertiser_strategy_info(self): """ 获取指定账户下客户的策略信息 为后续的写入库表,校验一致性做准备 :return: """ sql = """ select * from ctop_ai_kuaishou_advertiser_strategy where account_id = %d and status = 1 """ % self.account_id engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) advertiser_strategy_df = pd.read_sql(sql, engine) self.advertiser_strategy = advertiser_strategy_df.T.to_dict()[0] self.advertiser_strategy_id = self.advertiser_strategy['id'] def write_intelligence_strategy_table(self): """ 请求信息写入 ctop_ai_kuaishou_intelligence_strategy 表中 :return: """ intelligence_strategy_dict = \ {'advertiser_strategy_id': self.advertiser_strategy_id, 'account_id': self.account_id, 'ai_strategy_request_content': self.request_data, 'create_time': datetime.datetime.now()} df = pd.DataFrame.from_dict(intelligence_strategy_dict, orient='index').T engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) df.to_sql(name="ctop_ai_kuaishou_intelligence_strategy", con=engine, if_exists='append', index=False) # 获取新插入的 intelligence_strategy 的唯一标识ID # TODO 这个方法的缺点是不适合高并发。如果同时插入的时候返回的值可能不准确。 sql = """ select max(id) from ctop_ai_kuaishou_intelligence_strategy where account_id = %d """ % self.account_id df = pd.read_sql(sql,engine) self.ai_strategy_id = df['id'].values[0] # 如果手动请求的数据结构中,"campaign" 中的 campaign_id 为空则调用该新增计划方法 def add_campaign(self): """ 新增广告计划 :return: """ # 1、校验请求信息中 有没有 与客户策略信息 中存在冲突的地方 -- 计划层级只判断 campaign_type if self.campaign_info['campaign_type'] != self.advertiser_strategy['campaign_type']: # TODO 写入日志信息 return -1 # 3、从请求的信息中更新计划层级信息 # if len(self.request_data['campaign']) > 0: # for key, value in self.request_data['campaign'].items(): # self.campaign_dict.update({key: value}) create_campaign_req_data = {'account_id': self.account_id, 'campaign_name': self.campaign_info['campaign_name'], 'type': self.campaign_info['campaign_type']} url = "http://192.168.1.8:8080/jeecg-boot//jeecg-boot/kuaishou/create/campaignCreate" request = requests.post(url, create_campaign_req_data) res_data = simplejson.loads(request.text) if res_data['code'] == 0: campaign_info_to_db = {'account_id': self.account_id, 'ai_strategy_id': 1, 'campaign_name': self.campaign_info['campaign_name'], 'campaign_id': res_data['data']['campaign_id'], 'campaign_type': self.campaign_info['campaign_type'], 'operation_type': self.operation_type, 'campaign_create_time': res_data['data']['campaign_create_time'], 'create_time': datetime.datetime.now()} df = pd.DataFrame.from_dict(campaign_info_to_db, orient='index').T engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) df.to_sql(name="ctop_ai_kuaishou_campaign_level_operation_record", con=engine, if_exists='append', index=False) else: # TODO 写入计划创建失败的原因 return -1 # 写入数据库 ctop_ai_kuaishou_campaign_level_operation_record # 4、将计划层级的信息添加到 返回数据中 self.res_data['campaign'] = self.campaign_dict def get_group_dict(self): """ 拼接组和创意层级参数 :return: """ def update_intelligence_strategy_table(self): """ 拼接完成的信息更新到 ctop_ai_kuaishou_intelligence_strategy 表中 :return: """ pass # TODO 修改操作 class ParseModifyRequest(object): def __init__(self): pass # TODO 关停操作 class ParseShutDownRequest(object): def __init__(self): pass