質量聲明:原創文章,內容質量問題請評論吐槽。如對您產生干擾,可私信刪除。
主要參考:阿里云官方接口文檔
摘要: Python調用阿里云的智能語音交互接口,依靠對象存儲服務(OSS)上傳音頻,實現錄音文件識別,輸出為txt文本。支持單軌/雙軌的wav、mp3格式,最大支持文件512MB,最大錄音時長2個小時。
文章目錄
- 處理音頻
- 安裝 pydub
- 安裝 ffmpeg
- 音頻處理程序
- 上傳至OSS
- 錄音識別程序
- 識別結果
處理音頻
由于阿里、百度、騰訊、訊飛等語音處理平臺都對語音參數有特定要求,所以我們需要預先處理音頻。好在要求都基本一致,主要是 采樣率必須是16k Hz或8k Hz,采樣位數16bit,單/雙通道,wav或mp3 。這些通過調用pydub包即可實現轉換。
安裝 pydub
pip3
install
pydub
安裝 ffmpeg
- 下載地址:https://ffmpeg.zeranoe.com/builds/
- 本文版本:ffmpeg-20190826-0821bc4-win64-static.zip
- 下載好不需要安裝,解壓至某一文件夾,配置好環境變量(如: D:\Program Files\ffmpeg-20190826-0821bc4-win64-static\bin )即可
-
驗證安裝:命令行輸入
ffmpeg -version
音頻處理程序
- 實現時長剪輯、采樣率&采樣位數轉換、聲道選擇、格式轉換功能,支持WAV、MP3、ogg、flv格式。
from
pydub
import
AudioSegment
def
wavSample
(
from_path
,
to_path
,
frame_rate
=
16000
,
channels
=
1
,
startMin
=
0
,
endMin
=
None
)
:
# 根據文件的類型選擇導入方法
audio
=
AudioSegment
.
from_wav
(
from_path
)
# mp3_version = AudioSegment.from_mp3("never_gonna_give_you_up.mp3")
# ogg_version = AudioSegment.from_ogg("never_gonna_give_you_up.ogg")
# flv_version = AudioSegment.from_flv("never_gonna_give_you_up.flv")
startTime
=
startMin
*
60
*
1000
# 單位ms
endTime
=
endMin
*
60
*
1000
+
1
if
endMin
else
None
# 單位ms
audio
=
audio
[
startTime
:
endTime
]
mono
=
audio
.
set_frame_rate
(
frame_rate
)
.
set_channels
(
channels
)
# 設置聲道和采樣率
mono
.
export
(
to_path
,
format
=
'wav'
,
codec
=
'pcm_s16le'
)
# codec此參數本意是設定16bits pcm編碼器
if
__name__
==
'__main__'
:
wavSample
(
"sample.WAV"
,
"sample_new.WAV"
)
上傳至OSS
由于各大智能語音識別服務的平臺,都僅支持基于HTTP可訪問的URL地址,不支持提交本地文件,所以需要上傳至網絡。我用的是阿里云的OSS存儲,使用方法參考官方文檔:阿里云對象存儲 OSS 快速入門。主要瀏覽如何開通OSS,如何上傳錄音文件并開放讀權限即可。上傳完成,需要獲得訪問鏈接,格式如:
https://xxxx.oss-cn-beijing.aliyuncs.com/xxxx.WAV
錄音識別程序
開通服務: 參考官方文檔,瀏覽如何開通智能語音交互即可,需要獲得
- accessKey Id
- accessKey Secret
- appKey
調用代碼: 改動自官方Demo,新增識別結果解析,組合后存儲為txt文件
# -*- coding: utf8 -*-
import
json
import
time
from
aliyunsdkcore
.
acs_exception
.
exceptions
import
ClientException
from
aliyunsdkcore
.
acs_exception
.
exceptions
import
ServerException
from
aliyunsdkcore
.
client
import
AcsClient
from
aliyunsdkcore
.
request
import
CommonRequest
def
fileTrans
(
akId
,
akSecret
,
appKey
,
fileLink
)
:
# 地域ID,常量內容,請勿改變
REGION_ID
=
"cn-shanghai"
PRODUCT
=
"nls-filetrans"
DOMAIN
=
"filetrans.cn-shanghai.aliyuncs.com"
API_VERSION
=
"2018-08-17"
POST_REQUEST_ACTION
=
"SubmitTask"
GET_REQUEST_ACTION
=
"GetTaskResult"
# 請求參數key
KEY_APP_KEY
=
"appkey"
KEY_FILE_LINK
=
"file_link"
KEY_VERSION
=
"version"
KEY_ENABLE_WORDS
=
"enable_words"
# 是否開啟智能分軌
KEY_AUTO_SPLIT
=
"auto_split"
# 響應參數key
KEY_TASK
=
"Task"
KEY_TASK_ID
=
"TaskId"
KEY_STATUS_TEXT
=
"StatusText"
KEY_RESULT
=
"Result"
# 狀態值
STATUS_SUCCESS
=
"SUCCESS"
STATUS_RUNNING
=
"RUNNING"
STATUS_QUEUEING
=
"QUEUEING"
# 創建AcsClient實例
client
=
AcsClient
(
akId
,
akSecret
,
REGION_ID
)
# 提交錄音文件識別請求
postRequest
=
CommonRequest
(
)
postRequest
.
set_domain
(
DOMAIN
)
postRequest
.
set_version
(
API_VERSION
)
postRequest
.
set_product
(
PRODUCT
)
postRequest
.
set_action_name
(
POST_REQUEST_ACTION
)
postRequest
.
set_method
(
'POST'
)
# 新接入請使用4.0版本,已接入(默認2.0)如需維持現狀,請注釋掉該參數設置
# 設置是否輸出詞信息,默認為false,開啟時需要設置version為4.0
task
=
{
KEY_APP_KEY
:
appKey
,
KEY_FILE_LINK
:
fileLink
,
KEY_VERSION
:
"4.0"
,
KEY_ENABLE_WORDS
:
False
}
# 開啟智能分軌,如果開啟智能分軌 task中設置KEY_AUTO_SPLIT : True
# task = {KEY_APP_KEY : appKey, KEY_FILE_LINK : fileLink, KEY_VERSION : "4.0", KEY_ENABLE_WORDS : False, KEY_AUTO_SPLIT : True}
task
=
json
.
dumps
(
task
)
postRequest
.
add_body_params
(
KEY_TASK
,
task
)
taskId
=
""
try
:
postResponse
=
client
.
do_action_with_exception
(
postRequest
)
postResponse
=
json
.
loads
(
postResponse
)
statusText
=
postResponse
[
KEY_STATUS_TEXT
]
if
statusText
==
STATUS_SUCCESS
:
print
(
"錄音文件識別請求成功響應!"
)
taskId
=
postResponse
[
KEY_TASK_ID
]
else
:
print
(
"錄音文件識別請求失敗!"
)
return
except
ServerException
as
e
:
print
(
e
)
except
ClientException
as
e
:
print
(
e
)
# 創建CommonRequest,設置任務ID
getRequest
=
CommonRequest
(
)
getRequest
.
set_domain
(
DOMAIN
)
getRequest
.
set_version
(
API_VERSION
)
getRequest
.
set_product
(
PRODUCT
)
getRequest
.
set_action_name
(
GET_REQUEST_ACTION
)
getRequest
.
set_method
(
'GET'
)
getRequest
.
add_query_param
(
KEY_TASK_ID
,
taskId
)
# 提交錄音文件識別結果查詢請求
# 以輪詢的方式進行識別結果的查詢,直到服務端返回的狀態描述符為"SUCCESS"、"SUCCESS_WITH_NO_VALID_FRAGMENT",
# 或者為錯誤描述,則結束輪詢。
statusText
=
""
while
True
:
try
:
getResponse
=
client
.
do_action_with_exception
(
getRequest
)
getResponse
=
json
.
loads
(
getResponse
)
statusText
=
getResponse
[
KEY_STATUS_TEXT
]
if
statusText
==
STATUS_RUNNING
or
statusText
==
STATUS_QUEUEING
:
# 繼續輪詢
time
.
sleep
(
30
)
else
:
# 退出輪詢
break
except
ServerException
as
e
:
print
(
e
)
except
ClientException
as
e
:
print
(
e
)
# 結果解析與保存
if
statusText
==
STATUS_SUCCESS
:
texts
=
""
result
=
getResponse
[
"Result"
]
sentences
=
result
[
"Sentences"
]
maxlength
=
30
# 按長度分段
for
i
,
sentence
in
enumerate
(
sentences
)
:
index
=
i
%
(
maxlength
+
1
)
if
index
==
maxlength
:
# 以追加方式存入文件
with
open
(
"recognition.txt"
,
"a+"
)
as
f
:
f
.
write
(
texts
+
"\r\n\r\n"
)
texts
=
""
text
=
sentence
[
"Text"
]
texts
+=
text
print
(
"錄音文件識別成功!\n"
)
else
:
print
(
"錄音文件識別失敗!"
)
def
main
(
)
:
# 配置阿里云接口
accessKeyId
=
"填入開通服務的accessKey Id"
accessKeySecret
=
"填入開通服務的accessKey Secret "
appKey
=
"填入開通服務的appKey"
# 輸入錄音url
fileLink
=
"填入上傳至OSS的錄音的url"
# 執行錄音文件識別,識別成功后將輸出recognition.txt文件
fileTrans
(
accessKeyId
,
accessKeySecret
,
appKey
,
fileLink
)
if
__name__
==
'__main__'
:
main
(
)
識別結果
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索:z360901061

微信掃一掃加我為好友
QQ號聯系: 360901061
您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點擊下面給點支持吧,站長非常感激您!手機微信長按不能支付解決辦法:請將微信支付二維碼保存到相冊,切換到微信,然后點擊微信右上角掃一掃功能,選擇支付二維碼完成支付。
【本文對您有幫助就好】元
