import datetime import json import traceback import pandas as pd import requests import tornado.web from db_config import * class AiAutoCreative(tornado.web.RequestHandler): """ 定时任务调用,上新素材自动创建 """ def initialize(self, logger): self.logger = logger def post(self): data = self.request.body data = str(data, 'utf8') data = json.loads(data, encoding='utf8') self.logger.info("*************************************** NEW REQUEST ***************************************") self.logger.info("raw data from request is %s" % data) account_id = data.get('account_id') campaign_id = data.get('campaign_id') try: inst = GetRequestData(account_id, campaign_id) if inst.data is not None: request = requests.post(ai_strategy_request_url, inst.data) self.logger.info("ai策略的返回信息:%s" % request.text) self.write(request.text) self.flush() else: self.write(json.dumps({"message": "没有获取到视频新,不发送ai策略请求!"})) self.flush() except Exception: self.logger.error(traceback.format_exc()) self.write(json.dumps(traceback.format_exc())) self.flush() class GetRequestData(object): def __init__(self, account_id, campaign_id): self.video = None self.account_id = account_id self.campaign_id = campaign_id self.data = None self.campaign_name = None self.group_name = None self.get_video_info() self.get_campaign_and_group_name_rule() self.get_request_data() def get_video_info(self): video_request_data = {"accountId": self.account_id} request = requests.post(get_new_video_url, headers=headers, data=json.JSONEncoder().encode(video_request_data)) self.video = request def get_campaign_and_group_name_rule(self): if self.video['code'] == 0: sql = """ select campaign_name, unit_name from ctop_ai_kuaishou_advertiser_strategy where account_id = %d and status = 1 limit 1 """ % self.account_id df = pd.read_sql(sql, engine) self.campaign_name = df['campaign_name'].values[0] self.group_name = df['unit_name'].values[0] def get_request_data(self): if self.video['code'] == 0: self.data = { "video": self.video['data'], "operation_type": 1, "account_id": self.account_id, "ai_strategy_remark": "自动上新", "campaign_info": { "campaign_id": self.campaign_id, "campaign_name": self.campaign_name.replace('自定义', '上新') + str(datetime.datetime.now()) }, 'group_info': { 'group_name': self.group_name.replace('自定义', '上新') + str(datetime.datetime.now()), 'begin_time': str(datetime.date.today()) }, 'creative_info': { 'is_sticky': 0 } }