import uuid import datetime import tornado import json import traceback import pandas as pd import pymysql import os import yaml import random import requests from utils.commonFunc import get_db_engine, is_contains_gender, is_contains_age, is_contains_region from utils.BaseClass import NpEncoder from config.url import create_campaign_url, create_group_and_creative_url, headers import logging from concurrent_log import ConcurrentTimedRotatingFileHandler log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%Y/%m/%d %I:%M:%S %p') log_handler = ConcurrentTimedRotatingFileHandler("logs/ai_target_combine.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_target_combine_logger %s' % id(logger)) with open('config/config.yaml', mode='r', encoding='utf-8') as f: config = yaml.load(f.read(), Loader=yaml.FullLoader) class AiTargetCombine(tornado.web.RequestHandler): def post(self): # 解析接收到的参数 data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') account_id = data['account_id'] project_id = data['project_id'] logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s" % (account_id, os.getpid(), os.getppid())) try: # 1、创建类实例 ins = GetTargetAndAssemblyParameters(account_id, project_id) # 2、判断当天是否已经存在包含【优质定向】的广告计划,如果有则获取该计划id,否则需要新建计划 sql = """select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s and put_create_time >= curdate() and campaign_name like '%s'""" % (ins.account_id, '%%优质定向%%') df = pd.read_sql(sql, ins.engine) if df.empty: # ins.add_campaign() ins.campaign_id = 123123 else: ins.campaign_id = df['campaign_id'].values[0] # 3、拼装广告组和创意的参数 ins.assembly_group_and_creative_params() # 4、信息写如到智能策略表中 ins.write_intelligence_strategy_table() # 5、发送创建组和创意的请求 # request = requests.post(create_group_and_creative_url, # headers=headers, # data=json.dumps(ins.request_data, cls=NpEncoder)) # response_data = json.loads(request.text) # 6、接口的返回信息更新到数据库表 # ins.update_intelligence_strategy_table(message=response_data) ins.update_intelligence_strategy_table(message="异步执行") logger.info("account_id = %s, 策略uuid = %s 的请求返回信息:%s" % (ins.account_id, ins.ai_strategy_uuid, "异步执行")) self.write(json.dumps({"account_id": account_id, "message": "异步执行", "code": 1})) self.flush() except: logger.error("account_id = %s, traceback is %s" % (ins.account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class GetTargetAndAssemblyParameters(object): def __init__(self, account_id, project_id): self.account_id = account_id self.project_id = project_id self.engine = None self.product_engine = None self.db_config = None self.signature_target_combine = None self.final_target_combine = [] self.advertiser_strategy_id = None self.advertiser_strategy = {} self.ai_strategy_uuid = str(uuid.uuid4()) self.request_data = {} self.campaign_id = None self.operation_type = 1 # 新增广告组 self.get_database_engine() # 获取数据库信息 self.get_advertiser_strategy_info() # 获取账户配置信息 self.get_signature_and_target() # 获取定向组合 self.target_combine_is_subset() # 过滤出为账户相信子集的定向组合 def get_database_engine(self): # 数据库连接引擎,依据开发环境/生产环境 进行切换 if os.getenv('environment', 'unknown') == 'dev': self.engine = get_db_engine(config['devDB']) self.db_config = config['devDB'] else: self.engine = get_db_engine(config['productDB']) self.db_config = config['productDB'] # 有些表需要连接生产数据库,不管当前是开发环境还是测试环境 self.product_engine = get_db_engine(config['productDB']) def get_advertiser_strategy_info(self): """ 获取指定账户下客户的策略信息 为后续的写入库表,校验一致性做准备 """ sql = """ select * from ctop_ai_kuaishou_advertiser_strategy where account_id = %d limit 1 """ % self.account_id advertiser_strategy_df = pd.read_sql(sql, self.engine) self.advertiser_strategy_id = int(advertiser_strategy_df['id'].values[0]) self.advertiser_strategy = advertiser_strategy_df.T.to_dict()[0] def add_campaign(self): """ 新增广告计划,如果存在则跳过。 response_data = { "code": 0, "data": { "account_id": 23212, "campaign_id": 30956133, "campaign_create_time": "2021-01-13 13:50:16" }, "message": "SUCCESS"} """ # 1、请求创建广告计划的接口 self.advertiser_strategy['campaign_name'] = self.advertiser_strategy['campaign_name'].replace('{{自定义}}', '') campaign_name = self.advertiser_strategy['campaign_name'] + '_优质定向' create_campaign_request_data = {'account_id': self.account_id, 'campaign_name': campaign_name, 'type': self.advertiser_strategy['campaign_type']} request = requests.post(url=create_campaign_url, headers=headers, data=json.JSONEncoder().encode(create_campaign_request_data)) response_data = json.loads(request.text) # 2、对接口返回的结果处理 campaign_info_to_db = { 'campaign_uuid': str(uuid.uuid4()), 'account_id': self.account_id, 'ai_strategy_uuid': self.ai_strategy_uuid, 'campaign_name': campaign_name, 'campaign_type': self.advertiser_strategy['campaign_type'], 'create_time': datetime.datetime.now() } res = 0 if response_data['code'] == 0: self.campaign_id = response_data['data'].get('campaign_id', None) campaign_info_to_db['campaign_id'] = self.campaign_id campaign_info_to_db['campaign_create_time'] = response_data['data'].get('campaign_create_time', None) campaign_info_to_db['status'] = response_data.get('code', None) campaign_info_to_db['message'] = response_data.get('message', None) else: campaign_info_to_db['message'] = response_data['message'] campaign_info_to_db['status'] = response_data['code'] res = -1 # 3、写入计划层级的操作表 df = pd.DataFrame.from_dict(campaign_info_to_db, orient='index').T df.to_sql(name="ctop_ai_kuaishou_campaign_level_operation_record", con=self.engine, if_exists='append', index=False) return res def get_signature_and_target(self): """ 从 ctop_ai_kuaishou_signature_recommended_target_combine 表中读取素材和对应的定向 读取配置文件的条件,筛选出符合条件的定向组合 """ sql = """ select * from ctop_ai_kuaishou_signature_recommended_target_combine where project_id = %s -- and stat_date = curdate() """ % self.project_id df = pd.read_sql(sql, self.engine) # 计算组合的概率,相对于实际投放概率高出了百分之多少 df['improve_ratio'] = (df['combine_estimate_prob'] - df['actual_prob']) / df['actual_prob'] # 两种类型的过滤标准不一样,分开进行判断,然后对结果进行合并 df_1 = df[(df['target_type'] == 'action_ratio') & (df['improve_ratio'] >= config['filterTargetCombine']['actionRatio']['improveRatio']) & (df['sample_size'] >= config['filterTargetCombine']['actionRatio']['sampleSize'])] df_2 = df[(df['target_type'] == 'convertRatio') & (df['improve_ratio'] >= config['filterTargetCombine']['convertRatio']['improveRatio']) & (df['sample_size'] >= config['filterTargetCombine']['convertRatio']['sampleSize'])] merge_df = pd.concat([df_1, df_2], axis=0) # 'age', 'gender', 'city', 'business', 'province', 'client' features = [key for key, value in config['bayesDim'].items() if value['isOn']] self.signature_target_combine = merge_df[['id', 'signature'] + features].to_dict(orient='records') def write_intelligence_strategy_table(self): """ 拼接好的组创意的参数写入到 ctop_ai_kuaishou_intelligence_strategy 中的 ai_strategy_request_content """ intelligence_strategy_dict = \ { 'ai_strategy_uuid': self.ai_strategy_uuid, 'advertiser_strategy_id': self.advertiser_strategy_id, 'account_id': self.account_id, 'ai_strategy_request_content': str(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=self.engine, if_exists='append', index=False) def target_combine_is_subset(self): """ ctop_ai_kuaishou_signature_recommended_target_combine 表中的 city 和 province 都对应 self.advertiser_strategy 的 region [] varchar age 对应 ages_range [] varchar client 对应 platform_os int business_interest 对应 business_interest varchar gender 对应 gender int """ gender_bool = True age_bool = True region_bool = True for item in self.signature_target_combine: if 'gender' in item.keys(): gender_bool, gender_dict = is_contains_gender(item['gender'], self.advertiser_strategy['gender']) if 'age' in item.keys(): age_bool, age_dict = is_contains_age(item['age'], self.advertiser_strategy['age_min'], self.advertiser_strategy['age_max'], self.advertiser_strategy['ages_range']) if 'city' in item.keys(): region_bool, region_dict = is_contains_region(item['city'], self.advertiser_strategy['region']) if gender_bool & age_bool & region_bool: target_combine = {'signature': item['signature'], 'target_combine_id': item['id']} target_combine.update(gender_dict) if 'gender' in item.keys() else None target_combine.update(age_dict) if 'age' in item.keys() else None target_combine.update(region_dict) if 'city' in item.keys() else None self.final_target_combine.append(target_combine) else: # TODO logger pass # 从集合中随机选取N个组合 combine_cnt = config['filterTargetCombine']['combineCnt'] random.shuffle(self.final_target_combine) self.final_target_combine = self.final_target_combine[:combine_cnt] def assembly_group_and_creative_params(self): """ 拼接参数,请求创建接口 """ group_params, group_params_to_db = self.get_group_params() creative_params, creative_params_to_db = self.get_creative_params() self.request_data['group_list'] = [] group_cnt = 1 batch_group_params_to_db = [] batch_creative_params_to_db = [] for item in self.final_target_combine: # 使用优质定向更新组层级的参数 target_combine = item.copy() del target_combine['target_combine_id'] del target_combine['signature'] group_params.update(target_combine) # 写入数据库的字段,list类型需要转化为str group_params_to_db.update(target_combine) cols = ['ages_range', 'region'] for col in cols: if col in target_combine.keys(): group_params_to_db[col] = str(group_params_to_db[col]) group_uuid = str(uuid.uuid4()) group_name = group_params['unit_name'] # 获取素材名称, 在广告组命名中进行替换 (同时获取photo_id, 用于拼装创意层级的参数) sql = """select photo_id, photo_name from ctop_kuaishou_video_get where account_id = %s and signature = '%s' limit 1""" % \ (self.account_id, item['signature']) signature_df = pd.read_sql(sql, self.product_engine) photo_name = signature_df['photo_name'].values[0] if not signature_df.empty else 'unknown' photo_id = signature_df['photo_id'].values[0] if not signature_df.empty else None group_name.replace('{{素材名称}}', photo_name) group_name = group_name + '_优质定向' + '_' + str(group_cnt) # 写入数据表中组信息 single_group_params_to_db = group_params_to_db.copy() single_group_params_to_db['target_combine_id'] = item['target_combine_id'] 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() batch_group_params_to_db.append(single_group_params_to_db) # 拼装返回json的组信息 single_group_params_to_request = group_params.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 # 拼装广告组下面的创意信息 # 一个素材组合为15个一样的创意,因信息流已经下架,不需要提供封面信息 for creative_cnt in range(15): creative_uuid = str(uuid.uuid4()) creative_name = creative_params_to_db['creative_name'] + '_' + str(creative_cnt + 1) 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'] = int(photo_id) if photo_id else None single_creative_params_to_db['create_time'] = datetime.datetime.now() batch_creative_params_to_db.append(single_creative_params_to_db) single_creative_params_to_request = creative_params.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'] = int(photo_id) if photo_id else None # 单个创意信息加入到组里面的 creative_list 中 single_group_params_to_request['creative_list'].append(single_creative_params_to_request) # 组和创意信息添加到最终返回数据 group_list 中 self.request_data['group_list'].append(single_group_params_to_request) # 创意层级信息批量写入数据库 creative_df = pd.DataFrame(batch_creative_params_to_db) creative_df.to_sql(name="ctop_ai_kuaishou_creative_level_operation_record", con=self.engine, if_exists='append', index=False) # 组层级信息批量写入数据库 group_df = pd.DataFrame(batch_group_params_to_db) group_df.to_sql(name="ctop_ai_kuaishou_unit_level_operation_record", con=self.engine, if_exists='append', index=False) self.request_data['account_id'] = self.account_id self.request_data['campaign_id'] = self.campaign_id def get_group_params(self): # 1、从账户信息表读取基本的配置信息 group_params = {'unit_name': self.advertiser_strategy.get('group_name'), # 'put_status': self.advertiser_strategy.get('group_status'), 'template_id': self.advertiser_strategy.get('template_id'), 'bid_type': self.advertiser_strategy.get('bid_type'), 'use_app_market': self.advertiser_strategy.get('use_app_market'), 'app_store': self.advertiser_strategy.get('app_store'), 'bid': self.advertiser_strategy.get('bid'), 'cpa_bid': self.advertiser_strategy.get('cpa_bid'), 'smart_bid': self.advertiser_strategy.get('smart_bid'), 'ocpx_action_type': self.advertiser_strategy.get('ocpx_action_type'), 'deep_conversion_type': self.advertiser_strategy.get('deep_conversion_type'), 'scene_id':self.advertiser_strategy.get('scene_id'), 'unit_type': self.advertiser_strategy.get('unit_type'), 'begin_time': self.advertiser_strategy.get('begin_time'), 'end_time': self.advertiser_strategy.get('end_time'), 'schedule_time': self.advertiser_strategy.get('schedule_time'), 'day_budget': self.advertiser_strategy.get('unit_day_budget'), 'day_budget_schedule': self.advertiser_strategy.get('unit_day_budget_schedule'), 'convert_id': self.advertiser_strategy.get('convert_id'), 'url_type': self.advertiser_strategy.get('url_type'), 'web_uri_type': self.advertiser_strategy.get('web_uri_type'), 'url': self.advertiser_strategy.get('url'), 'schema_uri': self.advertiser_strategy.get('schema_uri'), 'app_id': self.advertiser_strategy.get('app_id'), 'show_mode': self.advertiser_strategy.get('show_mode'), 'speed': self.advertiser_strategy.get('speed'), 'site_type': self.advertiser_strategy.get('site_type'), 'gift_data': self.advertiser_strategy.get('gift_data'), 'video_landing_page': self.advertiser_strategy.get('video_landing_page'), 'auto_target': self.advertiser_strategy.get('auto_target'), 'auto_create_photo': self.advertiser_strategy.get('auto_create_photo'), # 'item_id': self.advertiser_strategy.get('item_id'), # 'merchant_item_put_type': self.advertiser_strategy.get('merchant_item_put_type'), # 'fiction_id': self.advertiser_strategy.get('fiction_id'), 'smart_cover': self.advertiser_strategy.get('smart_cover'), 'asset_mining': self.advertiser_strategy.get('asset_mining'), # 'consult_id': self.advertiser_strategy.get('consult_id'), # 'adv_card_option': self.advertiser_strategy.get('adv_card_option'), # 'adv_card_list': self.advertiser_strategy.get('adv_card_list'), # 'card_type': self.advertiser_strategy.get('card_type'), # 'merchandise_id': self.advertiser_strategy.get('merchandise_id'), # 'merchandise_type': self.advertiser_strategy.get('merchandise_type'), # 'intention_target': self.advertiser_strategy.get('intention_target'), # 'dpa_unit_param': self.advertiser_strategy.get('dpa_unit_param'), 'region': self.advertiser_strategy.get('region'), 'district_ids': self.advertiser_strategy.get('district_ids'), 'user_type': self.advertiser_strategy.get('user_type'), 'age_min': self.advertiser_strategy.get('age_min'), 'age_max': self.advertiser_strategy.get('age_min'), 'ages_range': self.advertiser_strategy.get('ages_range'), 'gender': self.advertiser_strategy.get('gender'), 'platform_os': self.advertiser_strategy.get('platform_os'), 'android_osv': self.advertiser_strategy.get('android_osv'), 'ios_osv': self.advertiser_strategy.get('ios_osv'), 'network': self.advertiser_strategy.get('network'), 'device_brand': self.advertiser_strategy.get('device_brand'), 'device_price': self.advertiser_strategy.get('device_price'), 'business_interest_type': self.advertiser_strategy.get('business_interest_type'), 'business_interest': eval(self.advertiser_strategy.get('business_interest')) if self.advertiser_strategy.get('business_interest') else None, 'fans_star': self.advertiser_strategy.get('fans_star'), 'interest_video': self.advertiser_strategy.get('interest_video'), 'app_interest': self.advertiser_strategy.get('app_interest'), # 'app_interest_ids': self.advertiser_strategy.get('app_interest_ids'), 'app_ids':self.advertiser_strategy.get('app_ids'), 'filter_converted_level': self.advertiser_strategy.get('filter_converted_level'), 'population': self.advertiser_strategy.get('population'), 'exclude_population': self.advertiser_strategy.get('exclude_population'), 'paid_audience': self.advertiser_strategy.get('paid_audience'), 'is_open': self.advertiser_strategy.get('is_open'), 'no_age_break': self.advertiser_strategy.get('no_age_break'), 'no_gender_break': self.advertiser_strategy.get('no_gender_break'), 'no_area_break': self.advertiser_strategy.get('no_area_break'), 'behavior_interest': self.advertiser_strategy.get('behavior_interest') } group_params['unit_name'] = group_params['unit_name'].replace('{{自定义}}', '') # 2、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 # 3、广告组名称,通配符的替换 # 900299 -{{广告位置}}--{{创意制作方式}}--{{日期}}- (日期、素材名称、应用名称、应用标记、创意制作方式、广告位置) # 3-1 获取应用标记和应用名称 sql = """select app_version, app_name from ctop_kuaishou_app_list where app_id = %s limit 1""" % \ group_params.get('app_id', -1) app_info_df = pd.read_sql(sql, self.engine) app_version = app_info_df['app_version'].values[0] if not app_info_df.empty else 'unknown' app_name = app_info_df['app_name'].values[0] if not app_info_df.empty else 'unknown' # 3-2 素材名称, 在拼装广告组和创意层级函数中获取 # 3-3 获取创意制作方式 unit_type = '自定义' if group_params.get('unit_type') == 4 \ else ('程序化创意' if group_params.get('unit_type') == 5 else '程序化创意2.0') # 3-4 获取广告位置 scene_id_dict = {1: '优选广告位', 3: '视频播放页广告-便利贴广告', 5: '联盟广告', 6: '上下滑大屏广告', 7: '信息流广告', 10: '联盟场景'} scene_id = '+'.join([scene_id_dict[item] for item in eval(group_params.get('scene_id'))]) unit_name = group_params['unit_name'].replace('{{应用标记}}', app_version). \ replace('{{应用名称}}', app_name). \ replace('{{创意制作方式}}', unit_type). \ replace('{{广告位置}}', scene_id) # 3-5 日期, 在拼装广告组和创意层级函数中获取 now_date = str(datetime.datetime.now().date()) if '日期' in group_params['unit_name']: unit_name = unit_name.replace('{{日期}}', now_date) else: unit_name = unit_name + '_' + now_date group_params['unit_name'] = unit_name # 4、处理 cpa_bid # 因为存在 【单出价请输入单数字,阶梯出价请用/隔开】 如:1900/2000/2100 cpa_bid varchar(255) DEFAULT NULL COMMENT '出价' # 处理方式按'/'进行分割,取最后一个值(也就是最大值),并转化为int型 group_params['cpa_bid'] = int(group_params['cpa_bid'].split('/')[-1]) if '/' in group_params['cpa_bid'] \ else (int(group_params['cpa_bid'].split('-')[-1]) if '-' in group_params['cpa_bid'] else int(group_params['cpa_bid'])) # 5、 写入数据库的组参数 group_params_to_db = group_params.copy() group_params_to_db['group_name'] = group_params_to_db.pop('unit_name') 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 # 6、用于组装发送请求的部分字段,需要转化为list类型 cols_to_list = ['app_store', 'scene_id', 'region','day_budget_schedule', 'ages_range', 'device_brand', 'business_interest', 'fans_star', 'interest_video', 'app_interest', 'app_interest_ids', 'app_ids', 'population','district_ids', 'exclude_population', 'paid_audience'] for col in cols_to_list: if col in group_params.keys(): group_params[col] = eval(group_params[col]) if group_params[col] else None return group_params, group_params_to_db def get_creative_params(self): # 1、从账户信息表读取基本的配置信息 creative_params = {'creative_name': self.advertiser_strategy.get('creative_name'), # 'put_status': self.advertiser_strategy.get('put_status'), 'action_bar_text': self.advertiser_strategy.get('action_bar_text'), 'description': self.advertiser_strategy.get('description'), 'short_slogan': self.advertiser_strategy.get('short_slogan'), 'sticker_title': self.advertiser_strategy.get('sticker_title'), 'overlay_type': self.advertiser_strategy.get('overlay_type'), 'expose_tag': self.advertiser_strategy.get('expose_tag'), 'new_expose_tag':self.advertiser_strategy.get('new_expose_tag'), 'site_id': self.advertiser_strategy.get('site_id'), 'click_track_url': self.advertiser_strategy.get('click_track_url'), 'impression_url': self.advertiser_strategy.get('impression_url'), 'ad_photo_played_t3s_url': self.advertiser_strategy.get('ad_photo_played_t3s_url'), 'actionbar_click_url': self.advertiser_strategy.get('actionbar_click_url'), 'creative_category': self.advertiser_strategy.get('creative_category'), 'creative_tag': self.advertiser_strategy.get('creative_tag'), # 'live_creative_type': self.advertiser_strategy.get('live_creative_type') } # 1、description 可能存在多个值,需要从中随机获取一个 if creative_params['description'] is not None: description_lst = eval(creative_params['description']) random.shuffle(description_lst) creative_params['description'] = description_lst[0] # 2 写入数据库的创意参数 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 # 3、用于组装发送请求的部分字段,需要转化为list类型 cols_to_list = ['creative_tag', 'new_expose_tag'] for col in cols_to_list: if col in creative_params.keys(): creative_params[col] = eval(creative_params[col]) if creative_params[col] else None return creative_params, creative_params_to_db def update_intelligence_strategy_table(self, message=None): """ 更新 intelligence_strategy_table :return: """ db_con = pymysql.connect(host=self.db_config['host'], port=self.db_config['port'], user=self.db_config['username'], password=self.db_config['password'], database=self.db_config['database'], charset='utf8') cursor = db_con.cursor() 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()