1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- from concurrent.futures import ThreadPoolExecutor
- import requests
- import datetime
- import json
- import pandas as pd
- import uuid
- import traceback
- import yaml
- from utils.commonFunc import get_db_engine
- from config.url import target_combine_url
- import logging
- from concurrent_log import ConcurrentTimedRotatingFileHandler
- log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%m/%d/%Y %I:%M:%S %p')
- log_handler = ConcurrentTimedRotatingFileHandler("/data/pythonProject/ai_target/logs/time_task_create_ad_by_target.log", when="midnight", backupCount=100)
- log_handler.setFormatter(log_formatter)
- logger = logging.getLogger('time_task_create_ad_by_target_logger')
- logger.addHandler(log_handler)
- logger.setLevel(logging.DEBUG)
- print('id of time_task_create_ad_by_target_logger %s' % id(logger))
- with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:
- config = yaml.load(f.read(), Loader=yaml.FullLoader)
- product_db_engine = get_db_engine(config['productDB'])
- project_ids = config['projectId'] # 目前支持优质定向功能的项目列表
- project_ids = project_ids if len(project_ids) > 1 else project_ids*2
- def send_request(args):
- try:
- request_data = json.dumps({"account_id": args[0], "project_id": args[1]})
- request = requests.post(target_combine_url, request_data)
- result = json.loads(request.text)
- logger.info("account_id= %s, project_id=%s, result=%s" % (args[0], args[1], result))
- return result
- except Exception:
- logger.error("account_id= %s, project_id=%s, traceback=%s" % (args[0], args[1], traceback.format_exc()))
- return "error"
- # 0、 打印该定时任务被调用的时间
- unique_uuid = str(uuid.uuid4())
- logger.info("***********优质定向创建-Start, unique_uuid = %s , time = %s **************" % (unique_uuid, datetime.datetime.now()))
- # 1、获取配置了【优质定向】的 account_id 及其 project_id
- # system_optimization 是开启系统优选 0否 1是
- sql = """
- select t1.account_id, t2.project_id
- from
- (select account_id from ctop_ai_kuaishou_advertiser_strategy where status = 1
- and system_optimization = 1
- ) t1
- left join
- (select project_id,account_id from ctop_user_allocation where project_id in %s) t2
- on t1.account_id = t2.account_id
- """ % (tuple(project_ids),)
- account_df = pd.read_sql(sql, product_db_engine)
- account_df.drop_duplicates('account_id', keep='first', inplace=True)
- acc_list = [(int(item['account_id']), int(item['project_id'])) for index, item in account_df.iterrows()]
- # TODO 注释测试代码 for test
- # acc_list = [(10456827, 458), (9774238, 458)]
- # acc_list = [(10456827, 458), (9774238, 458)]
- # acc_list = [(10456827, 458)]
- # acc_list = [(9774099, 458)]
- # acc_list = [(9774238, 458)]
- # acc_list = [(9767003, 458)]
- # 2、 5个账户一组进行发送请求
- batch_num = 5
- for i in range(0, len(acc_list), batch_num):
- if i + batch_num < len(acc_list):
- logger.info("time is %s, batch task is %s" % (datetime.datetime.now(), acc_list[i:i+batch_num]))
- with ThreadPoolExecutor(max_workers=batch_num) as pool:
- results = pool.map(send_request, tuple(acc_list[i: i+batch_num]))
- else:
- with ThreadPoolExecutor(max_workers=batch_num) as pool:
- logger.info("time is %s, batch task is %s" % (datetime.datetime.now(), acc_list[i: len(acc_list)]))
- results = pool.map(send_request, tuple(acc_list[i: len(acc_list)]))
- # 3、 打印该定时任务结束的时间
- logger.info("***********优质定向创建-End, unique_uuid = %s , time = %s **************" % (unique_uuid, datetime.datetime.now()))
|