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 with open('../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'] # 目前支持优质定向功能的项目列表 def send_request(args): request = None try: request_data = json.dumps({"account_id": args[0], "project_id": args[1]}) print('target_combine_url', target_combine_url) request = requests.post(target_combine_url, request_data) return json.loads(request.text) except Exception: return "account_id = %s, request = %s, traceback = %s " % (args[0], request, traceback.format_exc()) # 0、 打印该定时任务被调用的时间 unique_uuid = str(uuid.uuid4()) print("***********", "优质定向创建-Start, unique_uuid = ", unique_uuid, datetime.datetime.now(), "**************") # 1、获取配置了【优质定向】的 account_id 及其 project_id # TODO 读取是否配置优质定向的字段,进行判断 sql = """ select t1.account_id, t2.project_id from (select account_id from ctop_ai_kuaishou_advertiser_strategy where status = 1 -- and open_program_create in (0,2) ) 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 = [(item['account_id'], item['project_id']) for index, item in account_df.iterrows()] acc_list = [(9774238, 458)] # 2、 5个账户一组进行发送请求 batch_num = 5 for i in range(0, len(acc_list), batch_num): if i + batch_num < len(acc_list): print(datetime.datetime.now(), tuple(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])) for r in results: print('res = %s' % r) else: with ThreadPoolExecutor(max_workers=batch_num) as pool: print(datetime.datetime.now(), acc_list[i: len(acc_list)]) results = pool.map(send_request, tuple(acc_list[i: len(acc_list)])) for r in results: print('res = %s' % r) # 3、 打印该定时任务结束的时间 print("***********", "优质定向创建-End, unique_uuid = ", unique_uuid, datetime.datetime.now(), "**************")