get_video_from_ocean_engine.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219
  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
  8. import datetime
  9. from config.url import voice_to_script_task_submit_url, voice_to_script_task_result_url, get_material_info_from_ocean_engine_url, \
  10. get_video_info_from_ocean_engine_url
  11. import time
  12. def get_material_info(project_name, period_type):
  13. material_df = pd.DataFrame()
  14. has_more = True # 是否还存在分页数据, 初始化为 True
  15. limit = 10 # 每页获取30条
  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. request_path = get_material_info_from_ocean_engine_url + '?' + urlencode(request_data)
  31. request = requests.get(request_path)
  32. result = json.loads(request.text)
  33. material_page_df = pd.DataFrame(result['data']['materials'])
  34. material_df = material_df.append(material_page_df)
  35. if result.get('code') == 0 and result.get('data').get('has_more') is True:
  36. page += 1
  37. else:
  38. has_more = False
  39. # 数据类型的处理,方便入库
  40. # metrics dict to str
  41. # title list to str
  42. # video_type list to str
  43. # watermarks list to str
  44. material_df[['metrics', 'title', 'video_type', 'watermarks']] = \
  45. material_df[['metrics', 'title', 'video_type', 'watermarks']].astype(str)
  46. material_df.rename(columns={'vid': 'signature'}, inplace=True)
  47. # 添加项目名称和日期
  48. material_df['project_name'] = project_name
  49. material_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
  50. # 写入数据库
  51. material_df.to_sql(name="ctop_ai_material_info_from_ocean_engine",
  52. con=write_engine,
  53. if_exists='append',
  54. index=False,
  55. chunksize=chunk_size,
  56. method=mysql_replace_into)
  57. return material_df
  58. def get_video_info(vid, project_name):
  59. """
  60. 为了提高数据获取的完整性,每次只请求10条数据
  61. :param vid:
  62. :param project_name:
  63. :return:
  64. """
  65. video_df = pd.DataFrame()
  66. # 每次请求的视频个数
  67. cnt_per_request = 10
  68. # 总的视频个数
  69. total_cnt = len(vid)
  70. for i in range(0, total_cnt, cnt_per_request):
  71. if i + cnt_per_request < total_cnt:
  72. query_ids = vid[i: i + cnt_per_request]
  73. else:
  74. query_ids = vid[i:]
  75. request_data = {"query_ids": query_ids, "water_mark": "creative_center"}
  76. request = requests.post(url=get_video_info_from_ocean_engine_url,
  77. headers={'Content-Type': 'application/json'},
  78. data=json.dumps(request_data, cls=NpEncoder)
  79. )
  80. response_data = json.loads(request.text)
  81. if response_data.get('code') == 0 and response_data.get('data'):
  82. for key, value in response_data['data'].items():
  83. single_dict = value
  84. single_dict['signature'] = key
  85. single_df = pd.DataFrame([single_dict])
  86. video_df = video_df.append(single_df)
  87. # 数据类型的处理,方便入库
  88. # play_info list to str
  89. video_df['play_info'] = video_df['play_info'].astype(str)
  90. video_df.drop(labels='video_id', axis=1, inplace=True)
  91. # 添加项目名称和日期
  92. video_df['project_name'] = project_name
  93. video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
  94. # 写入数据库
  95. video_df.to_sql(name="ctop_ai_video_info_from_ocean_engine",
  96. con=write_engine,
  97. if_exists='append',
  98. index=False,
  99. chunksize=chunk_size,
  100. method=mysql_replace_into)
  101. return video_df
  102. def submit_script_task(df):
  103. """
  104. 向腾讯云提交语音转脚本的任务
  105. :param df: DataFrame columns 包含 signature 和 url
  106. :return: task_ids
  107. """
  108. # 1 获取已经被提交过的任务
  109. sql = """select md5 signature from tb_asr_result """
  110. submitted_task_df = pd.read_sql(sql, read_engine)
  111. # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长
  112. to_submit_task_df = df[~df.signature.isin(submitted_task_df.signature.values)]
  113. # 3 发送请求,提交任务
  114. for index, row in to_submit_task_df.iterrows():
  115. material_md5 = row['signature']
  116. material_url = row['video_url']
  117. request_data = {"md5": material_md5, "url": material_url}
  118. request_full_path = voice_to_script_task_submit_url + '?' + urlencode(request_data)
  119. request = requests.post(request_full_path)
  120. try:
  121. result = json.loads(request.text)
  122. print(result)
  123. except:
  124. print("error", request.text)
  125. # 4 获取素材对应的发送请求的 task_id
  126. sql = """
  127. select task_id from tb_asr_result where md5 in %s
  128. """ % (tuple(df.signature.values),)
  129. task_id_df = pd.read_sql(sql, read_engine)
  130. task_ids = task_id_df['task_id'].values
  131. return task_ids
  132. def get_result_from_tx(task_id_lst):
  133. """
  134. 从腾讯云获取脚本
  135. 每隔5分钟获取一次,直到没有 执行中或者等待执行 的任务为止
  136. Status Integer 任务状态码,0:任务等待,1:任务执行中,2:任务成功,3:任务失败。
  137. StatusStr String 任务状态,waiting:任务等待,doing:任务执行中,success:任务成功,failed:任务失败。
  138. ErrorMsg String 失败原因说明。
  139. """
  140. while True:
  141. print("sleep 5 mins")
  142. time.sleep(60 * 5)
  143. sql = """select task_id, task_status from tb_asr_result where task_id in %s and task_status in (0,1)""" % (tuple(task_id_lst),)
  144. task_status_df = pd.read_sql(sql, read_engine)
  145. if task_status_df.empty:
  146. break
  147. for task_id in task_status_df.task_id.values:
  148. request_data = {'task_id': task_id}
  149. request_full_path = voice_to_script_task_result_url + '?' + urlencode(request_data)
  150. request = requests.post(request_full_path)
  151. try:
  152. result = json.loads(request.text)
  153. print(task_id, result['status'])
  154. except:
  155. print("error", task_id, request.text)
  156. if __name__ == '__main__':
  157. # 1 读取配置文件
  158. with open('config/config.yaml', mode='r', encoding='utf-8') as f:
  159. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  160. # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
  161. # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库)
  162. # 注意: 该项目的读和写 都使用测试数据库
  163. if os.getenv('LYY_DEV', 'unknown') == 'dev':
  164. write_engine = get_db_engine(config['devDB'])
  165. else:
  166. write_engine = get_db_engine(config['devDB'])
  167. read_engine = get_db_engine(config['devDB'])
  168. # 1-2 每次写入数据库的行数
  169. chunk_size = config['chunkSize']
  170. # 1-3 读取项目列表
  171. project_name_lst = config['projectName']
  172. # 2 分项目获取巨量引擎数据
  173. for project in project_name_lst:
  174. # 2-1 获取物料列表并写入数据库
  175. material_info_df = get_material_info(project, 7)
  176. # 2-2 根据 signature 获取 url 并写入数据库
  177. vid_lst = material_info_df['signature'].values
  178. video_info_df = get_video_info(vid_lst, project)
  179. # 2-3 向腾讯云提交语音转脚本的任务
  180. task_df = video_info_df[['signature', 'video_url']]
  181. task_ids = submit_script_task(task_df)
  182. # 2-4 向腾讯云获取已提交任务的脚本
  183. get_result_from_tx(task_ids)