import json import os from urllib.parse import urlencode import pandas as pd import requests import yaml from common_func import get_db_engine, mysql_replace_into, NpEncoder import datetime from config.url import voice_to_script_task_submit_url, voice_to_script_task_result_url, get_material_info_from_ocean_engine_url, \ get_video_info_from_ocean_engine_url import time def get_material_info(project_name, period_type): material_df = pd.DataFrame() has_more = True # 是否还存在分页数据, 初始化为 True limit = 10 # 每页获取30条 page = 1 # 第几页 while has_more: request_data = {'list_type': 1, 'material_type': 3, 'order_by': 'click_show_rate', 'period_type': period_type, 'aggr_app_code': 4, 'aggr_category_list': '[]', 'video_type': '[]', 'keywords': project_name, 'landing_type': '[]', 'limit': limit, 'page': page, 'video_duration_type': 5} request_path = get_material_info_from_ocean_engine_url + '?' + urlencode(request_data) request = requests.get(request_path) result = json.loads(request.text) material_page_df = pd.DataFrame(result['data']['materials']) material_df = material_df.append(material_page_df) if result.get('code') == 0 and result.get('data').get('has_more') is True: page += 1 else: has_more = False # 数据类型的处理,方便入库 # metrics dict to str # title list to str # video_type list to str # watermarks list to str material_df[['metrics', 'title', 'video_type', 'watermarks']] = \ material_df[['metrics', 'title', 'video_type', 'watermarks']].astype(str) material_df.rename(columns={'vid': 'signature'}, inplace=True) # 添加项目名称和日期 material_df['project_name'] = project_name material_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d') # 写入数据库 material_df.to_sql(name="ctop_ai_material_info_from_ocean_engine", con=write_engine, if_exists='append', index=False, chunksize=chunk_size, method=mysql_replace_into) return material_df def get_video_info(vid, project_name): """ 为了提高数据获取的完整性,每次只请求10条数据 :param vid: :param project_name: :return: """ video_df = pd.DataFrame() # 每次请求的视频个数 cnt_per_request = 10 # 总的视频个数 total_cnt = len(vid) for i in range(0, total_cnt, cnt_per_request): if i + cnt_per_request < total_cnt: query_ids = vid[i: i + cnt_per_request] else: query_ids = vid[i:] request_data = {"query_ids": query_ids, "water_mark": "creative_center"} request = requests.post(url=get_video_info_from_ocean_engine_url, headers={'Content-Type': 'application/json'}, data=json.dumps(request_data, cls=NpEncoder) ) response_data = json.loads(request.text) if response_data.get('code') == 0 and response_data.get('data'): for key, value in response_data['data'].items(): single_dict = value single_dict['signature'] = key single_df = pd.DataFrame([single_dict]) video_df = video_df.append(single_df) # 数据类型的处理,方便入库 # play_info list to str video_df['play_info'] = video_df['play_info'].astype(str) video_df.drop(labels='video_id', axis=1, inplace=True) # 添加项目名称和日期 video_df['project_name'] = project_name video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d') # 写入数据库 video_df.to_sql(name="ctop_ai_video_info_from_ocean_engine", con=write_engine, if_exists='append', index=False, chunksize=chunk_size, method=mysql_replace_into) return video_df def submit_script_task(df): """ 向腾讯云提交语音转脚本的任务 :param df: DataFrame columns 包含 signature 和 url :return: task_ids """ # 1 获取已经被提交过的任务 sql = """select md5 signature from tb_asr_result """ submitted_task_df = pd.read_sql(sql, read_engine) # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长 to_submit_task_df = df[~df.signature.isin(submitted_task_df.signature.values)] # 3 发送请求,提交任务 for index, row in to_submit_task_df.iterrows(): material_md5 = row['signature'] material_url = row['video_url'] request_data = {"md5": material_md5, "url": material_url} request_full_path = voice_to_script_task_submit_url + '?' + urlencode(request_data) request = requests.post(request_full_path) try: result = json.loads(request.text) print(result) except: print("error", request.text) # 4 获取素材对应的发送请求的 task_id sql = """ select task_id from tb_asr_result where md5 in %s """ % (tuple(df.signature.values),) task_id_df = pd.read_sql(sql, read_engine) task_ids = task_id_df['task_id'].values return task_ids def get_result_from_tx(task_id_lst): """ 从腾讯云获取脚本 每隔5分钟获取一次,直到没有 执行中或者等待执行 的任务为止 Status Integer 任务状态码,0:任务等待,1:任务执行中,2:任务成功,3:任务失败。 StatusStr String 任务状态,waiting:任务等待,doing:任务执行中,success:任务成功,failed:任务失败。 ErrorMsg String 失败原因说明。 """ while True: print("sleep 5 mins") time.sleep(60 * 5) 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),) task_status_df = pd.read_sql(sql, read_engine) if task_status_df.empty: break for task_id in task_status_df.task_id.values: request_data = {'task_id': task_id} request_full_path = voice_to_script_task_result_url + '?' + urlencode(request_data) request = requests.post(request_full_path) try: result = json.loads(request.text) print(task_id, result['status']) except: print("error", task_id, request.text) if __name__ == '__main__': # 1 读取配置文件 with open('config/config.yaml', mode='r', encoding='utf-8') as f: config = yaml.load(f.read(), Loader=yaml.FullLoader) # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换 # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库) # 注意: 该项目的读和写 都使用测试数据库 if os.getenv('LYY_DEV', 'unknown') == 'dev': write_engine = get_db_engine(config['devDB']) else: write_engine = get_db_engine(config['devDB']) read_engine = get_db_engine(config['devDB']) # 1-2 每次写入数据库的行数 chunk_size = config['chunkSize'] # 1-3 读取项目列表 project_name_lst = config['projectName'] # 2 分项目获取巨量引擎数据 for project in project_name_lst: # 2-1 获取物料列表并写入数据库 material_info_df = get_material_info(project, 7) # 2-2 根据 signature 获取 url 并写入数据库 vid_lst = material_info_df['signature'].values video_info_df = get_video_info(vid_lst, project) # 2-3 向腾讯云提交语音转脚本的任务 task_df = video_info_df[['signature', 'video_url']] task_ids = submit_script_task(task_df) # 2-4 向腾讯云获取已提交任务的脚本 get_result_from_tx(task_ids)