[3]小度太弱了,乾脆自己用Python開發個對話機器人【爬蟲,資料庫,面向對象,人工智慧】

背景

麥叔因為百度導航的對話機器人太弱，一次只能講一個笑話，決定用Python開發一個會語音講笑話的機器人，想講幾個笑話就講幾個笑話。我們就稱它為笑笑吧。

本文是Python笑笑語音機器人綜合案例的第3篇。

由於案例比較綜合，整個案例會分多篇文章發出，隔一天發一篇。

前2篇文章，我們已經實現了爬蟲抓取笑話，保存到sqlite資料庫，也實現了基本的交互和2個語音引擎(ttsx和谷歌）。

強烈建議你先讀一下前2篇文章：連結在本文的最下面。

可以隨時在公眾號回覆：笑話，或者笑笑：

用到的技術

本文用到以下技術：

模塊 - 代碼分模塊放在多個文件中 [第2篇、第3篇]語音識別 - 識別用戶輸入的語音，把笑話轉換成語音 [第2篇、第3篇]語音合成

上一篇文章，我們已經實現了基於開源的pyttsx3庫和谷歌庫的兩種語音合成方式。

今天我們來實現基於科大訊飛的語音引擎。科大訊飛是國內頂尖的語音人工智慧功能。

上一篇文章，我們也發了3端語音。猜猜那一段是科大訊飛的？請在評論區留言。A:B:C:

主要流程

整個項目流程：

訊飛語音引擎流程：

代碼結構

為了代碼結構清晰，方便維護，我們根據功能把代碼放到了多個py文件中，每個文件各司其職。

我們把每個語音引擎稱為speaker，所以有個speaker的package(文件夾)保存所有speaker。每個speaker又有自己的文件夾，裡面存放相關代碼，保存生成的語音文件。

xunfei的文件夾下包含以下文件：

xunfei.py - 實現tell_joke接口xunfei_api.py - 調用訊飛的API，實現文本轉換pcm2wav.py - 訊飛返回的是pcm格式，需要轉換成wav格式或其他常見格式才能播放。__ init__.py - 把當前文件夾加到系統的模塊尋址目錄中audios文件夾 - 存放轉化好的笑話，避免多次調用註冊流程

要使用科大訊飛的語音合成服務，需要先註冊會員。每天都500條免費額度，足夠我們玩這個笑話項目了。

登錄訊飛開發平臺網址：

https://www.xfyun.cn/services/online_tts

點擊免費註冊鍵入註冊界面，點立即領取：

有幾種方式，我選擇手機快捷登錄：

創建應用，內容可以隨意填寫：

點應用的名字進入應用詳情：

在左邊可以看到應用的關鍵信息：

請記住這些信息，在代碼中我們需要用到。也可以先把他們複製到記事本。

寫代碼

現在開始寫代碼。在前2篇中已經寫了部分代碼，所以建議先看前2篇。也可以公眾號回復笑話或者笑笑獲取原始碼。

先在speakers目錄下創建文件夾xunfei

把pcm格式轉換成wav格式 pcm2wav.py

訊飛提供的音頻是wav格式，我們需要先把它轉換成常用格式，如wav。為了做這個轉換，我們需要安裝wave包。

在xunfei文件夾創建新文件pcm2wav.py安裝wav包：python -m pip install wave

import sys
import wave
import os

def pcm_to_wav(pcmpath, target_filename):
    '''把pcm格式的音頻轉成wav'''
    with open(pcmpath, 'rb') as pcmfile:
        pcmdata = pcmfile.read()
    with wave.open(target_filename, 'wb') as wavfile:
        wavfile.setparams((2, 2, 8000, 0, 'NONE', 'NONE'))
        wavfile.writeframes(pcmdata)

# 測試代碼
if __name__ == '__main__':
    folder = os.path.dirname(__file__)
    source = f'{folder}/audios/temp.pcm'
    target = f'{folder}/audios/test.wav'
    pcm_to_wav(source, target)
最下面有測試代碼，測試之前請確保audios目錄下有temp.pcm文件。如果沒有，可以去下載我的原始碼。下載方法前面已經提到。
調用訊飛api合成語音 xunfei_api.py
訊飛提供的是基於websocket的API。訊飛提供了一個基本python代碼教我們如何調用它的API做語音合成。
但這個示例程序結構不好，不能動態的指定音頻的名字，而且是不能播放的pcm格式。
所以我對這個音頻進行了優化，主要體現在：
安裝websocket：python -m pip install websocket-client
在xunfei目錄下創建xunfei.py
# -*- coding:utf-8 -*-
#
#   author: iflytek, 麥叔優化版
#
#  本demo測試時運行的環境為：Windows + Python3.7
#  本demo測試成功運行時所安裝的第三方庫及其版本如下：
#   cffi==1.12.3
#   gevent==1.4.0
#   greenlet==0.4.15
#   pycparser==2.19
#   six==1.12.0
#   websocket==0.2.1
#   websocket-client==0.56.0
#   合成小語種需要傳輸小語種文本、使用小語種發音人vcn、tte=unicode以及修改文本編碼方式
#  錯誤碼連結：https://www.xfyun.cn/document/error-code （code返回錯誤碼時必看）
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
import websocket
import datetime
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as thread
import os
import pcm2wav


STATUS_FIRST_FRAME = 0  # 第一幀的標識
STATUS_CONTINUE_FRAME = 1  # 中間幀標識
STATUS_LAST_FRAME = 2  # 最後一幀的標識

# 請把下面3行裡的信息換成前面註冊獲得的APP信息
APP_ID = "******"
APP_KEY = "******"
APP_SECRET = "******"

class Ws_Param(object):
    # 初始化
    def __init__(self, APPID, APIKey, APISecret, Text, FileID):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.Text = Text
        self.FileId = FileID

        # 公共參數(common)
        self.CommonArgs = {"app_id": self.APPID}
        # 業務參數(business)，更多個性化參數可在官網查看
        self.BusinessArgs = {
            "aue": "raw", "auf": "audio/L16;rate=16000", "vcn": "xiaoyan", "tte": "utf8"}
        self.Data = {"status": 2, "text": str(
            base64.b64encode(self.Text.encode('utf-8')), "UTF8")}
        #使用小語種須使用以下方式，此處的unicode指的是 utf16小端的編碼方式，即"UTF-16LE"」
        #self.Data = {"status": 2, "text": str(base64.b64encode(self.Text.encode('utf-16')), "UTF8")}

    # 生成url
    def create_url(self):
        url = 'wss://tts-api.xfyun.cn/v2/tts'
        # 生成RFC1123格式的時間戳
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

        # 拼接字符串
        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + "/v2/tts " + "HTTP/1.1"
        # 進行hmac-sha256進行加密
        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()
        signature_sha = base64.b64encode(
            signature_sha).decode(encoding='utf-8')

        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
            self.APIKey, "hmac-sha256", "host date request-line", signature_sha)
        authorization = base64.b64encode(
            authorization_origin.encode('utf-8')).decode(encoding='utf-8')
        # 將請求的鑑權參數組合為字典
        v = {
            "authorization": authorization,
            "date": date,
            "host": "ws-api.xfyun.cn"
        }
        # 拼接鑑權參數，生成url
        url = url + '?' + urlencode(v)
        # print("date: ",date)
        # print("v: ",v)
        # 此處列印出建立連接時候的url,參考本demo的時候可取消上方列印的注釋，比對相同參數時生成的url與自己代碼生成的url是否一致
        # print('websocket url :', url)
        return url


