get_video_from_ocean_engine.py 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. import json
  2. import os
  3. from urllib.parse import urlencode
  4. import pandas as pd
  5. import requests
  6. import yaml
  7. from common_func import get_db_engine, mysql_replace_into, NpEncoder, get_logger
  8. import datetime
  9. from config.url import get_material_info_from_ocean_engine_url, get_video_info_from_ocean_engine_url
  10. import traceback
  11. from get_script_from_tengxunyun import GetScriptFromTengXunYunServer
  12. def get_material_info(project_name, period_type):
  13. material_df = pd.DataFrame() # 初始化返回的结果
  14. has_more = True # 是否还存在分页数据, 初始化为 True
  15. limit = 10 # 每页获取10条
  16. page = 1 # 第几页
  17. while has_more:
  18. request_data = {'list_type': 1,
  19. 'material_type': 3,
  20. 'order_by': 'click_show_rate',
  21. 'period_type': period_type,
  22. 'aggr_app_code': 4,
  23. 'aggr_category_list': '[]',
  24. 'video_type': '[]',
  25. 'keywords': project_name,
  26. 'landing_type': '[]',
  27. 'limit': limit,
  28. 'page': page,
  29. 'video_duration_type': 5}
  30. try:
  31. request_path = get_material_info_from_ocean_engine_url + '?' + urlencode(request_data)
  32. request = requests.get(request_path)
  33. result = json.loads(request.text)
  34. material_page_df = pd.DataFrame(result['data']['materials'])
  35. material_df = material_df.append(material_page_df)
  36. if result.get('code') == 0 and result.get('data').get('has_more') is True:
  37. page += 1
  38. logger.info("project_name:%s, get_material_info_from_ocean_engine 分页获取第 %s 页" % (project_name, page))
  39. else:
  40. has_more = False
  41. except:
  42. logger.error("project_name:%s, get_material_info_from_ocean_engine 分页获取时发生异常信息: %s" %
  43. (project_name, traceback.format_exc()))
  44. logger.info("project_name:%s, get_material_info_from_ocean_engine 分页获取获取完成,共 %s 个物料信息!" %
  45. (project_name, len(material_df)))
  46. # 数据类型的处理,方便入库
  47. # metrics dict to str
  48. # title list to str
  49. # video_type list to str
  50. # watermarks list to str
  51. material_df[['metrics', 'title', 'video_type', 'watermarks']] = \
  52. material_df[['metrics', 'title', 'video_type', 'watermarks']].astype(str)
  53. material_df.rename(columns={'vid': 'signature'}, inplace=True)
  54. # 添加项目名称和日期
  55. material_df['query_word'] = project_name
  56. material_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
  57. # 写入数据库
  58. material_df.to_sql(name="ctop_ai_material_info_from_ocean_engine",
  59. con=write_engine,
  60. if_exists='append',
  61. index=False,
  62. chunksize=chunk_size,
  63. method=mysql_replace_into)
  64. return material_df
  65. def get_video_info(vid, project_name):
  66. """
  67. 为了提高数据获取的完整性,每次只请求10条数据
  68. :param vid:
  69. :param project_name:
  70. :return:
  71. """
  72. video_df = pd.DataFrame()
  73. # 每次请求的视频个数
  74. cnt_per_request = 10
  75. # 总的视频个数
  76. total_cnt = len(vid)
  77. logger.info("project_name:%s, get_video_info_from_ocean_engine,共 %s 个视频需要请求接口获取video_url" % (project_name, total_cnt))
  78. for i in range(0, total_cnt, cnt_per_request):
  79. if i + cnt_per_request < total_cnt:
  80. query_ids = vid[i: i + cnt_per_request]
  81. logger.info("project_name:%s, get_video_info_from_ocean_engine 分页获取第 %s 个 到 %s 个视频信息" %
  82. (project_name, i, i + cnt_per_request))
  83. else:
  84. query_ids = vid[i:]
  85. logger.info("project_name:%s, get_video_info_from_ocean_engine 分页获取第 %s 个 到 %s 个视频信息" %
  86. (project_name, i, total_cnt))
  87. try:
  88. request_data = {"query_ids": query_ids, "water_mark": "creative_center"}
  89. request = requests.post(url=get_video_info_from_ocean_engine_url,
  90. headers={'Content-Type': 'application/json'},
  91. data=json.dumps(request_data, cls=NpEncoder)
  92. )
  93. response_data = json.loads(request.text)
  94. if response_data.get('code') == 0 and response_data.get('data'):
  95. for key, value in response_data['data'].items():
  96. single_dict = value
  97. single_dict['signature'] = key
  98. single_df = pd.DataFrame([single_dict])
  99. video_df = video_df.append(single_df)
  100. except:
  101. logger.error("project_name:%s, get_video_info_from_ocean_engine 分页获取 %s,出现异常信息:%s" %
  102. (project_name, query_ids, traceback.format_exc()))
  103. # 数据类型的处理,方便入库
  104. # play_info list to str
  105. video_df['play_info'] = video_df['play_info'].astype(str)
  106. video_df.drop(labels='video_id', axis=1, inplace=True)
  107. # 添加项目名称和日期
  108. video_df['query_word'] = project_name
  109. video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
  110. # 写入数据库
  111. video_df.to_sql(name="ctop_ai_video_info_from_ocean_engine",
  112. con=write_engine,
  113. if_exists='append',
  114. index=False,
  115. chunksize=chunk_size,
  116. method=mysql_replace_into)
  117. return video_df
  118. if __name__ == '__main__':
  119. # 创建日志对象
  120. logger = get_logger(log_file_name="/data/pythonProject/video-to-word/logs/get_video_from_ocean_engine.log",
  121. log_name='get_video_from_ocean_engine_logger')
  122. logger.info("get_video_from_ocean_engine started! id of logger is: %s" % id(logger))
  123. # 1 读取配置文件
  124. with open('/data/pythonProject/video-to-word/config/config.yaml', mode='r', encoding='utf-8') as f:
  125. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  126. # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
  127. # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库)
  128. # 注意:该项目的读和写 都使用测试数据库
  129. write_engine = get_db_engine(config['devDB'])
  130. read_engine = get_db_engine(config['devDB'])
  131. # 1-2 分批写入数据库的行数
  132. chunk_size = config['chunkSize']
  133. # 1-3 读取项目列表
  134. project_name_lst = config['projectName']
  135. # 2 分项目获取巨量引擎数据
  136. for project in project_name_lst:
  137. try:
  138. logger.info("****************************** %s 项目开始执行 ******************************" % project)
  139. # 2-1 获取物料列表并写入数据库
  140. material_info_df = get_material_info(project, 7)
  141. # 2-2 根据 signature 获取 url并写入数据库
  142. vid_lst = material_info_df['signature'].values
  143. video_info_df = get_video_info(vid_lst, project)
  144. # 2-3 向腾讯云提交语音转脚本的任务
  145. task_df = video_info_df[['signature', 'video_url']]
  146. get_script_ins = GetScriptFromTengXunYunServer(logger, read_engine, task_df, task_ids=None)
  147. get_script_ins.submit_task()
  148. if get_script_ins.task_ids:
  149. get_script_ins.get_result()
  150. logger.info("****************************** %s 项目执行完成 ******************************" % project)
  151. except:
  152. logger.error("project_name: %s, 发生异常信息: %s" % (project, traceback.format_exc()))