1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677 |
- 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()} ******")
|