time_task_create_ad_by_target.py 2.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. import datetime
  2. import uuid
  3. from concurrent.futures import ThreadPoolExecutor
  4. import pandas as pd
  5. import yaml
  6. from loguru import logger
  7. from config.url_and_db import jeecg_product_db
  8. from ai_target_create_ads import ai_target_combine_create
  9. logger.remove() # 删去 import logger之后自动产生的handler,不删除的话会出现重复输出的现象
  10. logger.add("/data/pythonProject/ai_target/logs/time_task_create_ad_by_target.{time:YYYY-MM-DD}.log",
  11. rotation="00:00",
  12. format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}",
  13. level="INFO")
  14. with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:
  15. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  16. project_ids = config['projectId'] # 目前支持优质定向功能的项目列表
  17. project_ids = project_ids if len(project_ids) > 1 else project_ids * 2
  18. # 0、 打印该定时任务被调用的时间
  19. unique_uuid = str(uuid.uuid4())
  20. logger.info(f"*******优质定向创建-Start, unique_uuid = {unique_uuid} , {datetime.datetime.now()} ********")
  21. # 1、获取配置了【优质定向】的 account_id 及其 project_id
  22. # system_optimization 是开启系统优选 0否 1是
  23. sql = f"select t1.account_id, t2.project_id " \
  24. f"from (select account_id from ctop_ai_kuaishou_advertiser_strategy where status = 1 and system_optimization = 1) t1 " \
  25. f"left join (select project_id, account_id from ctop_user_allocation where project_id in {tuple(project_ids)} and account_status = 0) t2 " \
  26. f"on t1.account_id = t2.account_id " \
  27. f"where t2.account_id is not null"
  28. account_df = pd.read_sql(sql, jeecg_product_db)
  29. account_df.drop_duplicates('account_id', keep='first', inplace=True)
  30. acc_list = [(int(item['account_id']), int(item['project_id'])) for index, item in account_df.iterrows()]
  31. # TODO 注释测试代码 for test
  32. # acc_list = [(10456827, 458), (9774238, 458)]
  33. # acc_list = [(10456827, 458), (9774238, 458)]
  34. # acc_list = [(11041466, 458)] # 李琳的账户
  35. acc_list = [(10967859, 458)] # 郭浩的账户
  36. def task(args):
  37. account_id = args[0]
  38. project_id = args[1]
  39. ai_target_combine_create(account_id, project_id)
  40. # 2、 5个账户一组进行发送请求
  41. batch_num = 5
  42. for i in range(0, len(acc_list), batch_num):
  43. if i + batch_num < len(acc_list):
  44. logger.info(f"{datetime.datetime.now()}, batch task is {acc_list[i:i + batch_num]}")
  45. with ThreadPoolExecutor(max_workers=batch_num) as pool:
  46. results = pool.map(task, tuple(acc_list[i: i + batch_num]))
  47. else:
  48. logger.info(f"{datetime.datetime.now()}, batch task is {acc_list[i: len(acc_list)]}")
  49. with ThreadPoolExecutor(max_workers=batch_num) as pool:
  50. results = pool.map(task, tuple(acc_list[i: len(acc_list)]))
  51. # 3、 打印该定时任务结束的时间
  52. logger.info(f"******优质定向创建-End, unique_uuid = {unique_uuid} , {datetime.datetime.now()} ******")