fix: auto-fix code issues (cron)

- 修复重复导入/字段
- 修复异常处理
- 修复PEP8格式问题
- 添加类型注解
This commit is contained in:
OpenClaw Bot
2026-02-27 21:12:04 +08:00
parent 17bda3dbce
commit d767f0dddc
27 changed files with 3636 additions and 4158 deletions

View File

@@ -20,7 +20,7 @@ class TingwuClient:
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')
timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
# 简化签名,实际生产需要完整实现
# 这里使用基础认证头
@@ -39,25 +39,16 @@ class TingwuClient:
from alibabacloud_tingwu20230930 import models as tingwu_models
from alibabacloud_tingwu20230930.client import Client as TingwuSDKClient
config = open_api_models.Config(
access_key_id=self.access_key,
access_key_secret=self.secret_key
)
config = open_api_models.Config(access_key_id=self.access_key, access_key_secret=self.secret_key)
config.endpoint = "tingwu.cn-beijing.aliyuncs.com"
client = TingwuSDKClient(config)
request = tingwu_models.CreateTaskRequest(
type="offline",
input=tingwu_models.Input(
source="OSS",
file_url=audio_url
),
input=tingwu_models.Input(source="OSS", file_url=audio_url),
parameters=tingwu_models.Parameters(
transcription=tingwu_models.Transcription(
diarization_enabled=True,
sentence_max_length=20
)
)
transcription=tingwu_models.Transcription(diarization_enabled=True, sentence_max_length=20)
),
)
response = client.create_task(request)
@@ -81,10 +72,7 @@ class TingwuClient:
from alibabacloud_tingwu20230930 import models as tingwu_models
from alibabacloud_tingwu20230930.client import Client as TingwuSDKClient
config = open_api_models.Config(
access_key_id=self.access_key,
access_key_secret=self.secret_key
)
config = open_api_models.Config(access_key_id=self.access_key, access_key_secret=self.secret_key)
config.endpoint = "tingwu.cn-beijing.aliyuncs.com"
client = TingwuSDKClient(config)
@@ -128,25 +116,29 @@ class TingwuClient:
if transcription.sentences:
for sent in transcription.sentences:
segments.append({
"start": sent.begin_time / 1000,
"end": sent.end_time / 1000,
"text": sent.text,
"speaker": f"Speaker {sent.speaker_id}"
})
segments.append(
{
"start": sent.begin_time / 1000,
"end": sent.end_time / 1000,
"text": sent.text,
"speaker": f"Speaker {sent.speaker_id}",
}
)
return {
"full_text": full_text.strip(),
"segments": segments
}
return {"full_text": full_text.strip(), "segments": segments}
def _mock_result(self) -> dict[str, Any]:
"""Mock 结果"""
return {
"full_text": "这是一个示例转录文本,包含 Project Alpha 和 K8s 等术语。",
"segments": [
{"start": 0.0, "end": 5.0, "text": "这是一个示例转录文本,包含 Project Alpha 和 K8s 等术语。", "speaker": "Speaker A"}
]
{
"start": 0.0,
"end": 5.0,
"text": "这是一个示例转录文本,包含 Project Alpha 和 K8s 等术语。",
"speaker": "Speaker A",
}
],
}
def transcribe(self, audio_url: str, language: str = "zh") -> dict[str, Any]: