import datetime import json import uuid from logging.handlers import TimedRotatingFileHandler from concurrent_log import ConcurrentTimedRotatingFileHandler import pandas as pd import pymysql import requests from utils.DataBaseConfig import * from utils.UrlConfig import * import logging log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%m/%d/%Y %I:%M:%S %p') # log_handler = TimedRotatingFileHandler("logs/ai_strategy_request.log", # when="midnight", backupCount=100) log_handler = ConcurrentTimedRotatingFileHandler("logs/ai_strategy_request.log", when="midnight", backupCount=100) log_handler.setFormatter(log_formatter) logger = logging.getLogger('ai_strategy_request_logger') logger.addHandler(log_handler) logger.setLevel(logging.DEBUG) print('id of ai_strategy_request_logger %s' % id(logger)) def ai_strategy_request_parse(data): """ 发送创建组和创意的请求 :param data: :return: """ print(data) if data["operation_type"] == 1: inst = ParseAddCampaignOrAddGroupRequest(data) logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s" % (inst.account_id, os.getpid(), os.getppid())) # 1 获取客户策略信息 inst.get_advertiser_strategy_info() # 2 写入智能策略信息 inst.write_intelligence_strategy_table() # 3 判断 campaign_id 是否为空,为空则发送创建计划的请求 if inst.campaign_id == "" or inst.campaign_id is None: if inst.add_campaign() == -1: return {'code': -2, 'message': '广告计划创建失败!'} # 4 unit_type "4-自定义" or "7-程序化创意2.0" ,调用对应的拼装参数函数 if data['group_info'].get('unit_type', 4) == 4: inst.assemble_group_and_creative_params() if data['group_info'].get('unit_type', 4) == 7: inst.assemble_group_and_programme_creative_params() # 5 发送的请求信息更新到数据库表 inst.update_intelligence_strategy_table(request_content=inst.res_data, message=None) # 6 发送创建组和创意的请求 request = requests.post(create_group_and_creative_url, headers=headers, data=json.JSONEncoder().encode(inst.res_data)) response_data = json.loads(request.text) # 7 接受的返回信息更新到数据库表 inst.update_intelligence_strategy_table(message=response_data) # 8 写入日志信息 logger.info("策略uuid = %s 的请求返回信息:%s" % (inst.ai_strategy_uuid, response_data)) return {'code': 0, 'message': response_data} else: return {'code': -1, 'message': '暂时不支持非新增的操作请求!'} # 该类解析的是 手动请求的新增广告计划或者新增广告组的数据 class ParseAddCampaignOrAddGroupRequest(object): def __init__(self, request_data): self.ai_strategy_uuid = str(uuid.uuid4()) self.request_data = request_data self.video = self.request_data['video']['data'] self.account_id = self.request_data['account_id'] self.campaign_info = self.request_data['campaign_info'] self.campaign_id = self.request_data['campaign_info'].get('campaign_id', None) self.group_info = self.request_data.get('group_info', None) self.creative_info = self.request_data.get('creative_info',None) self.operation_type = self.request_data['operation_type'] self.advertiser_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 limit 1 """ % self.account_id advertiser_strategy_df = pd.read_sql(sql, engine) self.advertiser_strategy_id = int(advertiser_strategy_df['id'].values[0]) self.advertiser_strategy = advertiser_strategy_df.T.to_dict()[0] def write_intelligence_strategy_table(self): """ 请求信息写入 ctop_ai_kuaishou_intelligence_strategy 表中 :return: """ intelligence_strategy_dict = \ { 'ai_strategy_uuid': self.ai_strategy_uuid, 'advertiser_strategy_id': self.advertiser_strategy_id, 'account_id': self.account_id, 'ai_strategy_receive_content': str(self.request_data), 'ai_strategy_remark': self.request_data['ai_strategy_remark'], 'create_time': datetime.datetime.now() } df = pd.DataFrame.from_dict(intelligence_strategy_dict, orient='index').T df.to_sql(name="ctop_ai_kuaishou_intelligence_strategy", con=engine, if_exists='append', index=False) def update_intelligence_strategy_table(self, request_content=None, message=None): """ 更新 intelligence_strategy_table :return: """ db_con = pymysql.connect(host=host, port=port, user=user, password=passwd, database=db, charset=charset) cursor = db_con.cursor() if request_content: cursor.execute("""UPDATE ctop_ai_kuaishou_intelligence_strategy SET ai_strategy_request_content= %s WHERE ai_strategy_uuid = %s""", (str(request_content), self.ai_strategy_uuid)) if message: cursor.execute("""UPDATE ctop_ai_kuaishou_intelligence_strategy SET message = %s WHERE ai_strategy_uuid = %s""", (str(message), self.ai_strategy_uuid)) db_con.commit() db_con.close() def add_campaign(self): """ 新增广告计划, 创建成功之后写入计划层级操作表中 :return: """ create_campaign_req_data = \ {'account_id': self.account_id, 'campaign_name': self.campaign_info['campaign_name'], 'type': self.advertiser_strategy['campaign_type']} request = requests.post(url=create_campaign_url, headers=headers, data=json.JSONEncoder().encode(create_campaign_req_data)) res_data = json.loads(request.text) # res_data = { # "code": 0, # "data": { # "account_id": 23212, # "campaign_id": 30956133, # "campaign_create_time": "2021-01-13 13:50:16" # }, # "message": "SUCCESS"} campaign_info_to_db = { 'campaign_uuid': str(uuid.uuid4()), 'account_id': self.account_id, 'ai_strategy_uuid': self.ai_strategy_uuid, 'campaign_name': self.campaign_info['campaign_name'], 'campaign_type': self.advertiser_strategy['campaign_type'], 'operation_type': self.operation_type, 'create_time': datetime.datetime.now() } res = 0 logger.info("the res of create campaign is %s" % res_data) if res_data['code'] == 0: self.campaign_id = res_data['data'].get('campaign_id', None) campaign_info_to_db['campaign_id'] = self.campaign_id campaign_info_to_db['campaign_create_time'] = res_data['data'].get('campaign_create_time', None) campaign_info_to_db['status'] = res_data.get('code', None) campaign_info_to_db['message'] = res_data.get('message', None) logger.info("广告计划创建成功,campaign_id = %s, ai_strategy_uuid = %s" % (self.campaign_id, self.ai_strategy_uuid)) else: campaign_info_to_db['message'] = res_data['message'] campaign_info_to_db['status'] = res_data['code'] logger.error("广告计划创建失败:%s, ai_strategy_uuid = %s" % (res_data['message'], self.ai_strategy_uuid)) res = -1 # 写入计划层级的操作表 df = pd.DataFrame.from_dict(campaign_info_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_campaign_level_operation_record", con=engine, if_exists='append', index=False) return res def get_group_params(self): group_params = self.advertiser_strategy.copy() # 1 从客户策略表中,剔除掉不是组层级的参数(客户基本信息、计划层级信息、创意层级信息) drop_cols = ['id', 'account_id', 'status', 'campaign_type', 'campaign_name', 'creative_name', 'action_bar_text', 'description', 'short_slogan', 'sticker_title', 'overlay_type', 'expose_tag', 'new_expose_tag', 'site_id', 'click_track_url', 'impression_url', 'ad_photo_played_t3s_url', 'actionbar_click_url', 'creative_category', 'creative_tag', 'image_cnt', 'create_time', 'effective_time', 'expiry_time', 'timer_task_status'] for col in drop_cols: if col in group_params.keys(): del group_params[col] # 2 从请求的信息中更新组层级信息, 如组的名称、测试定向、测试出价等, update客户策略信息 if (self.group_info is not None) and (len(self.group_info) > 0): for key, value in self.group_info.items(): group_params.update({key: value}) # 3 smart_cover 和 asset_mining 需要将0/1 转化为 False/True group_params['smart_cover'] = True if group_params.get('smart_cover', 0) == 1 else False group_params['asset_mining'] = True if group_params.get('asset_mining', 0) == 1 else False # 4 拼装json的组参数 group_params_to_request = group_params.copy() # 4-1 拼接发送请求json时, 以下字段需要转为 list 类型 cols_to_list = ['app_store', 'scene_id', 'region', 'district_ids', 'ages_range', 'device_brand', 'device_price', 'business_interest', 'fans_star', 'interest_video', 'app_interest', 'app_ids', 'population', 'exclude_population', 'paid_audience', 'behavior_interest'] for col in cols_to_list: group_params_to_request[col] = eval(group_params_to_request[col]) \ if ((group_params_to_request[col] is not None) and (group_params_to_request[col] != '')) else None # 4-2 发送请求,需要将 group_name 改为 unit_name group_params_to_request['unit_name'] = group_params_to_request.pop('group_name') # 5 写入数据库的组参数 group_params_to_db = group_params.copy() group_params_to_db['ai_strategy_uuid'] = self.ai_strategy_uuid group_params_to_db['account_id'] = self.account_id group_params_to_db['campaign_id'] = self.campaign_id group_params_to_db['operation_type'] = self.operation_type group_params_to_db['status'] = 1 group_params_to_db['message'] = None return group_params_to_request, group_params_to_db def get_creative_params(self): # 1 从客户策略表中获取 创意层级基本信息 creative_cols = ['creative_name', 'action_bar_text', 'description', 'short_slogan', 'sticker_title', 'overlay_type', 'expose_tag', 'new_expose_tag', 'site_id', 'click_track_url', 'impression_url', 'ad_photo_played_t3s_url', 'actionbar_click_url', 'creative_category', 'creative_tag'] creative_params = {} for col in creative_cols: creative_params[col] = self.advertiser_strategy.get(col) # 2 从请求的信息中更新创意层级的信息,如广告语 if (self.creative_info is not None) and (len(self.creative_info) > 0): for key, value in self.creative_info.items(): creative_params.update({key: value}) # 3 拼装json的 创意参数 creative_params_to_request = creative_params.copy() # 3-1 拼接发送请求json时, 以下字段需要转为 list 类型 cols_to_list = ['creative_tag'] for col in cols_to_list: creative_params_to_request[col] = eval(creative_params_to_request[col]) \ if ((creative_params_to_request[col] is not None) and (creative_params_to_request[col] != '')) else None # 4 写入数据库的创意参数 creative_params_to_db = creative_params.copy() creative_params_to_db['ai_strategy_uuid'] = self.ai_strategy_uuid creative_params_to_db['account_id'] = self.account_id creative_params_to_db['campaign_id'] = self.campaign_id creative_params_to_db['operation_type'] = self.operation_type creative_params_to_db['status'] = 1 creative_params_to_db['message'] = None return creative_params_to_request, creative_params_to_db def get_programme_creative_params(self): # 1 从客户策略表中获取 创意层级基本信息 creative_cols = ['creative_name', 'action_bar_text', 'description', 'site_id', 'click_track_url', 'actionbar_click_url', 'creative_category', 'creative_tag'] creative_params = {} for col in creative_cols: creative_params[col] = self.advertiser_strategy.get(col) # 2 从请求的信息中更新创意层级的信息,如广告语,创意名称等 if (self.creative_info is not None) and (len(self.creative_info) > 0): for key, value in self.creative_info.items(): creative_params.update({key: value}) # 3 程序化创意部分字段发生改变,需要重命名 # creative_name --> package_name(程序化创意名称) # action_bar_text --> action_bar(行动号召按钮) # description --> captions(作品广告语) string[] # click_track_url --> click_url(点击监测链接) creative_params['package_name'] = creative_params.pop('creative_name') creative_params['action_bar'] = creative_params.pop('action_bar_text') description = creative_params.pop('description') creative_params['captions'] = str([description]) creative_params['click_url'] = creative_params.pop('click_track_url') # 4 拼装json的 创意参数 creative_params_to_request = creative_params.copy() # 4-1 拼接发送请求json时, 以下字段需要转为 list 类型 cols_to_list = ['creative_tag', 'captions'] for col in cols_to_list: creative_params_to_request[col] = eval(creative_params_to_request[col]) \ if ((creative_params_to_request[col] is not None) and (creative_params_to_request[col] != '')) else None # 5 写入数据库的创意参数 creative_params_to_db = creative_params.copy() creative_params_to_db['ai_strategy_uuid'] = self.ai_strategy_uuid creative_params_to_db['account_id'] = self.account_id creative_params_to_db['campaign_id'] = self.campaign_id creative_params_to_db['operation_type'] = self.operation_type creative_params_to_db['status'] = 1 creative_params_to_db['message'] = None return creative_params_to_request, creative_params_to_db def assemble_group_and_creative_params(self): """ 拼接组和创意层级参数 写入数据库 :return: """ # 获取组和创意的基本参数 group_params_to_request, group_params_to_db = self.get_group_params() creative_params_to_request, creative_params_to_db = self.get_creative_params() self.res_data['group_list'] = [] group_cnt = 1 for video in self.video: group_uuid = str(uuid.uuid4()) group_name = group_params_to_db['group_name'] + '-' + str(group_cnt) # 写入数据表中组信息 single_group_params_to_db = group_params_to_db.copy() single_group_params_to_db['group_uuid'] = group_uuid single_group_params_to_db['group_name'] = group_name single_group_params_to_db['create_time'] = datetime.datetime.now() df = pd.DataFrame.from_dict(single_group_params_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_unit_level_operation_record", con=engine, if_exists='append', index=False) # 拼装返回json的组信息 single_group_params_to_request = group_params_to_request.copy() single_group_params_to_request['group_uuid'] = group_uuid single_group_params_to_request['unit_name'] = group_name single_group_params_to_request['creative_list'] = [] group_cnt += 1 # 拼装组下面创意的信息 creative_cnt = 1 for image_md5 in video['imageList'][: self.advertiser_strategy['image_cnt']]: creative_uuid = str(uuid.uuid4()) creative_name = creative_params_to_db['creative_name'] + '_' + str(creative_cnt) single_creative_params_to_db = creative_params_to_db.copy() single_creative_params_to_db['creative_uuid'] = creative_uuid single_creative_params_to_db['creative_name'] = creative_name single_creative_params_to_db['photo_id'] = video['photo_id'] single_creative_params_to_db['image_md5'] = image_md5 single_creative_params_to_db['create_time'] = datetime.datetime.now() df = pd.DataFrame.from_dict(single_creative_params_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_creative_level_operation_record", con=engine, if_exists='append', index=False) single_creative_params_to_request = creative_params_to_request.copy() single_creative_params_to_request['creative_uuid'] = creative_uuid single_creative_params_to_request['creative_name'] = creative_name single_creative_params_to_request['photo_id'] = video['photo_id'] single_creative_params_to_request['image_md5'] = image_md5 # 单个创意信息加入到组里面的 creative_list 中 single_group_params_to_request['creative_list'].append(single_creative_params_to_request) creative_cnt += 1 # 组和创意信息添加到最终返回数据 group_list 中 self.res_data['group_list'].append(single_group_params_to_request) self.res_data['account_id'] = self.account_id self.res_data['campaign_id'] = self.campaign_id def assemble_group_and_programme_creative_params(self): """ 拼接组和程序化创意的参数 写入数据库 :return: """ # 获取组和创意的基本参数 group_params_to_request, group_params_to_db = self.get_group_params() creative_params_to_request, creative_params_to_db = self.get_programme_creative_params() # 每5个视频一组,来拼接组层级 和 程序化创意 层级的参数 self.res_data['group_list'] = [] cnt = 1 is_smart_cover = group_params_to_request.get('smart_cover', 0) for i in range(0, len(self.video), 5): group_uuid = str(uuid.uuid4()) group_name = group_params_to_db['group_name'] + '-' + str(cnt) # 写入数据表中组信息 single_group_params_to_db = group_params_to_db.copy() single_group_params_to_db['group_uuid'] = group_uuid single_group_params_to_db['group_name'] = group_name single_group_params_to_db['create_time'] = datetime.datetime.now() df = pd.DataFrame.from_dict(single_group_params_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_unit_level_operation_record", con=engine, if_exists='append', index=False) # 拼装返回json的组信息 single_group_params_to_request = group_params_to_request.copy() single_group_params_to_request['group_uuid'] = group_uuid single_group_params_to_request['unit_name'] = group_name single_group_params_to_request['creative_list'] = [] single_group_params_to_request['programCreative'] = None # 拼装程序化创意的信息 horizontal_photo_ids = [] vertical_photo_ids = [] cover_image_tokens = [] if i+5 < len(self.video): for j in range(i, i+5, 1): if self.video[j]['material_type'] == 2: horizontal_photo_ids.append(self.video[j]['photo_id']) cover_image_tokens.append(self.video[j]['imageList'][0]) if self.video[j]['material_type'] == 1: vertical_photo_ids.append(self.video[j]['photo_id']) cover_image_tokens.append(self.video[j]['imageList'][0]) if len(cover_image_tokens) >= 5: cover_image_tokens = cover_image_tokens[:4] creative_uuid = str(uuid.uuid4()) package_name = creative_params_to_db['package_name'] + '_' + str(cnt) single_creative_params_to_db = creative_params_to_db.copy() single_creative_params_to_db['creative_uuid'] = creative_uuid single_creative_params_to_db['package_name'] = package_name single_creative_params_to_db['horizontal_photo_ids'] = str(horizontal_photo_ids) single_creative_params_to_db['vertical_photo_ids'] = str(vertical_photo_ids) single_creative_params_to_db['cover_image_tokens'] = str(cover_image_tokens) if not is_smart_cover else None single_creative_params_to_db['create_time'] = datetime.datetime.now() df = pd.DataFrame.from_dict(single_creative_params_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_program_creative_level_operation_record", con=engine, if_exists='append', index=False) single_creative_params_to_request = creative_params_to_request.copy() single_creative_params_to_request['creative_uuid'] = creative_uuid single_creative_params_to_request['package_name'] = package_name single_creative_params_to_request['horizontal_photo_ids'] = horizontal_photo_ids single_creative_params_to_request['vertical_photo_ids'] = vertical_photo_ids if not is_smart_cover: single_creative_params_to_request['image_md5s'] = cover_image_tokens else: # material_type=1 竖版 # material_type=2 横版 for j in range(i, len(self.video)): if self.video[j]['material_type'] == 2: horizontal_photo_ids.append(self.video[j]['photo_id']) cover_image_tokens.append(self.video[j]['imageList'][0]) if self.video[j]['material_type'] == 1: vertical_photo_ids.append(self.video[j]['photo_id']) cover_image_tokens.append(self.video[j]['imageList'][0]) if len(cover_image_tokens) >= 5: cover_image_tokens = cover_image_tokens[:4] creative_uuid = str(uuid.uuid4()) package_name = creative_params_to_db['package_name'] + '_' + str(cnt) single_creative_params_to_db = creative_params_to_db.copy() single_creative_params_to_db['creative_uuid'] = creative_uuid single_creative_params_to_db['package_name'] = package_name single_creative_params_to_db['horizontal_photo_ids'] = str(horizontal_photo_ids) single_creative_params_to_db['vertical_photo_ids'] = str(vertical_photo_ids) single_creative_params_to_db['cover_image_tokens'] = str(cover_image_tokens) if not is_smart_cover else None single_creative_params_to_db['create_time'] = datetime.datetime.now() df = pd.DataFrame.from_dict(single_creative_params_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_program_creative_level_operation_record", con=engine, if_exists='append', index=False) single_creative_params_to_request = creative_params_to_request.copy() single_creative_params_to_request['creative_uuid'] = creative_uuid single_creative_params_to_request['package_name'] = package_name single_creative_params_to_request['horizontal_photo_ids'] = horizontal_photo_ids single_creative_params_to_request['vertical_photo_ids'] = vertical_photo_ids if not is_smart_cover: single_creative_params_to_request['image_md5s'] = cover_image_tokens single_group_params_to_request['programCreative'] = single_creative_params_to_request cnt += 1 self.res_data['group_list'].append(single_group_params_to_request) self.res_data['account_id'] = self.account_id self.res_data['campaign_id'] = self.campaign_id # TODO 修改操作 class ParseModifyRequest(object): def __init__(self): pass # TODO 关停操作 class ParseShutDownRequest(object): def __init__(self): pass