time_task_create_ad_by_target.py 3.4 KB

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