time_task_create_ad_by_target.py 3.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. from concurrent.futures import ThreadPoolExecutor
  2. import requests
  3. import datetime
  4. import json
  5. import pandas as pd
  6. import uuid
  7. import traceback
  8. import yaml
  9. from utils.commonFunc import get_db_engine
  10. from config.url import target_combine_url
  11. import logging
  12. from concurrent_log import ConcurrentTimedRotatingFileHandler
  13. log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%m/%d/%Y %I:%M:%S %p')
  14. log_handler = ConcurrentTimedRotatingFileHandler("/data/pythonProject/ai_target/logs/time_task_create_ad_by_target.log", when="midnight", backupCount=100)
  15. log_handler.setFormatter(log_formatter)
  16. logger = logging.getLogger('time_task_create_ad_by_target_logger')
  17. logger.addHandler(log_handler)
  18. logger.setLevel(logging.DEBUG)
  19. print('id of time_task_create_ad_by_target_logger %s' % id(logger))
  20. with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:
  21. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  22. product_db_engine = get_db_engine(config['productDB'])
  23. project_ids = config['projectId'] # 目前支持优质定向功能的项目列表
  24. project_ids = project_ids if len(project_ids) > 1 else project_ids*2
  25. def send_request(args):
  26. try:
  27. request_data = json.dumps({"account_id": args[0], "project_id": args[1]})
  28. request = requests.post(target_combine_url, request_data)
  29. result = json.loads(request.text)
  30. logger.info("account_id= %s, project_id=%s, result=%s" % (args[0], args[1], result))
  31. return result
  32. except Exception:
  33. logger.error("account_id= %s, project_id=%s, traceback=%s" % (args[0], args[1], traceback.format_exc()))
  34. return "error"
  35. # 0、 打印该定时任务被调用的时间
  36. unique_uuid = str(uuid.uuid4())
  37. logger.info("***********优质定向创建-Start, unique_uuid = %s , time = %s **************" % (unique_uuid, datetime.datetime.now()))
  38. # 1、获取配置了【优质定向】的 account_id 及其 project_id
  39. # system_optimization 是开启系统优选 0否 1是
  40. sql = """
  41. select t1.account_id, t2.project_id
  42. from
  43. (select account_id from ctop_ai_kuaishou_advertiser_strategy where status = 1
  44. and system_optimization = 1
  45. ) t1
  46. left join
  47. (select project_id,account_id from ctop_user_allocation where project_id in %s) t2
  48. on t1.account_id = t2.account_id
  49. """ % (tuple(project_ids),)
  50. account_df = pd.read_sql(sql, product_db_engine)
  51. account_df.drop_duplicates('account_id', keep='first', inplace=True)
  52. acc_list = [(int(item['account_id']), int(item['project_id'])) for index, item in account_df.iterrows()]
  53. # TODO 注释测试代码 for test
  54. # acc_list = [(10456827, 458), (9774238, 458)]
  55. # acc_list = [(10456827, 458), (9774238, 458)]
  56. # acc_list = [(10456827, 458)]
  57. # acc_list = [(9774099, 458)]
  58. # acc_list = [(9774238, 458)]
  59. # acc_list = [(9767003, 458)]
  60. # 2、 5个账户一组进行发送请求
  61. batch_num = 5
  62. for i in range(0, len(acc_list), batch_num):
  63. if i + batch_num < len(acc_list):
  64. logger.info("time is %s, batch task is %s" % (datetime.datetime.now(), acc_list[i:i+batch_num]))
  65. with ThreadPoolExecutor(max_workers=batch_num) as pool:
  66. results = pool.map(send_request, tuple(acc_list[i: i+batch_num]))
  67. else:
  68. with ThreadPoolExecutor(max_workers=batch_num) as pool:
  69. logger.info("time is %s, batch task is %s" % (datetime.datetime.now(), acc_list[i: len(acc_list)]))
  70. results = pool.map(send_request, tuple(acc_list[i: len(acc_list)]))
  71. # 3、 打印该定时任务结束的时间
  72. logger.info("***********优质定向创建-End, unique_uuid = %s , time = %s **************" % (unique_uuid, datetime.datetime.now()))