fix: auto-fix code issues (cron)
- 修复重复导入/字段 - 修复异常处理 - 修复PEP8格式问题 - 添加类型注解
This commit is contained in:
@@ -6,7 +6,7 @@
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any
|
||||
from typing import Any
|
||||
|
||||
|
||||
class TingwuClient:
|
||||
@@ -18,7 +18,7 @@ class TingwuClient:
|
||||
if not self.access_key or not self.secret_key:
|
||||
raise ValueError("ALI_ACCESS_KEY and ALI_SECRET_KEY required")
|
||||
|
||||
def _sign_request(self, method: str, uri: str, query: str = "", body: str = "") -> Dict[str, str]:
|
||||
def _sign_request(self, method: str, uri: str, query: str = "", body: str = "") -> dict[str, str]:
|
||||
"""阿里云签名 V3"""
|
||||
timestamp = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
|
||||
|
||||
@@ -35,9 +35,9 @@ class TingwuClient:
|
||||
def create_task(self, audio_url: str, language: str = "zh") -> str:
|
||||
"""创建听悟任务"""
|
||||
try:
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
from alibabacloud_tingwu20230930 import models as tingwu_models
|
||||
from alibabacloud_tingwu20230930.client import Client as TingwuSDKClient
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
|
||||
config = open_api_models.Config(
|
||||
access_key_id=self.access_key,
|
||||
@@ -74,12 +74,12 @@ class TingwuClient:
|
||||
print(f"Tingwu API error: {e}")
|
||||
return f"mock_task_{int(time.time())}"
|
||||
|
||||
def get_task_result(self, task_id: str, max_retries: int = 60, interval: int = 5) -> Dict[str, Any]:
|
||||
def get_task_result(self, task_id: str, max_retries: int = 60, interval: int = 5) -> dict[str, Any]:
|
||||
"""获取任务结果"""
|
||||
try:
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
from alibabacloud_tingwu20230930 import models as tingwu_models
|
||||
from alibabacloud_tingwu20230930.client import Client as TingwuSDKClient
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
|
||||
config = open_api_models.Config(
|
||||
access_key_id=self.access_key,
|
||||
@@ -114,7 +114,7 @@ class TingwuClient:
|
||||
|
||||
raise TimeoutError(f"Task {task_id} timeout")
|
||||
|
||||
def _parse_result(self, data) -> Dict[str, Any]:
|
||||
def _parse_result(self, data) -> dict[str, Any]:
|
||||
"""解析结果"""
|
||||
result = data.result
|
||||
transcription = result.transcription
|
||||
@@ -140,7 +140,7 @@ class TingwuClient:
|
||||
"segments": segments
|
||||
}
|
||||
|
||||
def _mock_result(self) -> Dict[str, Any]:
|
||||
def _mock_result(self) -> dict[str, Any]:
|
||||
"""Mock 结果"""
|
||||
return {
|
||||
"full_text": "这是一个示例转录文本,包含 Project Alpha 和 K8s 等术语。",
|
||||
@@ -149,7 +149,7 @@ class TingwuClient:
|
||||
]
|
||||
}
|
||||
|
||||
def transcribe(self, audio_url: str, language: str = "zh") -> Dict[str, Any]:
|
||||
def transcribe(self, audio_url: str, language: str = "zh") -> dict[str, Any]:
|
||||
"""一键转录"""
|
||||
task_id = self.create_task(audio_url, language)
|
||||
print(f"Tingwu task: {task_id}")
|
||||
|
||||
Reference in New Issue
Block a user