import datetime import os import sys import uuid from concurrent.futures import ThreadPoolExecutor import pandas as pd import yaml from loguru import logger curr_path = os.path.abspath(os.path.dirname(__file__)) project_root_path = curr_path[:curr_path.find("ai_target") + len("ai_target")] sys.path.append(project_root_path) from config.url_and_db import jeecg_product_db from ai_target_create_ads import ai_target_combine_create logger.remove() # 删去 import logger之后自动产生的handler,不删除的话会出现重复输出的现象 logger.add("/data/pythonProject/ai_target/logs/time_task_create_ad_by_target.{time:YYYY-MM-DD}.log", rotation="00:00", format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}", level="INFO") with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f: config = yaml.load(f.read(), Loader=yaml.FullLoader) project_ids = config['projectId'] # 目前支持优质定向功能的项目列表 project_ids = project_ids if len(project_ids) > 1 else project_ids * 2 # 0、 打印该定时任务被调用的时间 unique_uuid = str(uuid.uuid4()) logger.info(f"*******优质定向创建-Start, unique_uuid = {unique_uuid} , {datetime.datetime.now()} ********") # 1、获取配置了【优质定向】的 account_id 及其 project_id # 旧版自动投放策略: ctop_ai_kuaishou_advertiser_strategy # 新版自动投放策略: ctop_ai_kuaishou_account_auto_strategy # system_optimization 是开启系统优选 0否 1是 sql = f"select t1.account_id, t2.project_id from " \ f"(select account_id from ctop_ai_kuaishou_advertiser_strategy where status = 1 and system_optimization = 1 " \ f"union select account_id from ctop_ai_kuaishou_account_auto_strategy where status = 1 and system_optimization = 1) t1 " \ f"left join " \ f"(select project_id, account_id from ctop_user_allocation where project_id in {tuple(project_ids)} and account_status = 0) t2 " \ f"on t1.account_id = t2.account_id " \ f"where t2.project_id is not null" account_df = pd.read_sql(sql, jeecg_product_db) 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 = [(11041466, 458)] # 李琳的账户 # acc_list = [(10967859, 458)] # 郭浩的账户 def task(args): account_id = args[0] project_id = args[1] ai_target_combine_create(account_id, project_id) # 2、 5个账户一组进行发送请求 batch_num = 5 for i in range(0, len(acc_list), batch_num): if i + batch_num < len(acc_list): logger.info(f"{datetime.datetime.now()}, batch task is {acc_list[i:i + batch_num]}") with ThreadPoolExecutor(max_workers=batch_num) as pool: results = pool.map(task, tuple(acc_list[i: i + batch_num])) else: logger.info(f"{datetime.datetime.now()}, batch task is {acc_list[i: len(acc_list)]}") with ThreadPoolExecutor(max_workers=batch_num) as pool: results = pool.map(task, tuple(acc_list[i: len(acc_list)])) # 3、 打印该定时任务结束的时间 logger.info(f"******优质定向创建-End, unique_uuid = {unique_uuid} , {datetime.datetime.now()} ******")