CommonFunction.py 2.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. import requests
  2. from utils.UrlConfig import headers
  3. import json
  4. import pandas as pd
  5. from utils.DataBaseConfig import *
  6. from logging.handlers import TimedRotatingFileHandler
  7. from utils.LogConfig import *
  8. log_handler = TimedRotatingFileHandler("logs/ai_auto_create/ai_auto_create.log",
  9. when="midnight", backupCount=100)
  10. log_handler.setFormatter(log_formatter)
  11. logger = logging.getLogger('ai_common_func_logger')
  12. logger.addHandler(log_handler)
  13. logger.setLevel(logging.DEBUG)
  14. print('id of ai_common_func_logger %s' % id(logger))
  15. logger.info("ai_common_func started!")
  16. def get_lower_case_name(name):
  17. lst = []
  18. for index, char in enumerate(name):
  19. if char.isupper() and index != 0:
  20. lst.append("_")
  21. lst.append(char)
  22. return "".join(lst).lower()
  23. def get_new_video_info(account_id, url):
  24. request_data = {"accountId": account_id}
  25. request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
  26. request = json.loads(request)
  27. if request['code'] != 0:
  28. logger.info('the res of get_new_video_info is %s' % request)
  29. return None
  30. return request['data']
  31. def get_missing_video_info(account_id, cnt, url, start_time, end_time):
  32. request_data = {"accountId": account_id, "videoCnt": cnt, "startTime": start_time, "endTime": end_time}
  33. request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
  34. request = json.loads(request)
  35. if request['code'] != 0:
  36. logger.info('the res of get_missing_video_info is %s' % request)
  37. return None
  38. return request['data']
  39. def get_top_video_info(account_id, cnt, start_time, end_time, url):
  40. request_data = {"accountId": account_id, "num": cnt, "startTime": start_time, "endTime": end_time}
  41. request = requests.post(url, headers=headers, data=json.JSONEncoder().encode(request_data)).text
  42. request = json.loads(request)
  43. if request['code'] != 0:
  44. logger.info('the res of get_top_video_info is %s' % request)
  45. return None
  46. return request['data']
  47. def get_campaign_and_group_name_rule(account_id):
  48. sql = """select campaign_name, group_name
  49. from ctop_ai_kuaishou_advertiser_strategy
  50. where account_id = %s
  51. and status = 1
  52. limit 1
  53. """ % account_id
  54. df = pd.read_sql(sql, engine)
  55. if df.empty:
  56. return None, None
  57. campaign_name = df['campaign_name'].values[0]
  58. group_name = df['group_name'].values[0]
  59. return campaign_name, group_name