123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475 |
- 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))
|