import pandas as pd from utils.UrlConfig import * import json import logging import traceback import random import tornado.web import requests from concurrent_log import ConcurrentTimedRotatingFileHandler from utils.ConstantConfig import * from utils.DataBaseConfig import * from utils.CommonFunction import get_history_video_info, get_new_video_info, get_missing_video_info, \ get_top_video_info, get_campaign_and_group_name_rule, get_app_detail, get_request_data, \ refresh_video, get_app_list from ai_strategy_request_func import ai_strategy_request_parse log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%m/%d/%Y %I:%M:%S %p') log_handler = ConcurrentTimedRotatingFileHandler("logs/ai_auto_create.log", when="midnight", backupCount=100) log_handler.setFormatter(log_formatter) logger = logging.getLogger('ai_time_task_creative_logger') logger.addHandler(log_handler) logger.setLevel(logging.DEBUG) print('id of ai_time_task_creative_logger %s' % id(logger)) logger.info("ai_time_task_creative_server started!") class AiHistoricalMissingMaterial(tornado.web.RequestHandler): """ 补充历史遗漏素材(关联创意个数为0个素材) """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') account_id = data.get('account_id') campaign_id = data.get('campaign_id') # test_video_cnt 是给 新账户测试使用的 test_video_cnt = data.get('test_video_cnt', None) logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: sql = """ select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1 """ % account_id df = pd.read_sql(sql, engine) channel_type = df['channel_type'].values[0] if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info video_list = get_missing_video_info(channel_type=channel_type, account_id=account_id, cnt=test_video_cnt if test_video_cnt else missing_video_cnt, url=get_missing_video_url, start_time=missing_video_start_time, end_time=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") if test_video_cnt else missing_video_end_time) # 返回视频中有通用视频和特定视频,需要分开处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", if video_list['code'] == 0: video_df = pd.DataFrame(video_list['data']) # 1-非通用视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '遗漏素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version + '--优选广告位-' + '遗漏素材', group_name=app_version + '-优选广告位-不限-自定义-' + '遗漏素材', ai_strategy_remark="补充遗漏素材", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 补充遗漏素材-特定视频,ai策略的返回信息:%s" % (account_id, app_id, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] if not general_video_df.empty: general_video_cnt = len(general_video_df) app_version_list = list(app_dict) random.shuffle(app_version_list) video_cnt = 0 for app_version in app_version_list[:general_video_cnt]: campaign_id = None app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '遗漏素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break single_video = [general_video_df.iloc[video_cnt, :].T.to_dict()] request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=single_video, campaign_name=app_version + '--优选广告位-' + '遗漏素材', group_name=app_version + '-优选广告位-不限-自定义-' + '遗漏素材', ai_strategy_remark="补充遗漏素材", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) video_cnt += 1 logger.info("account_id=%s, app_id=%s, 补充遗漏素材-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: logger.info("account_id=%s, 补充遗漏素材: 返回的视频中没有通用视频" % account_id) self.write(json.dumps({"message": "account_id=%s, 多应用的补充遗漏素材请求已走完!" % account_id})) self.flush() else: video_info = get_missing_video_info(channel_type=channel_type, account_id=account_id, cnt=test_video_cnt if test_video_cnt else missing_video_cnt, url=get_missing_video_url, start_time=missing_video_start_time, end_time=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") if test_video_cnt else missing_video_end_time) if video_info['code'] != 0: logger.info("account_id = %s,补充历史遗漏素材,没有获取到视频,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "补充历史遗漏素材,没有获取到视频,不发送ai策略请求!"})) self.flush() else: campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.info("没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!") self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_info['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="补充遗漏素材", name_replace="补充遗漏素材") request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, 补充遗漏素材,ai策略的返回信息:%s" % (account_id, request)) self.write(json.dumps(request)) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class AiAutoCreative(tornado.web.RequestHandler): """ 每5分钟,检查一次是否有上新素材,有的话,就自动创建 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") account_id = data.get('account_id') campaign_id = data.get('campaign_id') logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: # 刷新快手视频信息 refresh_video(account_id=account_id, url=refresh_video_url) sql = """ select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1 """ % account_id df = pd.read_sql(sql, engine) if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info video_list = get_new_video_info(channel_type=df['channel_type'].values[0],account_id=account_id, url=get_new_video_url) # 返回的视频中有通用视频和特定视频,需要分开处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", if video_list['code'] == 0: video_df = pd.DataFrame(video_list['data']) # 1-特定视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '素材自动上新') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version + '--优选广告位-' + '素材自动上新', group_name=app_version + '-优选广告位-不限-自定义-' + '素材自动上新', ai_strategy_remark="素材自动上新", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 素材自动上新-特定视频,ai策略的返回信息:%s" % (account_id, app_id, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] if not general_video_df.empty: general_video_cnt = len(general_video_df) app_version_list = list(app_dict) random.shuffle(app_version_list) video_cnt = 0 for app_version in app_version_list[:general_video_cnt]: campaign_id = None app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '素材自动上新') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break single_video = [general_video_df.iloc[video_cnt, :].T.to_dict()] request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=single_video, campaign_name=app_version + '--优选广告位-' + '素材自动上新', group_name=app_version + '-优选广告位-不限-自定义-' + '素材自动上新', ai_strategy_remark="素材自动上新", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) video_cnt += 1 logger.info("account_id=%s, app_id=%s, 素材自动上新-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: logger.info("account_id=%s, 素材自动上新: 返回的视频中没有通用视频" % account_id) else: logger.info("account_id=%s, 素材自动上新: 没有获取到视频" % account_id) self.write(json.dumps({"message": "account_id=%s, 多应用的素材自动上新已走完!" % account_id})) self.flush() else: start_time = (datetime.date.today()).strftime('%Y-%m-%d %H:%M:%S') end_time = (datetime.datetime.now()).strftime('%Y-%m-%d %H:%M:%S') # 1-1 获取当天计划中包含“自动上新”计划的,计划id sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s and put_create_time >= '%s' and put_create_time <= '%s' """ % (account_id, start_time, end_time) df = pd.read_sql(sql, engine) channel_type = df['channel_type'].values[0] for row in df.itertuples(): if '素材自动上新' in getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break video_info = get_new_video_info(channel_type=channel_type, account_id=account_id, url=get_new_video_url) if video_info['code'] != 0: self.write(json.dumps({"account_id": account_id, "message": "素材自动上新没有获取到视频,不发送ai策略请求!"})) self.flush() else: campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.info("account_id = %s, 素材自动上新 没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "素材自动上新 没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_info['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="自动上新", name_replace="素材自动上新") request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, 素材自动上新 ai策略的返回信息:%s" % (account_id, request)) self.write(json.dumps(request)) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class AiHighQualityMaterial(tornado.web.RequestHandler): """ 补充历史高质量素材 -- 测试通过 在AD后台创建成功,检查AD信息和数据库表 没有问题 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") account_id = data.get('account_id') campaign_id = data.get('campaign_id') logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: sql = """select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1 """ % account_id df = pd.read_sql(sql, engine) channel_type = df['channel_type'].values[0] if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info video_list = get_top_video_info(channel_type=channel_type, account_id=account_id, cnt=high_quality_video_cnt, start_time=(datetime.datetime.now() + datetime.timedelta(days=-high_quality_video_days)). strftime("%Y-%m-%d %H:%M:%S"), end_time=(datetime.datetime.now() + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), url=get_high_quality_video_url) # 单独处理返回的时候中,有通用视频,需要单独处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", if video_list['code'] == 0: video_df = pd.DataFrame(video_list['data']) # 1-特定视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '高质量素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version + '--优选广告位-' + '高质量素材', group_name=app_version + '-优选广告位-不限-自定义-' + '高质量素材', ai_strategy_remark="高质量素材复建", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 高质量素材复建-特定视频,ai策略的返回信息:%s" % (account_id, app_id, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] if not general_video_df.empty: general_video_cnt = len(general_video_df) app_version_list = list(app_dict) random.shuffle(app_version_list) video_cnt = 0 for app_version in app_version_list[:general_video_cnt]: campaign_id = None app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '高质量素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break single_video = [general_video_df.iloc[video_cnt, :].T.to_dict()] request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=single_video, campaign_name=app_version + '--优选广告位-' + '高质量素材', group_name=app_version + '-优选广告位-不限-自定义-' + '高质量素材', ai_strategy_remark="高质量素材复建", unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) video_cnt += 1 logger.info("account_id=%s, app_id=%s, 高质量素材复建-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: logger.info("account_id=%s, 高质量素材复建: 返回的视频中没有通用视频" % account_id) else: logger.info("account_id=%s, 高质量素材复建: 没有获取到视频" % account_id) self.write(json.dumps({"message": "account_id=%s, 多应用的高质量素材复建请求已走完!" % account_id})) self.flush() else: video_info = get_top_video_info(channel_type=channel_type, account_id=account_id, cnt=high_quality_video_cnt, start_time=(datetime.datetime.now() + datetime.timedelta(days=-high_quality_video_days)). strftime("%Y-%m-%d %H:%M:%S"), end_time=(datetime.datetime.now() + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), url=get_high_quality_video_url) if video_info['code'] != 0: logger.info("account_id = %s, 高质量素材复建,没有获取到视频,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "高质量素材复建,没有获取到视频,不发送ai策略请求!"})) self.flush() else: campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.error("account_id = %s, 没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_info['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="高质量素材复建", name_replace="高质量素材") request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, 高质量素材复建,ai策略的返回信息:%s" % (account_id, request)) self.write(json.dumps(request)) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class AiUpCreativeByHour(tornado.web.RequestHandler): """ 1点到20点,每小时执行一次,检查当前时点创意个数是否达到hour*100 没有达到则按照当前时点值,补充不同类型的视频 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') account_id = data.get('account_id') campaign_id = data.get('campaign_id') logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: # 1 从客户策略表中获取该账户下 每个素材搭配的封面个数, 来计算需要补充的素材个数 sql = """select image_cnt from ctop_ai_kuaishou_advertiser_strategy where account_id = %s limit 1""" % account_id df = pd.read_sql(sql, engine) image_cnt = int(df['image_cnt'].values[0]) # 2、获取从0点到now()为止,该账户下已有的创意个数 M sql = """select count(1) creative_cnt from ctop_kuaishou_creative where account_id = %s and creative_id is not NULL and creative_create_time <= '%s' and creative_create_time >= '%s'""" % (account_id, (datetime.datetime.now()).strftime('%Y-%m-%d %H:%M:%S'), (datetime.datetime.now()).strftime('%Y-%m-%d')) df = pd.read_sql(sql, engine) exist_creative_cnt = int(df['creative_cnt'].values[0]) now_hour = datetime.datetime.now().hour now = datetime.datetime.now() channel_type = df['channel_type'].values[0] should_creative_cnt = now_hour * 100 # 3、计算需要补充的创意个数和视频个数,并按当前小时值,补充不同类似的视频 if exist_creative_cnt < should_creative_cnt: need_video_cnt = (should_creative_cnt - exist_creative_cnt) // image_cnt logger.info("account_id = %s, 按小时需要补充的视频数量 = %s" % (account_id, need_video_cnt)) if now_hour % 5 == 0: # campaign_group_name_place 用于广告计划和广告组名称的标识 campaign_group_name_place = "高质量素材" video_list = get_top_video_info(channel_type=channel_type, account_id=account_id, url=get_high_quality_video_url, start_time=(now + datetime.timedelta(days=-high_quality_video_days)). strftime("%Y-%m-%d %H:%M:%S"), end_time=(now + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), cnt=need_video_cnt, app_version=None) elif (now_hour % 5 == 1) or (now_hour % 5 == 2): campaign_group_name_place = "历史素材打捞" video_list = get_history_video_info(channel_type=channel_type, account_id=account_id, url=get_history_video_url, start_time=history_video_start_time, end_time=history_video_end_time, video_cnt=need_video_cnt, related_creative_max_cnt=related_creative_max_cnt) else: campaign_group_name_place = "遗漏素材" video_list = get_missing_video_info(channel_type=channel_type, account_id=account_id, cnt=need_video_cnt, url=get_missing_video_url, start_time=missing_video_start_time, end_time=missing_video_end_time) if video_list['code'] == 0: # 根据获取到的视频进行创建 sql = """select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1""" % account_id df = pd.read_sql(sql, engine) if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info # 单独处理返回的时候中,有通用视频,需要单独处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", video_df = pd.DataFrame(video_list['data']) # 1-特定视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + campaign_group_name_place) == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version+'--优选广告位-'+campaign_group_name_place, group_name=app_version+'-优选广告位-不限-自定义-'+campaign_group_name_place, ai_strategy_remark="按小时补充创意-"+campaign_group_name_place, unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, %s补充创意-特定视频,ai策略的返回信息:%s" % (account_id, app_id, campaign_group_name_place, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] if not general_video_df.empty: # 每 batch_num 一组 随机选择一个app_id 进行创建 batch_num = 10 general_video_cnt = len(general_video_df) app_version_list = list(app_dict) random.shuffle(app_version_list) need_app_id_cnt = general_video_cnt // batch_num + 1 video_cnt = 0 for app_version in app_version_list[:need_app_id_cnt]: campaign_id = None app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + campaign_group_name_place) == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break batch_general_video_list = [] for k, v in general_video_df.iloc[video_cnt: video_cnt+batch_num, :].T.to_dict().items(): batch_general_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=batch_general_video_list, campaign_name=app_version + '--优选广告位-' + campaign_group_name_place, group_name=app_version + '-优选广告位-不限-自定义-' + campaign_group_name_place, ai_strategy_remark="按小时补充创意-"+campaign_group_name_place, unit_type=4, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) video_cnt += batch_num logger.info("account_id=%s, app_id=%s, %s补充创意-通用视频,ai策略的返回信息:%s" % (account_id, app_id, campaign_group_name_place, request)) else: logger.info("account_id=%s, %s补充创意: 返回的视频中没有通用视频" % (account_id, campaign_group_name_place)) self.write(json.dumps({"message": "account_id=%s, %s补充创意请求已走完!" % (account_id, campaign_group_name_place)})) self.flush() else: # 获取当天计划中包含“campaign_group_name_place”计划的,计划id sql = """select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s and put_create_time <= '%s' and put_create_time >= '%s' """ % (account_id, (datetime.datetime.now()).strftime('%Y-%m-%d %H:%M:%S'), (datetime.datetime.now()).strftime('%Y-%m-%d')) df = pd.read_sql(sql, engine) for row in df.itertuples(): if campaign_group_name_place in getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.info("account_id = %s, 没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_list['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="按小时补充创意-"+campaign_group_name_place, name_replace=campaign_group_name_place) request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, %s补充创意,ai策略的返回信息:%s" % (account_id, campaign_group_name_place, request)) self.write(json.dumps(request)) self.flush() else: logger.info("account_id=%s, 按小时补充创意: 没有成功获取到视频!" % account_id) self.write(json.dumps({"message": "account_id=%s, 按小时补充创意: 没有成功获取到视频!" % account_id})) self.flush() else: logger.info("account_id=%s, 按小时补充创意: 不需要补充!" % account_id) self.write(json.dumps({"message": "account_id=%s, 按小时补充创意: 不需要补充!" % account_id})) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class AiProgramCreativeHighQualityMaterial(tornado.web.RequestHandler): """ 使用历史跑量素材创建程序化创意 跑量素材150个(近14天的素材) 计划名称--跑量素材程序化 timeTask: 每天上午10点执行 --(0 0 10 * * ?) ---------------------------------------- 拼装参数和写入数据库时,需要修改名称(程序化与非程序化不一致): creative_name --> package_name(程序化创意名称) action_bar_text --> action_bar(行动号召按钮) description --> captions(作品广告语) click_track_url --> click_url 写入数据库时: image_md5s --> cover_image_tokens 程序化与非程序化的逻辑不一样: 需要5个视频4张封面一组,拼装程序化创意参数需要单独抽取一个方法出来 所有操作记录表中,添加status信息(已提交,成功,失败),和回调时的 message信息 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") account_id = data.get('account_id') campaign_id = data.get('campaign_id') logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: sql = """select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1 """ % account_id df = pd.read_sql(sql, engine) channel_type = df['channel_type'].values[0] if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info video_list = get_top_video_info(channel_type=channel_type, account_id=account_id, cnt=multi_app_id_programme_high_quality_video_cnt, start_time=(datetime.datetime.now() + datetime.timedelta( days=-high_quality_video_days)).strftime("%Y-%m-%d %H:%M:%S"), end_time=(datetime.datetime.now() + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), url=get_high_quality_video_url) # 单独处理返回的时候中,有通用视频,需要单独处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", if video_list['code'] == 0: video_df = pd.DataFrame(video_list['data']) # 1-特定视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化高质量素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version + '--优选广告位-' + '程序化高质量素材', group_name=app_version + '-优选广告位-不限-程序化-' + '高质量素材', ai_strategy_remark="程序化高质量素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 程序化高质量素材-特定视频,ai策略的返回信息:%s" % (account_id, app_id, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] general_video_list = [] for k, v in general_video_df.T.to_dict().items(): general_video_list.append(v) if general_video_list: app_version_list = list(app_dict) random.shuffle(app_version_list) # 每5个通用视频创建一个程序化广告组 batch_num = 5 app_id_cnt = 0 for i in range(0, len(general_video_list), 5): if i + 5 < len(general_video_list): batch_general_video = general_video_list[i: i + batch_num] app_version = app_version_list[app_id_cnt] app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化高质量素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=batch_general_video, campaign_name=app_version + '--优选广告位-' + '程序化高质量素材', group_name=app_version + '-优选广告位-不限-程序化-' + '高质量素材', ai_strategy_remark="程序化高质量素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) app_id_cnt += 1 logger.info("account_id=%s, app_id=%s, 程序化高质量素材-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: batch_general_video = general_video_list[i: len(general_video_list)] app_version = app_version_list[app_id_cnt] app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化高质量素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=batch_general_video, campaign_name=app_version + '--优选广告位-' + '程序化高质量素材', group_name=app_version + '-优选广告位-不限-程序化-' + '高质量素材', ai_strategy_remark="程序化高质量素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 程序化高质量素材-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: logger.info("account_id=%s, 程序化高质量素材: 返回的视频中没有通用视频" % account_id) self.write(json.dumps({"message": "account_id=%s, 程序化高质量素材请求已走完!" % account_id})) self.flush() else: video_info = get_top_video_info(channel_type=channel_type, account_id=account_id, cnt=programme_high_quality_video_cnt, start_time=(datetime.datetime.now() + datetime.timedelta( days=-high_quality_video_days)).strftime("%Y-%m-%d %H:%M:%S"), end_time=(datetime.datetime.now() + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), url=get_high_quality_video_url) if video_info['code'] != 0: logger.info("account_id = %s, 程序化高质量素材,没有获取到视频,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "没有获取到视频,不发送ai策略请求!"})) self.flush() else: campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.info("account_id = %s, 没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!" % account_id) self.write(json.dumps({"account_id": account_id, "message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_info['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="程序化高质量素材", name_replace="程序化高质量素材", unit_type=7) request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, 程序化高质量素材, ai策略的返回信息:%s" % (account_id, request)) self.write(json.dumps(request)) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush() class AiProgramCreativeNewMaterial(tornado.web.RequestHandler): """ 使用N天内 创意数关联小于M个(如当天创意关联数小于10的,150个素材)的素材创建程序化创意 计划名称--新素材程序化 timeTask: 每天上午10点执行 --(0 0 10 * * ?) """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") account_id = data.get('account_id') campaign_id = data.get('campaign_id') test_video_cnt = data.get('test_video_cnt', None) logger.info("账户信息=%s, 当前进程=%s, 当前进程的主进程=%s, raw data from request is %s" % (account_id, os.getpid(), os.getppid(), data)) try: sql = """select account_id, single_appid, app_id_array, general_track from ctop_ai_kuaishou_advertiser_strategy where account_id=%s order by create_time desc limit 1 """ % account_id df = pd.read_sql(sql, engine) channel_type = df['channel_type'].values[0] if df['general_track'].values[0] == 0: app_id_array = eval(df['app_id_array'].values[0]) app_list = get_app_list(account_id=account_id, url=get_app_list_url) # app_dict key = ("appVersion": "ksjyrlyn-人类一脑涂地大败") , value = 完整的app_id信息 app_dict = {} for app_info in app_list['data']: if app_info['appId'] in app_id_array: app_dict[app_info['appVersion']] = app_info video_list = get_history_video_info(channel_type=channel_type, account_id=account_id, url=get_history_video_url, start_time=(datetime.datetime.now() + datetime.timedelta(days=-7)). strftime("%Y-%m-%d %H:%M:%S"), end_time=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") if test_video_cnt else (datetime.datetime.now() + datetime.timedelta(days=-1)).strftime("%Y-%m-%d %H:%M:%S"), video_cnt=test_video_cnt if test_video_cnt else multi_app_id_programme_new_video_cnt, related_creative_max_cnt=related_creative_max_cnt) # 单独处理返回的时候中,有通用视频,需要单独处理 # "photo_name": "all", # "photo_name": "ksjytdrca-糖豆人冲啊", if video_list['code'] == 0: video_df = pd.DataFrame(video_list['data']) # 1-特定视频的创建 special_video_df = video_df[video_df.photo_name != 'all'] for photo_name in special_video_df['photo_name'].unique(): if photo_name in app_dict.keys(): campaign_id = None app_id = app_dict[photo_name].get('appId', 'error') app_version = app_dict[photo_name].get('appVersion', 'error') track_url = app_dict[photo_name].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化上新素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break app_id_video_df = special_video_df[special_video_df.photo_name == photo_name] app_id_video_list = [] for k, v in app_id_video_df.T.to_dict().items(): app_id_video_list.append(v) request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=app_id_video_list, campaign_name=app_version + '--优选广告位-' + '程序化上新素材', group_name=app_version + '-优选广告位-不限-程序化-' + '上新素材', ai_strategy_remark="程序化上新素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 程序化上新素材-特定视频,ai策略的返回信息:%s" % (account_id, app_id, request)) # 2-通用视频的创建,random.shuffle 选择 app_id general_video_df = video_df[video_df.photo_name == 'all'] general_video_list = [] for k, v in general_video_df.T.to_dict().items(): general_video_list.append(v) if general_video_list: app_version_list = list(app_dict) random.shuffle(app_version_list) # 每5个通用视频创建一个程序化广告组 batch_num = 5 app_id_cnt = 0 for i in range(0, len(general_video_list), 5): if i + 5 < len(general_video_list): batch_general_video = general_video_list[i: i + batch_num] app_version = app_version_list[app_id_cnt] app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化上新素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=batch_general_video, campaign_name=app_version + '--优选广告位-' + '程序化上新素材', group_name=app_version + '-优选广告位-不限-程序化-' + '上新素材', ai_strategy_remark="程序化上新素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) app_id_cnt += 1 logger.info("account_id=%s, app_id=%s, 程序化上新素材-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: batch_general_video = general_video_list[i: len(general_video_list)] app_version = app_version_list[app_id_cnt] app_id = app_dict[app_version].get('appId', 'error') track_url = app_dict[app_version].get('trackUrl', 'error') sql = """ select campaign_id, campaign_name from ctop_kuaishou_campaign where account_id = %s """ % account_id campaign_df = pd.read_sql(sql, engine) for row in campaign_df.itertuples(): if (app_version + '--优选广告位-' + '程序化上新素材') == getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=batch_general_video, campaign_name=app_version + '--优选广告位-' + '程序化上新素材', group_name=app_version + '-优选广告位-不限-程序化-' + '上新素材', ai_strategy_remark="程序化上新素材", unit_type=7, app_id=app_id, click_track_url=track_url, creative_name=app_version) request = ai_strategy_request_parse(request_data) logger.info("account_id=%s, app_id=%s, 程序化上新素材-通用视频,ai策略的返回信息:%s" % (account_id, app_id, request)) else: logger.info("account_id=%s, 程序化上新素材: 返回的视频中没有通用视频" % account_id) self.write(json.dumps({"message": "account_id=%s, 程序化上新素材请求已走完!" % account_id})) self.flush() else: video_info = get_history_video_info(channel_type=channel_type, account_id=account_id, url=get_history_video_url, start_time=(datetime.datetime.now() + datetime.timedelta(days=-7)). strftime("%Y-%m-%d %H:%M:%S"), end_time=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") if test_video_cnt else (datetime.datetime.now() + datetime.timedelta(days=-1)). strftime("%Y-%m-%d %H:%M:%S"), video_cnt=test_video_cnt if test_video_cnt else programme_new_video_cnt, related_creative_max_cnt=related_creative_max_cnt) if video_info['code'] != 0: self.write(json.dumps({"account_id": account_id, "message": "程序化上新素材,没有获取到视频,不发送ai策略请求!"})) self.flush() else: campaign_name, group_name = get_campaign_and_group_name_rule(account_id) if (campaign_name is None) or (group_name is None): logger.info("没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!") self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id=account_id, campaign_id=campaign_id, video_info=video_info['data'], campaign_name=campaign_name, group_name=group_name, ai_strategy_remark="程序化上新素材", name_replace="程序化上新素材", unit_type=7) request = ai_strategy_request_parse(request_data) logger.info("account_id = %s, 程序化上新素材,ai策略的返回信息:%s" % (account_id, request)) self.write(json.dumps(request)) self.flush() except Exception: logger.error("account_id = %s, traceback is %s" % (account_id, traceback.format_exc())) self.write(json.dumps({"account_id": account_id, "message": traceback.format_exc(), "code": -1})) self.flush()