134 lines
4.2 KiB
Python
134 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
创建真正的测试音频文件
|
||
"""
|
||
|
||
import os
|
||
import wave
|
||
import struct
|
||
import math
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
def create_test_audio():
|
||
"""创建一个简单的测试音频文件"""
|
||
try:
|
||
import oss2
|
||
|
||
# 创建一个简单的正弦波音频（1秒，440Hz）
|
||
sample_rate = 16000
|
||
duration = 1.0 # 1秒
|
||
frequency = 440 # A4音符
|
||
|
||
# 生成音频数据
|
||
samples = []
|
||
for i in range(int(sample_rate * duration)):
|
||
t = i / sample_rate
|
||
sample = int(32767 * 0.3 * math.sin(2 * math.pi * frequency * t))
|
||
samples.append(sample)
|
||
|
||
# 创建 WAV 文件
|
||
wav_filename = "test_audio.wav"
|
||
with wave.open(wav_filename, 'w') as wav_file:
|
||
wav_file.setnchannels(1) # 单声道
|
||
wav_file.setsampwidth(2) # 16-bit
|
||
wav_file.setframerate(sample_rate) # 16kHz
|
||
|
||
# 写入音频数据
|
||
for sample in samples:
|
||
wav_file.writeframes(struct.pack('<h', sample))
|
||
|
||
print(f"✅ 创建测试音频文件: {wav_filename}")
|
||
print(f"📊 文件大小: {os.path.getsize(wav_filename)} 字节")
|
||
|
||
# 上传到 OSS
|
||
access_key_id = os.getenv('ALIYUN_OSS_ACCESS_KEY_ID')
|
||
access_key_secret = os.getenv('ALIYUN_OSS_ACCESS_KEY_SECRET')
|
||
bucket_name = os.getenv('ALIYUN_OSS_BUCKET_NAME')
|
||
endpoint = os.getenv('ALIYUN_OSS_ENDPOINT')
|
||
|
||
auth = oss2.Auth(access_key_id, access_key_secret)
|
||
bucket = oss2.Bucket(auth, endpoint, bucket_name)
|
||
|
||
# 上传文件
|
||
test_key = "voice_call/real_test_audio.wav"
|
||
with open(wav_filename, 'rb') as f:
|
||
result = bucket.put_object(test_key, f)
|
||
|
||
if result.status == 200:
|
||
file_url = f"https://{bucket_name}.{endpoint.replace('https://', '')}/{test_key}"
|
||
print(f"✅ 上传成功: {file_url}")
|
||
|
||
# 清理本地文件
|
||
os.remove(wav_filename)
|
||
|
||
return file_url
|
||
else:
|
||
print(f"❌ 上传失败: {result.status}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 创建测试音频失败: {e}")
|
||
return None
|
||
|
||
def test_real_audio_asr(audio_url):
|
||
"""使用真实音频测试 ASR"""
|
||
try:
|
||
from dashscope.audio.asr import Transcription
|
||
import dashscope
|
||
|
||
api_key = os.getenv('DASHSCOPE_API_KEY')
|
||
dashscope.api_key = api_key
|
||
|
||
print(f"\n🎵 测试真实音频 ASR: {audio_url}")
|
||
|
||
task_response = Transcription.async_call(
|
||
model='paraformer-v2',
|
||
file_urls=[audio_url]
|
||
)
|
||
|
||
print(f"📊 任务状态: {task_response.status_code}")
|
||
|
||
if task_response.status_code == 200:
|
||
task_id = task_response.output.task_id
|
||
print(f"✅ 任务创建成功: {task_id}")
|
||
|
||
result = Transcription.wait(task=task_id)
|
||
print(f"📊 识别状态: {result.status_code}")
|
||
print(f"📋 完整结果: {result}")
|
||
|
||
if result.status_code == 200:
|
||
if result.output.task_status == "SUCCEEDED":
|
||
print(f"🎉 ASR 识别成功！")
|
||
return True
|
||
else:
|
||
print(f"❌ ASR 任务失败: {result.output.message}")
|
||
else:
|
||
print(f"❌ ASR 识别失败")
|
||
else:
|
||
print(f"❌ 任务创建失败")
|
||
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ ASR 测试失败: {e}")
|
||
return False
|
||
|
||
def main():
|
||
print("🚀 创建真实音频文件并测试 ASR...")
|
||
|
||
audio_url = create_test_audio()
|
||
if audio_url:
|
||
success = test_real_audio_asr(audio_url)
|
||
|
||
if success:
|
||
print("🎉 真实音频 ASR 测试成功！")
|
||
else:
|
||
print("💥 真实音频 ASR 测试失败！")
|
||
else:
|
||
print("💥 无法创建测试音频文件！")
|
||
|
||
if __name__ == "__main__":
|
||
main() |