import datetime import traceback from logging.handlers import TimedRotatingFileHandler from utils.BaseClass import * from utils.CommonFunction import * from utils.ConstantConfig import * from utils.LogConfig import * from ai_strategy_request_func import ai_strategy_request_parse log_handler = TimedRotatingFileHandler("logs/ai_auto_create/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个素材) -- 测试通过 在AD后台创建成功,检查AD信息和数据库表 没有问题 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") logger.info("raw data from request is %s" % data) account_id = data.get('account_id') campaign_id = data.get('campaign_id') try: video_info = get_missing_video_info(account_id, missing_video_cnt, get_missing_video_url, missing_video_start_time, missing_video_end_time) if video_info is None: self.write(json.dumps({"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): self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id, campaign_id, video_info, campaign_name, group_name, "补充遗漏素材", "补充遗漏素材") request = ai_strategy_request_parse(request_data) logger.info("ai策略的返回信息:%s" % request) self.write(json.dumps(request)) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) 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 ***************************************") logger.info("raw data from request is %s" % data) account_id = data.get('account_id') campaign_id = data.get('campaign_id') try: video_info = get_new_video_info(account_id, get_new_video_url) if video_info is None: self.write(json.dumps({"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): self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id, campaign_id, video_info, campaign_name, group_name, "自动上新", "上新") request = ai_strategy_request_parse(request_data) logger.info("ai策略的返回信息:%s" % request) self.write(json.dumps(request)) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) 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 ***************************************") logger.info("raw data from request is %s" % data) account_id = data.get('account_id') campaign_id = data.get('campaign_id') try: video_info = get_top_video_info(account_id, high_quality_video_cnt, (datetime.datetime.now() + datetime.timedelta(days=-high_quality_video_days)).strftime("%Y-%m-%d"), str(datetime.date.today()), get_high_quality_video_url) if video_info is None: self.write(json.dumps({"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): self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id, campaign_id, video_info, campaign_name, group_name, "高质量素材", "高质量素材") request = ai_strategy_request_parse(request_data) logger.info("ai策略的返回信息:%s" % request) self.write(json.dumps(request)) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) self.flush() class AiCheckAndUpTOFullCreative(tornado.web.RequestHandler): """ 每天22:00 调用该任务,检查当天创意是否创建达到上限,否则使用历史高质量素材填满 -- 测试通过 在AD后台创建成功,检查AD信息和数据库表 没有问题 """ def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') logger.info("*************************************** NEW REQUEST ***************************************") logger.info("raw data from request is %s" % data) account_id = data.get('account_id') campaign_id = data.get('campaign_id') try: 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) for row in df.itertuples(): if '高质量素材' in getattr(row, 'campaign_name'): campaign_id = getattr(row, 'campaign_id') break # 2-1 从客户策略表中获取该账户下 每个素材搭配的封面个数, 来计算需要补充的素材个数 sql = """ select image_cnt from ctop_ai_kuaishou_advertiser_strategy where account_id = %s and status = 1 limit 1 """ % account_id df = pd.read_sql(sql, engine) image_cnt = df['image_cnt'].values[0] # 3-1 从 ctop_kuaishou_creative 获取当天该账户到此刻 已经创建的创意个数 video_cnt = 0 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, end_time, start_time) df = pd.read_sql(sql, engine) creative_cnt = df['creative_cnt'].values[0] if creative_cnt >= 2000: video_cnt = 0 else: video_cnt = (2000 - creative_cnt) // image_cnt # 4-1 获取视频信息 if video_cnt == 0: logger.info("创意已经建满,不需要补充素材!") self.write(json.dumps("message: 创意已经建满,不需要补充素材!")) self.flush() else: now = datetime.datetime.now() video_info = get_top_video_info(account_id, video_cnt, (now + datetime.timedelta(days=-2*high_quality_video_days)).strftime("%Y-%m-%d"), (now + datetime.timedelta(days=-high_quality_video_days)).strftime("%Y-%m-%d"), get_high_quality_video_url) if video_info is None: self.write(json.dumps({"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): self.write(json.dumps({"message": "没有获取到广告计划命名规则 或 广告组命名规则,不发送ai策略请求!"})) self.flush() else: request_data = get_request_data(account_id, campaign_id, video_info, campaign_name, group_name, "高质量素材", "高质量素材") request = ai_strategy_request_parse(request_data) logger.info("ai策略的返回信息:%s" % request) self.write(json.dumps(request)) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) self.flush() def get_request_data(account_id, campaign_id, video_info, campaign_name, group_name,ai_strategy_remark, name_replace): res = { "video": video_info, "operation_type": 1, "account_id": account_id, "ai_strategy_remark": ai_strategy_remark, "campaign_info": {"campaign_id": campaign_id, "campaign_name": campaign_name.replace('自定义', name_replace) + str(datetime.datetime.now())}, 'group_info': {'group_name': group_name.replace('自定义', name_replace) + str(datetime.datetime.now()), 'begin_time': str(datetime.date.today())}, 'creative_info': {'is_sticky': 0} } return res