get_high_quality_script.py 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125
  1. import datetime
  2. from datetime import timedelta
  3. import pandas as pd
  4. import yaml
  5. from common_func import get_db_engine, get_logger
  6. from get_script_from_tengxunyun import GetScriptFromTengXunYunServer
  7. import traceback
  8. if __name__ == '__main__':
  9. # 0 创建日志对象
  10. logger = get_logger(log_file_name="/data/pythonProject/video-to-word/logs/get_high_quality_script.log",
  11. log_name='get_high_quality_script_logger')
  12. logger.info("get_high_quality_script_logger started! id of logger is: %s" % id(logger))
  13. # 1 读取配置文件,获取数据库链接信息
  14. with open('/data/pythonProject/video-to-word/config/config.yaml', mode='r', encoding='utf-8') as f:
  15. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  16. online_read_engine = get_db_engine(config['productDB'])
  17. test_read_engine = get_db_engine(config['devDB'])
  18. # 2 获取素材信息
  19. two_week_date = (datetime.date.today() + timedelta(days=-14)).strftime('%Y-%m-%d')
  20. one_week_date = (datetime.date.today() + timedelta(days=-7)).strftime('%Y-%m-%d')
  21. today = datetime.date.today().strftime('%Y-%m-%d')
  22. project_info_lst = config['projectInfoForExportScript']
  23. # 分项目进行遍历
  24. for project_info in project_info_lst:
  25. try:
  26. project_id = project_info['project_id']
  27. project_name = project_info['project_name']
  28. channel = project_info['channel']
  29. logger.info("项目:%s 开始执行!" % project_name)
  30. writer = pd.ExcelWriter('/data/pythonProject/video-to-word/data/%s项目近一周高质量&低质量脚本_%s.xlsx' % (project_name, today))
  31. if 1 in channel:
  32. logger.info("项目:%s, 汇创思拓内部素材,开始执行!" % project_name)
  33. # 2-1 获取执行项目两周内的素材日报数据 (单天获取后对数据进行拼接,防止一次性取不出数据)
  34. full_date_df = pd.DataFrame()
  35. for date in pd.date_range(start=two_week_date, freq='D', periods=14):
  36. stat_date = date.strftime('%Y-%m-%d')
  37. sql = """select signature, sum(activation) activation , sum(charge) charge, stat_date
  38. from ctop_kuaishou_report_daily_material
  39. where account_id in (select account_id from ctop_user_allocation where project_id in %s)
  40. and stat_date = '%s'
  41. group by signature,stat_date""" % (project_id, stat_date)
  42. one_date_df = pd.read_sql(sql, online_read_engine)
  43. full_date_df = full_date_df.append(one_date_df)
  44. # 2-2 获取高质量素材(近一周累计激活个数>=50)
  45. one_week_df = full_date_df[full_date_df.stat_date >= one_week_date]
  46. g = one_week_df.groupby('signature').agg({'activation': sum, 'charge': sum})
  47. g.reset_index(drop=False, inplace=True)
  48. high_material_df = g[g['activation'] >= 50]
  49. logger.info("项目:%s, 高质量素材个数为 %s!" % (project_name, len(high_material_df)))
  50. # 2-3 获取低质量素材(冷启动失败或衰退):上两周已经在投放,且近一周的累计激活个数<=10 & >=1个 -- 保证素材有一周的表现期
  51. g = full_date_df.groupby('signature')
  52. two_week_materials = g.filter(lambda x: x['stat_date'].min() <= one_week_date).signature.unique()
  53. one_week_df = full_date_df[(full_date_df.signature.isin(two_week_materials)) & (full_date_df.stat_date >= one_week_date)]
  54. g = one_week_df.groupby('signature').agg({'activation': sum, 'charge': sum})
  55. g.reset_index(drop=False, inplace=True)
  56. low_material_df = g[(g['activation'] <= 10) & (g['activation'] >= 1)]
  57. # 对低质量素材进行抽样 100个
  58. n = 100 if len(low_material_df) > 100 else len(low_material_df)
  59. low_material_df = low_material_df.sample(n=n, random_state=2077)
  60. logger.info("项目:%s, 低质量素材个数为 %s!" % (project_name, len(low_material_df)))
  61. # 2-4 获取 video url
  62. merge_df = pd.concat([low_material_df, high_material_df], axis=0)
  63. sql = """
  64. select url video_url, signature from ctop_kuaishou_video_get
  65. where account_id in (select account_id from ctop_user_allocation where project_id in %s)
  66. and signature in %s
  67. group by signature
  68. """ % (project_id, tuple(merge_df.signature.values),)
  69. url_df = pd.read_sql(sql, online_read_engine)
  70. merge_df = merge_df.merge(url_df, on='signature', how='inner')
  71. # 2-5 获取脚本数据
  72. task_df = merge_df[['signature', 'video_url']]
  73. get_script_ins = GetScriptFromTengXunYunServer(logger, test_read_engine, task_df, task_ids=None)
  74. get_script_ins.submit_task()
  75. if get_script_ins.task_ids:
  76. get_script_ins.get_result()
  77. sql = """
  78. select md5 signature,task_result from tb_asr_result where word_text is not null and md5 in %s""" \
  79. % (tuple(merge_df.signature.values),)
  80. script_df = pd.read_sql(sql, test_read_engine)
  81. script_df.drop_duplicates('signature', keep='first', inplace=True)
  82. script_df['script'] = script_df['task_result'].apply(lambda x: eval(x)['Data']['Result'].split(']')[1].strip())
  83. charge_script_df = script_df.merge(merge_df, on='signature', how='inner')
  84. charge_script_df.rename(columns={'charge': '消耗', 'script': '脚本'}, inplace=True)
  85. charge_script_df[charge_script_df.activation >= 50][['消耗', '脚本', 'video_url']]. \
  86. to_excel(writer, sheet_name='汇创思拓_%s项目近一周高质量素材脚本' % project_name, index=False, header=True)
  87. charge_script_df[charge_script_df.activation <= 10][['消耗', '脚本', 'video_url']]. \
  88. to_excel(writer, sheet_name='汇创思拓_%s项目近一周低质量素材脚本' % project_name, index=False, header=True)
  89. if 0 in channel:
  90. logger.info("项目:%s, 巨量引擎外部素材,开始执行!" % project_name)
  91. sql = """select signature, video_url from ctop_ai_video_info_from_ocean_engine
  92. where project_name = '%s' and stat_date >= '%s'""" % (project_name, one_week_date)
  93. df = pd.read_sql(sql, test_read_engine)
  94. logger.info("项目:%s, 巨量引擎优质素材个数为 %s!" % (project_name, len(df)))
  95. sql = """select md5 signature,task_result from tb_asr_result where word_text is not null and md5 in %s""" \
  96. % (tuple(df.signature.values),)
  97. script_df = pd.read_sql(sql, test_read_engine)
  98. script_df.drop_duplicates('signature', keep='first', inplace=True)
  99. script_df['script'] = script_df['task_result'].apply(lambda x: eval(x)['Data']['Result'].split(']')[1].strip())
  100. script_df.rename(columns={'script': '脚本'}, inplace=True)
  101. merge_df = script_df.merge(df, on='signature', how='inner')
  102. merge_df[['脚本', 'video_url']].to_excel(writer, sheet_name='巨量引擎_%s项目近一周高质量素材脚本' % project_name,
  103. index=False,
  104. header=True)
  105. writer.save()
  106. logger.info("项目: %s,文件导出完成!" % project_name)
  107. except:
  108. logger.error("项目:%s, 发生异常信息为%s!" % (project_info['project_name'], traceback.format_exc()))