import pandas as pd from urllib.parse import urlencode import requests import json import time from config.url import voice_to_script_task_submit_url, voice_to_script_task_result_url class GetScriptFromTengXunYunServer(object): def __init__(self, logger, db_engine, task_df, task_ids=None): self.logger = logger self.db_engine = db_engine self.task_ids = task_ids self.task_df = task_df def submit_task(self): """ 向腾讯云提交语音转脚本的任务 task_df: DataFrame columns 包含 signature 和 video_url """ # 1 获取已经被提交过的任务 sql = """select md5 signature from tb_asr_result """ submitted_task_df = pd.read_sql(sql, self.db_engine) # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长 to_submit_task_df = self.task_df[~self.task_df.signature.isin(submitted_task_df.signature.values)] self.logger.info("向腾讯云提交语音转脚本的任务个数为 %s" % len(to_submit_task_df)) # 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) self.logger.info("素材:%s, 向腾讯云提交语音转脚本任务的返回信息为:%s " % (material_md5, result)) except: self.logger.error("素材:%s, 向腾讯云提交语音转脚本任务的返回信息为:%s " % (material_md5, request.text)) # 4 获取素材对应的发送请求的 task_id sql = """select task_id from tb_asr_result where md5 in %s""" % (tuple(self.task_df.signature.values),) task_id_df = pd.read_sql(sql, self.db_engine) self.task_ids = list(task_id_df['task_id'].values) def get_result(self): """ 从腾讯云获取脚本 每隔1分钟获取一次,直到没有 执行中或者等待执行 的任务为止 Status Integer 任务状态码,0:任务等待,1:任务执行中,2:任务成功,3:任务失败。 StatusStr String 任务状态,waiting:任务等待,doing:任务执行中,success:任务成功,failed:任务失败。 ErrorMsg String 失败原因说明。 """ self.logger.info("从腾讯云获取脚本的个数为%s" % len(self.task_ids)) while True: sql = """select task_id, task_status from tb_asr_result where task_id in %s and task_status in (0,1)""" % \ (tuple(self.task_ids),) task_status_df = pd.read_sql(sql, self.db_engine) if task_status_df.empty: break else: self.logger.info("从腾讯云获取脚本, 休眠一分钟,等待腾讯云任务计算完成。") time.sleep(60 * 1) 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) self.logger.info("task_id:%s, 从腾讯云获取脚本返回信息为:%s " % (task_id, result)) except: self.logger.error("task_id:%s, 从腾讯云获取脚本返回信息为:%s " % (task_id, request.text))