import datetime import json import traceback import pandas as pd import requests import tornado.web from logging.handlers import TimedRotatingFileHandler from utils.DataBaseConfig import * from utils.BaseClass import * from utils.DataBaseConfig import * from utils.LogConfig import * from utils.UrlConfig import * from utils.CommonFunction import * from utils.ConstantConfig import * log_handler = TimedRotatingFileHandler("logs/ai_auto_create/ai_auto_create.log", when="midnight", backupCount=100) log_handler.setFormatter(log_formatter) logger = logging.getLogger('ai_auto_create_logger') logger.addHandler(log_handler) logger.setLevel(logging.DEBUG) print('id of logger %s' % id(logger)) class AiHistoricalMissingMaterial(tornado.web.RequestHandler): 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_video_info(account_id, historical_missing_video_cnt, get_historical_missing_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 = { "video": video_info, "operation_type": 1, "account_id": account_id, "ai_strategy_remark": "补充遗漏素材", "campaign_info": {"campaign_id": campaign_id, "campaign_name": campaign_name.replace('自定义', '补充遗漏素材') + str(datetime.datetime.now())}, 'group_info': {'group_name': group_name.replace('自定义', '补充遗漏素材') + str(datetime.datetime.now()), 'begin_time': str(datetime.date.today())}, 'creative_info': {'is_sticky': 0} } request = requests.post(ai_strategy_request_url, request_data) logger.info("ai策略的返回信息:%s" % request.text) self.write(request.text) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) self.flush() class AiAutoCreative(tornado.web.RequestHandler): 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_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 = requests.post(ai_strategy_request_url, request_data) logger.info("ai策略的返回信息:%s" % request.text) self.write(request.text) self.flush() except Exception: logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) self.flush() class AiHistoricalHighQualityMaterial(tornado.web.RequestHandler): 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_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 = requests.post(ai_strategy_request_url, request_data) logger.info("ai策略的返回信息:%s" % request.text) self.write(request.text) 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