|
|
# -*- coding: utf-8 -*-
|
|
|
import json
|
|
|
import time
|
|
|
from typing import Dict, Tuple
|
|
|
|
|
|
from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException
|
|
|
from aliyunsdkcore.client import AcsClient
|
|
|
from aliyunsdkcore.request import CommonRequest
|
|
|
from Config import *
|
|
|
# 服务常量配置
|
|
|
CONFIG = {
|
|
|
"REGION_ID": "cn-shanghai",
|
|
|
"PRODUCT": "nls-filetrans",
|
|
|
"DOMAIN": "filetrans.cn-shanghai.aliyuncs.com",
|
|
|
"API_VERSION": "2018-08-17",
|
|
|
"MAX_RETRIES": 20,
|
|
|
"POLL_INTERVAL": 10
|
|
|
}
|
|
|
|
|
|
|
|
|
class TranscriptionError(Exception):
|
|
|
"""自定义语音识别异常基类"""
|
|
|
pass
|
|
|
|
|
|
|
|
|
class APIConnectionError(TranscriptionError):
|
|
|
"""API连接异常"""
|
|
|
pass
|
|
|
|
|
|
|
|
|
class TaskSubmissionError(TranscriptionError):
|
|
|
"""任务提交异常"""
|
|
|
pass
|
|
|
|
|
|
|
|
|
class TaskTimeoutError(TranscriptionError):
|
|
|
"""任务超时异常"""
|
|
|
pass
|
|
|
|
|
|
|
|
|
def submit_transcription_task(client: AcsClient, app_key: str, file_link: str) -> str:
|
|
|
"""提交语音识别任务
|
|
|
|
|
|
:param client: 阿里云客户端实例
|
|
|
:param app_key: 应用密钥
|
|
|
:param file_link: 音频文件URL
|
|
|
:return: 任务ID
|
|
|
:raises TaskSubmissionError: 任务提交失败时抛出
|
|
|
"""
|
|
|
task_config = {
|
|
|
"appkey": app_key,
|
|
|
"file_link": file_link,
|
|
|
"version": "4.0",
|
|
|
"enable_words": False
|
|
|
}
|
|
|
|
|
|
request = CommonRequest()
|
|
|
request.set_domain(CONFIG["DOMAIN"])
|
|
|
request.set_version(CONFIG["API_VERSION"])
|
|
|
request.set_product(CONFIG["PRODUCT"])
|
|
|
request.set_action_name("SubmitTask")
|
|
|
request.set_method('POST')
|
|
|
request.add_body_params("Task", json.dumps(task_config))
|
|
|
|
|
|
try:
|
|
|
response = client.do_action_with_exception(request)
|
|
|
response_data = json.loads(response)
|
|
|
|
|
|
if response_data.get("StatusText") != "SUCCESS":
|
|
|
raise TaskSubmissionError(f"任务提交失败: {response_data.get('StatusText', '未知错误')}")
|
|
|
|
|
|
return response_data["TaskId"]
|
|
|
|
|
|
except (ServerException, ClientException) as e:
|
|
|
raise APIConnectionError(f"API连接异常: {str(e)}") from e
|
|
|
except KeyError as e:
|
|
|
raise TaskSubmissionError("响应缺少TaskId字段") from e
|
|
|
|
|
|
|
|
|
def poll_transcription_result(client: AcsClient, task_id: str) -> Dict:
|
|
|
"""轮询识别任务结果
|
|
|
|
|
|
:param client: 阿里云客户端实例
|
|
|
:param task_id: 任务ID
|
|
|
:return: 识别结果字典
|
|
|
:raises TaskTimeoutError: 超时未完成时抛出
|
|
|
"""
|
|
|
request = CommonRequest()
|
|
|
request.set_domain(CONFIG["DOMAIN"])
|
|
|
request.set_version(CONFIG["API_VERSION"])
|
|
|
request.set_product(CONFIG["PRODUCT"])
|
|
|
request.set_action_name("GetTaskResult")
|
|
|
request.set_method('GET')
|
|
|
request.add_query_param("TaskId", task_id)
|
|
|
|
|
|
retries = 0
|
|
|
while retries < CONFIG["MAX_RETRIES"]:
|
|
|
try:
|
|
|
response = client.do_action_with_exception(request)
|
|
|
result = json.loads(response)
|
|
|
status = result.get("StatusText", "")
|
|
|
|
|
|
if status == "SUCCESS":
|
|
|
return result.get("Result", {})
|
|
|
if status in ("RUNNING", "QUEUEING"):
|
|
|
time.sleep(CONFIG["POLL_INTERVAL"])
|
|
|
retries += 1
|
|
|
else:
|
|
|
raise TaskSubmissionError(f"识别失败，状态: {status}")
|
|
|
|
|
|
except (ServerException, ClientException) as e:
|
|
|
raise APIConnectionError(f"查询异常: {str(e)}") from e
|
|
|
|
|
|
raise TaskTimeoutError(f"超过最大重试次数({CONFIG['MAX_RETRIES']})，任务未完成")
|
|
|
|
|
|
|
|
|
def save_transcription_result(result: Dict, output_path: Path) -> Path:
|
|
|
"""保存识别结果到文件
|
|
|
|
|
|
:param result: 识别结果字典
|
|
|
:param output_path: 输出文件路径
|
|
|
:return: 实际保存路径
|
|
|
:raises IOError: 文件保存失败时抛出
|
|
|
"""
|
|
|
if not result.get('Sentences'):
|
|
|
raise ValueError("识别结果为空")
|
|
|
|
|
|
try:
|
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
output_path.write_text(str(result['Sentences']), encoding='utf-8')
|
|
|
return output_path.resolve()
|
|
|
except (IOError, PermissionError) as e:
|
|
|
raise IOError(f"文件保存失败: {str(e)}") from e
|
|
|
|
|
|
|
|
|
def transcribe_audio_file(
|
|
|
access_key_id: str,
|
|
|
access_key_secret: str,
|
|
|
app_key: str,
|
|
|
audio_url: str,
|
|
|
output_path: Path = Path("识别结果.txt")
|
|
|
) -> Tuple[bool, Path, str]:
|
|
|
"""语音识别主流程
|
|
|
|
|
|
:param access_key_id: 阿里云访问密钥ID
|
|
|
:param access_key_secret: 阿里云访问密钥
|
|
|
:param app_key: 应用密钥
|
|
|
:param audio_url: 音频文件URL
|
|
|
:param output_path: 输出文件路径
|
|
|
:return: (成功标志, 输出路径, 错误信息)
|
|
|
"""
|
|
|
client = AcsClient(access_key_id, access_key_secret, CONFIG["REGION_ID"])
|
|
|
|
|
|
try:
|
|
|
task_id = submit_transcription_task(client, app_key, audio_url)
|
|
|
print(f"任务已提交，ID: {task_id}")
|
|
|
|
|
|
result = poll_transcription_result(client, task_id)
|
|
|
saved_path = save_transcription_result(result, output_path)
|
|
|
|
|
|
return True, saved_path, ""
|
|
|
|
|
|
except TranscriptionError as e:
|
|
|
return False, Path(), str(e)
|
|
|
except Exception as e:
|
|
|
return False, Path(), f"未处理的异常: {str(e)}"
|
|
|
|
|
|
def ShiBie(audio_url, output_path):
|
|
|
# 配置参数（应通过环境变量或配置文件获取）
|
|
|
CREDENTIALS = {
|
|
|
"access_key_id": ALY_AK,
|
|
|
"access_key_secret": ALY_SK,
|
|
|
"app_key": "OIpiw501l4o6MYEe",
|
|
|
"audio_url":audio_url,
|
|
|
"output_path": output_path
|
|
|
}
|
|
|
|
|
|
# 执行识别
|
|
|
success, path, error = transcribe_audio_file(**CREDENTIALS)
|
|
|
|
|
|
if success:
|
|
|
print(f"✅ 识别成功，文件已保存至: {path}")
|
|
|
else:
|
|
|
print(f"❌ 识别失败: {error}")
|
|
|
exit(1)
|
|
|
|
|
|
|