def on_message(wsParam, ws, message):
    try:
        message = json.loads(message)
        code = message["code"]
        sid = message["sid"]
        audio = message["data"]["audio"]
        audio = base64.b64decode(audio)
        status = message["data"]["status"]
        print(message)
        if status == 2:
            print("ws is closed")
            ws.close()
        if code != 0:
            errMsg = message["message"]
            print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))
        else:
            temp_file = f'{os.path.dirname(__file__)}/audios/{wsParam.FileId}.pcm'
            target_file = f'{os.path.dirname(__file__)}/audios/{wsParam.FileId}.wav'

            with open(temp_file, 'ab') as f:
                f.write(audio)
            #生成wav語音
            pcm2wav.pcm_to_wav(temp_file, target_file)
            

    except Exception as e:
        print("receive msg,but parse exception:", e)


# 收到websocket錯誤的處理
def on_error(ws, error):
    print("### error:", error)


# 收到websocket關閉的處理
def on_close(ws):
    print("### closed ###")


# 收到websocket連接建立的處理
def on_open(wsParam, ws):
    def run(*args):
        d = {"common": wsParam.CommonArgs,
             "business": wsParam.BusinessArgs,
             "data": wsParam.Data,
             }
        d = json.dumps(d)
        print("->開始發送文本數據")
        ws.send(d)
       
        temp_file = f'{os.path.dirname(__file__)}/audios/{wsParam.FileId}.pcm'
        if os.path.exists(temp_file=temp_file):
            os.remove(temp_file)

    thread.start_new_thread(run, ())

