Ai_GirlFriend/create_test_audio.py

134 lines
4.2 KiB
Python
Raw Normal View History

2026-03-03 19:06:01 +08:00
#!/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()