1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071 |
- import requests
- from utils.UrlConfig import headers
- import json
- import pandas as pd
- from utils.DataBaseConfig import *
- from logging.handlers import TimedRotatingFileHandler
- from utils.LogConfig 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_common_func_logger')
- logger.addHandler(log_handler)
- logger.setLevel(logging.DEBUG)
- print('id of ai_common_func_logger %s' % id(logger))
- logger.info("ai_common_func started!")
- def get_lower_case_name(name):
- lst = []
- for index, char in enumerate(name):
- if char.isupper() and index != 0:
- lst.append("_")
- lst.append(char)
- return "".join(lst).lower()
- def get_new_video_info(account_id, url):
- request_data = {"accountId": account_id}
- request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
- request = json.loads(request)
- if request['code'] != 0:
- logger.info('the res of get_new_video_info is %s' % request)
- return None
- return request['data']
- def get_missing_video_info(account_id, cnt, url, start_time, end_time):
- request_data = {"accountId": account_id, "videoCnt": cnt, "startTime": start_time, "endTime": end_time}
- request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
- request = json.loads(request)
- if request['code'] != 0:
- logger.info('the res of get_missing_video_info is %s' % request)
- return None
- return request['data']
- def get_top_video_info(account_id, cnt, start_time, end_time, url):
- request_data = {"accountId": account_id, "num": cnt, "startTime": start_time, "endTime": end_time}
- request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
- request = json.loads(request)
- if request['code'] != 0:
- logger.info('the res of get_top_video_info is %s' % request)
- return None
- return request['data']
- def get_campaign_and_group_name_rule(account_id):
- sql = """select campaign_name, group_name
- from ctop_ai_kuaishou_advertiser_strategy
- where account_id = %s
- and status = 1
- limit 1
- """ % account_id
- df = pd.read_sql(sql, engine)
- if df.empty:
- return None, None
- campaign_name = df['campaign_name'].values[0]
- group_name = df['group_name'].values[0]
- return campaign_name, group_name
|