def generate_audio(file_id, text):
    # 測試時候在此處正確填寫相關信息即可運行
    wsParam = Ws_Param(APPID=APP_ID, APISecret=APP_SECRET,
                       APIKey=APP_KEY,
                       Text=text, FileID=file_id)
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()
    ws = websocket.WebSocketApp(
        wsUrl, on_error=on_error, on_close=on_close)
    ws.on_open = lambda x:on_open(wsParam, x)
    ws.on_message = lambda x, message: on_message(wsParam, x, message)
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
    


if __name__ == "__main__":
    generate_audio(9527, '我是麥叔，我愛這個世界！')
注意：請把代碼第38-40行的APP_ID等信息換成自己在前面註冊得到的信息。
如果已經安裝好websocket-client庫，直接運行上面的代碼，就會再audios目錄下生成9527.wav。
運行一下9527.wav，對比本文最開始的音頻，判斷一下那個是訊飛的音頻。
實現tell_joke接口: xunfei.py
別忘了，根據我們的設計，任何一個speaker(語音引擎)都需要實現tell_joke接口。我們現在來為訊飛實現tell_joke接口。
在xunfei目錄下創建xunfei.py
import os, sys
import playsound
import time
import xunfei_api

def tell_joke(joke_id, title, detail):
    filename = f"{joke_id}.wav"
    speaker_path = os.path.dirname(__file__)
    full_path = f'{speaker_path}/audios/{filename}'

    #判斷聲音文件是否已經生成過，如果已經存在了，直接返回
    if os.path.exists(full_path):
        print('文件已經生成，直接返回')
        playsound.playsound(full_path, block=True)
    else:
        text = f'{title}, {detail}'
        xunfei_api.generate_audio(joke_id, text)

        #循環判斷，等待文件生成
        while(not os.path.exists(full_path)):
            time.sleep(0.1)
        
        playsound.playsound(full_path, block=True)

if __name__ == '__main__':
    tell_joke(9528, '結婚以後', '女：為什麼從前你對我百依百順，可結婚才三天，你就跟我吵了兩天的架？男：因為我的忍耐是有限度的。')
這個代碼除了調用xunfei_api.py生成語音，還調用了playsound來播放語音。
運行代碼，測試最下面的笑話。
繼承進主程序：joke_ui.py
回到joke_ui.py，引入xunfei.py，注釋掉原來的語音引擎：
import random 
from joke import Joke
import joke_db
# 要使用哪個語音引擎，就反注釋哪個，最開始只有printer，後面陸續實現其他
#from speakers.ttsx.ttsx import tell_joke
#from speakers.google.google import tell_joke
#from speakers.printer import tell_joke
from speakers.xunfei.xunfei import tell_joke
運行tell_joke.ui，會發現程序竟然報錯了。這是因為程序找不到被引用的xunfei_api.py以及pcm2wav.py。所以我們需要下一個步驟：
__ init__.py
每個py文件被稱為一個模塊(module)，每個包含py文件的文件夾被稱為包(package)。
當我們使用import引入一個模塊時，Python解釋器會去預先設定的目錄下去找，或者在當前文件夾下找。
我們運行joke_ui.py的時候，它在當前目錄下，和Python預先設定的目下都找不到，所以就報錯了。我們要做的就是在預先設定的目下添加xunfei這個目錄。
這個添加工作，可以放在xunfei目錄下的__ init__.py文件中。因為當一個目錄(package)下的任何文件被使用時，都會先運行init.py。
在xunfei目錄下添加__ init__.py
import sys, os

# 在Python的尋址路徑中添加當前目錄
sys.path.append(os.path.dirname(__file__))
再次運行joke_ui.py，就可以正常講笑話啦！
這樣我們的語音引擎部分就結束了，我們實現了3種不同的語音引擎，他們各有優點：
ttsx是免費的，本地的引擎，可以無限免費使用且速度很快。訊飛的引擎似乎更有中國味，但API的易用性和谷歌有差距。未完待續
這個綜合的實現案例，我們還差兩部：
下一篇文章會在後天發出，請保持關注！
結尾
可以公眾號回覆：笑話，獲得系列文章連結，原始碼下載地址，加入實戰討論群。
精品回顧：
[2]小度太弱了，乾脆自己開發個對話機器人【爬蟲，資料庫，面向對象，人工智慧】
[1]小度太弱了，乾脆自己開發個對話機器人【爬蟲，資料庫，面向對象，人工智慧】
2個程式設計師1周搞定4萬塊小程序項目的技術要點
搞瘋爬蟲程式設計師的8個難點！
Markdown Nice 最全功能介紹
我是麥叔：教你學編程，陪你走職場的路！