一、前言說明
1.功能簡述
Python學(xué)習(xí)資料或者需要代碼、視頻加Python學(xué)習(xí)群：960410445
登錄后進(jìn)入聊天界面，如果服務(wù)器都在同一個(gè)地址，則都進(jìn)入同一個(gè)房間
進(jìn)入/離開/發(fā)消息同一房間用戶都可以看到，輸入“tuling”或“chatbot”可以切換為和Tuling機(jī)器人或者ChatBot聊天
按住Say說話，自動(dòng)錄音并轉(zhuǎn)為文本發(fā)送
如果是跟機(jī)器人聊天，則自動(dòng)回復(fù)文本并播放文本語音
Tuling，是圖靈，已經(jīng)錄入大量中文對話，直接調(diào)用接口即可實(shí)現(xiàn)自動(dòng)回復(fù)，實(shí)用于開發(fā)聊天軟件
ChatBot，可自行訓(xùn)練機(jī)器人，讓機(jī)器人擁有自己的語料庫，實(shí)用于企業(yè)智能聊天個(gè)性化
2.需要的核心技術(shù)
a. 輸入語音，識(shí)別后轉(zhuǎn)換為輸入文字
b. Tuling或ChatBot兩種機(jī)器人，回復(fù)輸出文字
c. 輸出文字，轉(zhuǎn)換為輸出語音并播放
以上a和c主要調(diào)用Baidu提供的API進(jìn)行轉(zhuǎn)換，如果理解本文，可自行嘗試調(diào)用Google提供的API實(shí)現(xiàn)，Google技術(shù)強(qiáng)大但是否對中文支持良好，博主不曾嘗試不妄自揣度
3.環(huán)境說明
系統(tǒng)環(huán)境Win10，運(yùn)行環(huán)境Python3.6，運(yùn)行工具Pycharm
二、源碼設(shè)計(jì)（貼上完整源碼自己去理解）
1.運(yùn)行順序
?。?）TrainChat.py訓(xùn)練本地chatbot機(jī)器人（每次更新訓(xùn)練內(nèi)容，運(yùn)行一次即可）
?。?） server.py開啟服務(wù)器
（3）client.py運(yùn)行n次，每次運(yùn)行都可登陸一個(gè)用戶
2.服務(wù)器server.py??
主要處理用戶的登錄校驗(yàn)，房間的人員消息處理
此處通過config.py中配置的列表PORT = range(1, 3)生成兩個(gè)房間，地址分別是127.0.0.1:1和127.0.0.1:2（實(shí)用時(shí)可以無限個(gè)）
啟用客戶端前，這個(gè)服務(wù)器要先運(yùn)行，代碼中CommandHandler類拆解client客戶端發(fā)送的信息中的命令，并綁定函數(shù)
import asynchat
import asyncore
from config import PORT
# 定義結(jié)束異常類
class EndSession(Exception):
? ? pass
class ChatServer(asyncore.dispatcher):
? ? """
? ? 聊天服務(wù)器
? ? """
? ? def __init__(self, port):
? ? ? ? asyncore.dispatcher.__init__(self)
? ? ? ? # 創(chuàng)建socket
? ? ? ? self.create_socket()
? ? ? ? # 設(shè)置 socket 為可重用
? ? ? ? self.set_reuse_addr()
? ? ? ? # 監(jiān)聽端口
? ? ? ? self.bind(('', port))
? ? ? ? self.listen(5)
? ? ? ? self.users = {}
? ? ? ? self.main_room = ChatRoom(self)
? ? def handle_accept(self):
? ? ? ? conn, addr = self.accept()
? ? ? ? ChatSession(self, conn)
class ChatSession(asynchat.async_chat):
? ? """
? ? 負(fù)責(zé)和客戶端通信
? ? """
? ? def __init__(self, server, sock):
? ? ? ? asynchat.async_chat.__init__(self, sock)
? ? ? ? self.server = server
? ? ? ? self.set_terminator(b'\n')
? ? ? ? self.data = []
? ? ? ? self.name = None
? ? ? ? self.enter(LoginRoom(server))
? ? def enter(self, room):
? ? ? ? # 從當(dāng)前房間移除自身，然后添加到指定房間
? ? ? ? try:
? ? ? ? ? ? cur = self.room
? ? ? ? except AttributeError:
? ? ? ? ? ? pass
? ? ? ? else:
? ? ? ? ? ? cur.remove(self)
? ? ? ? self.room = room
? ? ? ? room.add(self)
? ? def collect_incoming_data(self, data):
? ? ? ? # 接收客戶端的數(shù)據(jù)
? ? ? ? self.data.append(data.decode("utf-8"))
? ? def found_terminator(self):
? ? ? ? # 當(dāng)客戶端的一條數(shù)據(jù)結(jié)束時(shí)的處理
? ? ? ? line = ''.join(self.data)
? ? ? ? self.data = []
? ? ? ? try:
? ? ? ? ? ? self.room.handle(self, line.encode("utf-8"))
? ? ? ? # 退出聊天室的處理
? ? ? ? except EndSession:
? ? ? ? ? ? self.handle_close()
? ? def handle_close(self):
? ? ? ? # 當(dāng) session 關(guān)閉時(shí)，將進(jìn)入 LogoutRoom
? ? ? ? asynchat.async_chat.handle_close(self)
? ? ? ? self.enter(LogoutRoom(self.server))
class CommandHandler:
? ? """
? ? 命令處理類
? ? """
? ? def unknown(self, session, cmd):
? ? ? ? # 響應(yīng)未知命令
? ? ? ? # 通過 aynchat.async_chat.push 方法發(fā)送消息
? ? ? ? session.push(('Unknown command {} \n'.format(cmd)).encode("utf-8"))
? ? def handle(self, session, line):
? ? ? ? line = line.decode()
? ? ? ? # 命令處理
? ? ? ? if not line.strip():
? ? ? ? ? ? return
? ? ? ? parts = line.split(' ', 1)
? ? ? ? cmd = parts[0]
? ? ? ? try:
? ? ? ? ? ? line = parts[1].strip()
? ? ? ? except IndexError:
? ? ? ? ? ? line = ''
? ? ? ? # 通過協(xié)議代碼執(zhí)行相應(yīng)的方法
? ? ? ? method = getattr(self, 'do_' + cmd, None)
? ? ? ? try:
? ? ? ? ? ? method(session, line)
? ? ? ? except TypeError:
? ? ? ? ? ? self.unknown(session, cmd)
class Room(CommandHandler):
? ? """
? ? 包含多個(gè)用戶的環(huán)境，負(fù)責(zé)基本的命令處理和廣播
? ? """
? ? def __init__(self, server):
? ? ? ? self.server = server
? ? ? ? self.sessions = []
? ? def add(self, session):
? ? ? ? # 一個(gè)用戶進(jìn)入房間
? ? ? ? self.sessions.append(session)
? ? def remove(self, session):
? ? ? ? # 一個(gè)用戶離開房間
? ? ? ? self.sessions.remove(session)
? ? def broadcast(self, line):
? ? ? ? # 向所有的用戶發(fā)送指定消息
? ? ? ? # 使用 asynchat.asyn_chat.push 方法發(fā)送數(shù)據(jù)
? ? ? ? for session in self.sessions:
? ? ? ? ? ? session.push(line)
? ? def do_logout(self, session, line):
? ? ? ? # 退出房間
? ? ? ? raise EndSession
class LoginRoom(Room):
? ? """
? ? 處理登錄用戶
? ? """
? ? def add(self, session):
? ? ? ? # 用戶連接成功的回應(yīng)
? ? ? ? Room.add(self, session)
? ? ? ? # 使用 asynchat.asyn_chat.push 方法發(fā)送數(shù)據(jù)
? ? ? ? session.push(b'Connect Success')
? ? def do_login(self, session, line):
? ? ? ? # 用戶登錄邏輯
? ? ? ? name = line.strip()
? ? ? ? # 獲取用戶名稱
? ? ? ? if not name:
? ? ? ? ? ? session.push(b'UserName Empty')
? ? ? ? # 檢查是否有同名用戶
? ? ? ? elif name in self.server.users:
? ? ? ? ? ? session.push(b'UserName Exist')
? ? ? ? # 用戶名檢查成功后，進(jìn)入主聊天室
? ? ? ? else:
? ? ? ? ? ? session.name = name
? ? ? ? ? ? session.enter(self.server.main_room)
class LogoutRoom(Room):
? ? """
? ? 處理退出用戶
? ? """
? ? def add(self, session):
? ? ? ? # 從服務(wù)器中移除
? ? ? ? try:
? ? ? ? ? ? del self.server.users[session.name]
? ? ? ? except KeyError:
? ? ? ? ? ? pass
class ChatRoom(Room):
? ? """
? ? 聊天用的房間
? ? """
? ? def add(self, session):
? ? ? ? # 廣播新用戶進(jìn)入
? ? ? ? session.push(b'Login Success')
? ? ? ? self.broadcast((session.name + ' has entered the room.\n').encode("utf-8"))
? ? ? ? self.server.users[session.name] = session
? ? ? ? Room.add(self, session)
? ? def remove(self, session):
? ? ? ? # 廣播用戶離開
? ? ? ? Room.remove(self, session)
? ? ? ? self.broadcast((session.name + ' has left the room.\n').encode("utf-8"))
? ? def do_say(self, session, line):
? ? ? ? # 客戶端發(fā)送消息
? ? ? ? self.broadcast((session.name + ': ' + line + '\n').encode("utf-8"))
? ? def do_noone_say(self, session, line):
? ? ? ? # 圖靈回復(fù)消息
? ? ? ? self.broadcast((line + '\n').encode("utf-8"))
? ? def do_chatbot_say(self, session, line):
? ? ? ? # 圖靈回復(fù)消息
? ? ? ? self.broadcast(('ChatBot: ' + line + '\n').encode("utf-8"))
? ? def do_tuling_say(self, session, line):
? ? ? ? # 圖靈回復(fù)消息
? ? ? ? self.broadcast(('Tuling: ' + line + '\n').encode("utf-8"))
? ? def do_look(self, session, line):
? ? ? ? # 查看在線用戶
? ? ? ? session.push(b'All Online Users Are:\n')
? ? ? ? for other in self.sessions:
? ? ? ? ? ? session.push((other.name + '\n').encode("utf-8"))
if __name__ == '__main__':
? ? for i in range(len(PORT)):
? ? ? ? ChatServer(PORT[i])
? ? ? ? print("Chat server run at '127.0.0.1:{0}'".format(PORT[i]))
? ? try:
? ? ? ? asyncore.loop()
? ? except KeyboardInterrupt:
? ? ? ? print("Chat server exit")
server.py
3.訓(xùn)練chatbot的TrainChat.py
主要用來訓(xùn)練chatbot機(jī)器人，數(shù)據(jù)保存在本地sqlite數(shù)據(jù)庫（如果沒有數(shù)據(jù)庫自動(dòng)創(chuàng)建）
個(gè)人學(xué)習(xí)此數(shù)據(jù)足以，作為企業(yè)可改為mongodb保存數(shù)據(jù)，速度會(huì)有保障
#!/usr/bin/python
# -*- coding: utf-8 -*-
from chatterbot import ChatBot
from chatterbot.trainers import ListTrainer
from chatterbot.trainers import ChatterBotCorpusTrainer
my_bot = ChatBot("Training demo",
? ? ? ? ? ? ? ? database="./db.sqlite3")
# 直接寫語句訓(xùn)練
my_bot.set_trainer(ListTrainer)
my_bot.train(["你叫什么名字？", "我叫小白兔！", ])
my_bot.train([
? ? "Test1",
? ? "Test2",
? ? "Test3",
? ? "Test4",
])
# 使用自定義語句訓(xùn)練它
my_bot.set_trainer(ChatterBotCorpusTrainer)
my_bot.train("chatterbot.corpus.mytrain")
# while True:
#? ? print(my_bot.get_response(input("user:")))
TrainChat.py
4.訓(xùn)練chatbot的語料庫
提供了兩種語料訓(xùn)練方法
?。?）TrainChat.py里面可以直接寫訓(xùn)練語句，也可開啟通過聊天時(shí)候的語句自動(dòng)訓(xùn)練
（2）自定義語料庫訓(xùn)練，自定義語料格式，直接參照chatbot提供的一些寫就行
找到安裝chatbot后默認(rèn)提供的中文語料格式D:\Python\Lib\site-packages\chatterbot_corpus\data\chinese
打開后格式就有了，這里我們按照格式新增一個(gè)mytrain文件夾，寫入自己的語料文件，如我寫的phone.yml
categories:
- phone
conversations:
- - iPhoneX
? - iPhone X是Apple（蘋果公司）于北京時(shí)間2017年9月13日凌晨1點(diǎn)，在Apple Park新總部的史蒂夫·喬布斯劇院會(huì)上發(fā)布的新機(jī)型。其中“X”是羅馬數(shù)字“10”的意思，代表著蘋果向iPhone問世十周年致敬。iPhone X屬于高端版機(jī)型，采用全新設(shè)計(jì)，搭載色彩銳利的OLED屏幕，配備升級(jí)后的相機(jī)，使用3D面部識(shí)別（Face ID）傳感器解鎖手機(jī)，支持AirPower(空中能量)無線充電。分為64GB、256GB兩個(gè)版本，中國大陸起售價(jià)8388人民幣，美國起售價(jià)999美元，2017年10月27日預(yù)售，11月3號(hào)正式開賣。
- - 三星Galaxy S6
? - 三星Galaxy S6是三星公司（SAMSUNG）在2015年3月2日推出的一款手機(jī)，已于2015年4月11日正式上市。\n三星Galaxy S6采用5.1英寸屏幕，2560×1440像素，像素密度高達(dá)573ppi，內(nèi)置Exynos 7420八核64位處理器，能夠提供更強(qiáng)的性能以及更低的功耗；采用前500W像素+后1600W像素的雙鏡頭設(shè)計(jì)，均支持F1.9大光圈，感光元件是索尼IMX 240，支持OIS光學(xué)防抖和自動(dòng)HDR技術(shù)。
- - 華為P8
? - P系列是華為手機(jī)中的旗艦系列，到2017年1月，共有6款機(jī)型：P1、P2、P6、P7、P8、P8 MAX、P9、P9 Plus。從2012年1月11日在美國拉斯維加斯發(fā)布全球最薄6.68毫米的P1開始，P系列便創(chuàng)立了以驚艷ID設(shè)計(jì)融合強(qiáng)大均衡軟硬件配置為主的旗艦產(chǎn)品地位。之后，華為于2013年6月18日發(fā)布P6，2014年5月7日發(fā)布P7，均分別輕松創(chuàng)下了數(shù)百萬銷量的佳績，一舉奠定了華為在國內(nèi)領(lǐng)先、國際一流的品牌地位
phone.yml
5.錄音并保存文件recorder.py
提供錄音功能并將錄音文件保存在本地
#!/usr/bin/python3
# -*- coding: utf-8 -*-
from pyaudio import PyAudio, paInt16
import numpy as np
from datetime import datetime
import wave
import sys
import time
class Recoder:
? ? NUM_SAMPLES = 2000? ? ? # py audio內(nèi)置緩沖大小
? ? SAMPLING_RATE = 8000? ? # 取樣頻率
? ? LEVEL = 500? ? ? ? # 聲音保存的閾值
? ? COUNT_NUM = 20? ? ? # NUM_SAMPLES個(gè)取樣之內(nèi)出現(xiàn)COUNT_NUM個(gè)大于LEVEL的取樣則記錄聲音
? ? SAVE_LENGTH = 8? ? ? ? # 聲音記錄的最小長度：SAVE_LENGTH * NUM_SAMPLES 個(gè)取樣
? ? TIME_COUNT = 10? ? # 錄音時(shí)間，單位s
? ? Voice_String = []
? ? def savewav(self, filename):
? ? ? ? wf = wave.open(filename, 'wb')
? ? ? ? wf.setnchannels(1)
? ? ? ? wf.setsampwidth(2)
? ? ? ? wf.setframerate(self.SAMPLING_RATE)
? ? ? ? wf.writeframes(np.array(self.Voice_String).tostring())
? ? ? ? # wf.writeframes(self.Voice_String.decode())
? ? ? ? wf.close()
? ? def recoder(self):
? ? ? ? pa = PyAudio()
? ? ? ? stream = pa.open(format=paInt16, channels=1, rate=self.SAMPLING_RATE, input=True,
? ? ? ? ? ? ? ? ? ? ? ? frames_per_buffer=self.NUM_SAMPLES)
? ? ? ? save_count = 0
? ? ? ? save_buffer = []
? ? ? ? time_count = self.TIME_COUNT
? ? ? ? while True:
? ? ? ? ? ? time_count -= 1
? ? ? ? ? ? # print time_count
? ? ? ? ? ? # 讀入NUM_SAMPLES個(gè)取樣
? ? ? ? ? ? string_audio_data = stream.read(self.NUM_SAMPLES)
? ? ? ? ? ? # 將讀入的數(shù)據(jù)轉(zhuǎn)換為數(shù)組
? ? ? ? ? ? audio_data = np.fromstring(string_audio_data, dtype=np.short)
? ? ? ? ? ? # 計(jì)算大于LEVEL的取樣的個(gè)數(shù)
? ? ? ? ? ? large_sample_count = np.sum( audio_data > self.LEVEL )
? ? ? ? ? ? print(np.max(audio_data))
? ? ? ? ? ? # 如果個(gè)數(shù)大于COUNT_NUM，則至少保存SAVE_LENGTH個(gè)塊
? ? ? ? ? ? if large_sample_count > self.COUNT_NUM:
? ? ? ? ? ? ? ? save_count = self.SAVE_LENGTH
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? save_count -= 1
? ? ? ? ? ? if save_count < 0:
? ? ? ? ? ? ? ? save_count = 0
? ? ? ? ? ? if save_count > 0:
? ? ? ? ? ? ? ? # 將要保存的數(shù)據(jù)存放到save_buffer中
? ? ? ? ? ? ? ? # print? save_count > 0 and time_count >0
? ? ? ? ? ? ? ? save_buffer.append(string_audio_data )
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? # print save_buffer
? ? ? ? ? ? ? ? # 將save_buffer中的數(shù)據(jù)寫入WAV文件，WAV文件的文件名是保存的時(shí)刻
? ? ? ? ? ? ? ? # print "debug"
? ? ? ? ? ? ? ? if len(save_buffer) > 0 :
? ? ? ? ? ? ? ? ? ? self.Voice_String = save_buffer
? ? ? ? ? ? ? ? ? ? save_buffer = []
? ? ? ? ? ? ? ? ? ? print("Recode a piece of? voice successfully!")
? ? ? ? ? ? ? ? ? ? return True
? ? ? ? ? ? if time_count == 0:
? ? ? ? ? ? ? ? if len(save_buffer)>0:
? ? ? ? ? ? ? ? ? ? self.Voice_String = save_buffer
? ? ? ? ? ? ? ? ? ? save_buffer = []
? ? ? ? ? ? ? ? ? ? print("Recode a piece of? voice successfully!")
? ? ? ? ? ? ? ? ? ? return True
? ? ? ? ? ? ? ? else:
? ? ? ? ? ? ? ? ? ? return False
def recording():
? ? r = Recoder()
? ? r.recoder()
? ? r.savewav(r"E:\Python_Doc\voice_say\say_voice.wav")
recorder.py
6. chatbot.py
提供播放音頻文件
調(diào)用圖靈Tuling接口返回文本信息
調(diào)用chatbot返回文本信息
調(diào)用百度api語音識(shí)別
調(diào)用百度api轉(zhuǎn)文本為語音(有兩個(gè)百度api都可用，第一個(gè)不用密匙)，其中chatbot的數(shù)據(jù)庫配置要和TrainChat.py中配置的名稱一致
import pygame
from chatterbot import ChatBot
import requests
import json
from config import *
import time
import os
import random
import urllib.request
import base64
# 初始化百度返回的音頻文件地址，后面會(huì)變?yōu)槿肿兞?，隨需改變
mp3_url = 'E:\Python_Doc\\voice_du\\voice_ss.mp3'
# 播放Mp3文件
def play_mp3():
? ? # 接受服務(wù)器的消息
? ? pygame.mixer.init()
? ? pygame.mixer.music.load(mp3_url)
? ? pygame.mixer.music.play()
? ? while pygame.mixer.music.get_busy():
? ? ? ? time.sleep(1)
? ? pygame.mixer.music.stop()
? ? pygame.mixer.quit()
# 刪除聲音文件
def remove_voice():
? ? path = r"E:\Python_Doc\voice_du"
? ? for i in os.listdir(path):
? ? ? ? path_file = os.path.join(path, i)
? ? ? ? try:
? ? ? ? ? ? os.remove(path_file)
? ? ? ? except:
? ? ? ? ? ? continue
# 圖靈自動(dòng)回復(fù)
def tuling(info):
? ? url = tuling_url + "?key=%s&info=%s" % (tuling_app_key, info)
? ? content = requests.get(url, headers=headers)
? ? answer = json.loads(content.text)
? ? return answer['text']
# 聊天機(jī)器人回復(fù)
def chatbot(info):
? ? my_bot = ChatBot("", read_only=True,
? ? ? ? ? ? ? ? ? ? database="./db.sqlite3")
? ? res = my_bot.get_response(info)
? ? return str(res)
# 百度講文本轉(zhuǎn)為聲音文件保存在本地 tts地址，無需token實(shí)時(shí)認(rèn)證
def baidu_api(answer):
? ? api_url = '{11}?idx={0}&tex={1}&cuid={2}&cod={3}&lan={4}&ctp={5}&pdt={6}&spd={7}&per={8}&vol={9}&pit={10}'\
? ? ? ? .format(baidu_api_set["idx"], answer, baidu_api_set["cuid"], baidu_api_set["cod"], baidu_api_set["lan"],
? ? ? ? ? ? ? ? baidu_api_set["ctp"], baidu_api_set["pdt"], baidu_api_set["spd"], baidu_api_set["per"],
? ? ? ? ? ? ? ? baidu_api_set["vol"], baidu_api_set["pit"], baidu_api_url)
? ? res = requests.get(api_url, headers=headers2)
? ? # 本地Mp3語音文件保存位置
? ? iname = random.randrange(1, 99999)
? ? global mp3_url
? ? mp3_url = 'E:\Python_Doc\\voices\\voice_tts' + str(iname) + '.mp3'
? ? with open(mp3_url, 'wb') as f:
? ? ? ? f.write(res.content)
# 百度講文本轉(zhuǎn)為聲音文件保存在本地 方法2 tsn地址
def baidu_api2(answer):
? ? # 獲取access_token
? ? token = getToken()
? ? get_url = baidu_api_url2 % (urllib.parse.quote(answer), "test", token)
? ? voice_data = urllib.request.urlopen(get_url).read()
? ? # 本地Mp3語音文件保存位置
? ? name = random.randrange(1, 99999)
? ? global mp3_url
? ? mp3_url = 'E:\Python_Doc\\voice_du\\voice_tsn' + str(name) + '.mp3'
? ? voice_fp = open(mp3_url, 'wb+')
? ? voice_fp.write(voice_data)
? ? voice_fp.close()
? ? return
# 百度語音轉(zhuǎn)文本
def getText(filename):
? ? # 獲取access_token
? ? token = getToken()
? ? data = {}
? ? data['format'] = 'wav'
? ? data['rate'] = 16000
? ? data['channel'] = 1
? ? data['cuid'] = str(random.randrange(123456, 999999))
? ? data['token'] = token
? ? wav_fp = open(filename, 'rb')
? ? voice_data = wav_fp.read()
? ? data['len'] = len(voice_data)
? ? data['speech'] = base64.b64encode(voice_data).decode('utf-8')
? ? post_data = json.dumps(data)
? ? # 語音識(shí)別的api url
? ? upvoice_url = 'http://vop.baidu.com/server_api'
? ? r_data = urllib.request.urlopen(upvoice_url, data=bytes(post_data, encoding="utf-8")).read()
? ? print(json.loads(r_data))
? ? err = json.loads(r_data)['err_no']
? ? if err == 0:
? ? ? ? return json.loads(r_data)['result'][0]
? ? else:
? ? ? ? return json.loads(r_data)['err_msg']
# 獲取百度API調(diào)用的認(rèn)證，實(shí)時(shí)生成，因?yàn)橛袝r(shí)間限制
def getToken():
? ? # token認(rèn)證的url
? ? api_url = "https://openapi.baidu.com/oauth/2.0/token?" \
? ? ? ? ? ? ? ? ? ? "grant_type=client_credentials&client_id=%s&client_secret=%s"
? ? token_url = api_url % (BaiDu_API_Key_GetVoi, BaiDu_Secret_Key_GetVoi)
? ? r_str = urllib.request.urlopen(token_url).read()
? ? token_data = json.loads(r_str)
? ? token_str = token_data['access_token']
? ? return token_str
chatbot.py
7.client.py
提供登錄窗口，聊天窗口，已及響應(yīng)事件
say按鈕綁定sayDown錄音和sayUp獲取語音文本并發(fā)送兩個(gè)事件
Users顯示當(dāng)前房間所有用戶...
import wx
import telnetlib
from time import sleep
import _thread as thread
import time
import os
from chatbot import baidu_api2, chatbot, tuling, play_mp3, remove_voice, getText
from config import BOTS, BOT, default_server, VOICE_SWITCH
from recorder import recording
bot_use = BOT
class LoginFrame(wx.Frame):
? ? """
? ? 登錄窗口
? ? """
? ? def __init__(self, parent, id, title, size):
? ? ? ? # 初始化，添加控件并綁定事件
? ? ? ? wx.Frame.__init__(self, parent, id, title)
? ? ? ? self.SetSize(size)
? ? ? ? self.Center()
? ? ? ? self.serverAddressLabel = wx.StaticText(self, label="Server Address", pos=(15, 40), size=(120, 25))
? ? ? ? self.userNameLabel = wx.StaticText(self, label="UserName", pos=(45, 90), size=(120, 25))
? ? ? ? self.serverAddress = wx.TextCtrl(self, value=default_server,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? pos=(120, 37), size=(150, 25), style=wx.TE_PROCESS_ENTER)
? ? ? ? self.userName = wx.TextCtrl(self, pos=(120, 87), size=(150, 25), style=wx.TE_PROCESS_ENTER)
? ? ? ? self.loginButton = wx.Button(self, label='Login', pos=(50, 145), size=(90, 30))
? ? ? ? self.exitButton = wx.Button(self, label='Exit', pos=(180, 145), size=(90, 30))
? ? ? ? # 綁定登錄方法
? ? ? ? self.loginButton.Bind(wx.EVT_BUTTON, self.login)
? ? ? ? # 綁定退出方法
? ? ? ? self.exitButton.Bind(wx.EVT_BUTTON, self.exit)
? ? ? ? # 服務(wù)器輸入框Tab事件
? ? ? ? self.serverAddress.SetFocus()
? ? ? ? self.Bind(wx.EVT_TEXT_ENTER, self.usn_focus, self.serverAddress)
? ? ? ? # 用戶名回車登錄
? ? ? ? self.Bind(wx.EVT_TEXT_ENTER, self.login, self.userName)
? ? ? ? self.Show()
? ? # 回車調(diào)到用戶名輸入欄
? ? def usn_focus(self, event):
? ? ? ? self.userName.SetFocus()
? ? def login(self, event):
? ? ? ? # 登錄處理
? ? ? ? try:
? ? ? ? ? ? serverAddress = self.serverAddress.GetLineText(0).split(':')
? ? ? ? ? ? con.open(serverAddress[0], port=int(serverAddress[1]), timeout=10)
? ? ? ? ? ? response = con.read_some()
? ? ? ? ? ? if response != b'Connect Success':
? ? ? ? ? ? ? ? self.showDialog('Error', 'Connect Fail!', (200, 100))
? ? ? ? ? ? ? ? return
? ? ? ? ? ? con.write(('login ' + str(self.userName.GetLineText(0)) + '\n').encode("utf-8"))
? ? ? ? ? ? response = con.read_some()
? ? ? ? ? ? if response == b'UserName Empty':
? ? ? ? ? ? ? ? self.showDialog('Error', 'UserName Empty!', (200, 100))
? ? ? ? ? ? elif response == b'UserName Exist':
? ? ? ? ? ? ? ? self.showDialog('Error', 'UserName Exist!', (200, 100))
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? self.Close()
? ? ? ? ? ? ? ? ChatFrame(None, 2, title='當(dāng)前用戶：'+str(self.userName.GetLineText(0)), size=(515, 400))
? ? ? ? except Exception:
? ? ? ? ? ? self.showDialog('Error', 'Connect Fail!', (95, 20))
? ? def exit(self, event):
? ? ? ? self.Close()
? ? # 顯示錯(cuò)誤信息對話框
? ? def showDialog(self, title, content, size):
? ? ? ? dialog = wx.Dialog(self, title=title, size=size)
? ? ? ? dialog.Center()
? ? ? ? wx.StaticText(dialog, label=content)
? ? ? ? dialog.ShowModal()
class ChatFrame(wx.Frame):
? ? """
? ? 聊天窗口
? ? """
? ? def __init__(self, parent, id, title, size):
? ? ? ? # 初始化，添加控件并綁定事件
? ? ? ? wx.Frame.__init__(self, parent, id, title, style=wx.SYSTEM_MENU | wx.CAPTION | wx.CLOSE_BOX |
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? wx.DEFAULT_FRAME_STYLE)
? ? ? ? self.SetSize(size)
? ? ? ? self.Center()
? ? ? ? self.chatFrame = wx.TextCtrl(self, pos=(5, 5), size=(490, 310), style=wx.TE_MULTILINE | wx.TE_READONLY)
? ? ? ? self.sayButton = wx.Button(self, label="Say", pos=(5, 320), size=(58, 25))
? ? ? ? self.message = wx.TextCtrl(self, pos=(65, 320), size=(240, 25), style=wx.TE_PROCESS_ENTER)
? ? ? ? self.sendButton = wx.Button(self, label="Send", pos=(310, 320), size=(58, 25))
? ? ? ? self.usersButton = wx.Button(self, label="Users", pos=(373, 320), size=(58, 25))
? ? ? ? self.closeButton = wx.Button(self, label="Close", pos=(436, 320), size=(58, 25))
? ? ? ? # 發(fā)送按鈕綁定發(fā)送消息方法
? ? ? ? self.sendButton.Bind(wx.EVT_BUTTON, self.send)
? ? ? ? # 輸入框回車發(fā)送信息
? ? ? ? self.message.SetFocus()
? ? ? ? # 發(fā)送消息
? ? ? ? self.sayButton.Bind(wx.EVT_LEFT_DOWN, self.sayDown)
? ? ? ? self.sayButton.Bind(wx.EVT_LEFT_UP, self.sayUp)
? ? ? ? # 發(fā)送消息
? ? ? ? self.Bind(wx.EVT_TEXT_ENTER, self.send, self.message)
? ? ? ? # Users按鈕綁定獲取在線用戶數(shù)量方法
? ? ? ? self.usersButton.Bind(wx.EVT_BUTTON, self.lookUsers)
? ? ? ? # 關(guān)閉按鈕綁定關(guān)閉方法
? ? ? ? self.closeButton.Bind(wx.EVT_BUTTON, self.close)
? ? ? ? thread.start_new_thread(self.receive, ())
? ? ? ? # self.ShowFullScreen(True)
? ? ? ? self.Show()
? ? def sayDown(self, event):
? ? ? ? thread.start_new_thread(recording, ())
? ? ? ? # print("ON")
? ? def sayUp(self, event):
? ? ? ? sayText = getText(r"E:\Python_Doc\voice_say\say_voice.wav")
? ? ? ? self.message.AppendText(str(sayText))
? ? ? ? self.send(self)
? ? def send(self, event):
? ? ? ? # 發(fā)送消息
? ? ? ? message = str(self.message.GetLineText(0)).strip()
? ? ? ? global bot_use
? ? ? ? if message != '':
? ? ? ? ? ? if message == "chatbot":
? ? ? ? ? ? ? ? bot_use = "ChatBot"
? ? ? ? ? ? ? ? self.message.Clear()
? ? ? ? ? ? ? ? con.write(('noone_say You have been changed ChatBot-Chat' + '\n').encode("utf-8"))
? ? ? ? ? ? ? ? return
? ? ? ? ? ? elif message == "tuling":
? ? ? ? ? ? ? ? bot_use = "TuLing"
? ? ? ? ? ? ? ? self.message.Clear()
? ? ? ? ? ? ? ? con.write(('noone_say You have been changed TuLing-Chat' + '\n').encode("utf-8"))
? ? ? ? ? ? ? ? return
? ? ? ? ? ? elif message == "user":
? ? ? ? ? ? ? ? bot_use = "User"
? ? ? ? ? ? ? ? self.message.Clear()
? ? ? ? ? ? ? ? con.write(('noone_say You have been changed User-Chat' + '\n').encode("utf-8"))
? ? ? ? ? ? ? ? return
? ? ? ? ? ? con.write(('say ' + message + '\n').encode("utf-8"))
? ? ? ? ? ? self.message.Clear()
? ? ? ? ? ? # 機(jī)器人回復(fù)
? ? ? ? ? ? if bot_use == "ChatBot":
? ? ? ? ? ? ? ? answer = chatbot(message)
? ? ? ? ? ? ? ? con.write(('chatbot_say ' + answer + '\n').encode("utf-8"))
? ? ? ? ? ? elif bot_use == "TuLing":
? ? ? ? ? ? ? ? answer = tuling(message)
? ? ? ? ? ? ? ? con.write(('tuling_say ' + answer + '\n').encode("utf-8"))
? ? ? ? ? ? elif bot_use == "User":
? ? ? ? ? ? ? ? return
? ? ? ? ? ? if VOICE_SWITCH:
? ? ? ? ? ? ? ? # 寫本地音樂文件
? ? ? ? ? ? ? ? baidu_api2(answer)
? ? ? ? ? ? ? ? # 新建線程播放音樂
? ? ? ? ? ? ? ? thread.start_new_thread(play_mp3, ())
? ? ? ? return
? ? def lookUsers(self, event):
? ? ? ? # 查看當(dāng)前在線用戶
? ? ? ? con.write(b'look\n')
? ? def close(self, event):
? ? ? ? # 關(guān)閉窗口
? ? ? ? thread.start_new_thread(remove_voice, ())
? ? ? ? con.write(b'logout\n')
? ? ? ? con.close()
? ? ? ? self.Close()
? ? def receive(self):
? ? ? ? # 接受服務(wù)器的消息
? ? ? ? while True:
? ? ? ? ? ? sleep(0.6)
? ? ? ? ? ? result = con.read_very_eager()
? ? ? ? ? ? if result != '':
? ? ? ? ? ? ? ? self.chatFrame.AppendText(result)
if __name__ == '__main__':
? ? app = wx.App()
? ? con = telnetlib.Telnet()
? ? LoginFrame(None, -1, title="Login", size=(320, 250))
? ? app.MainLoop()
client.py
8.config配置文件
百度API的KEY等內(nèi)容也可自行去對應(yīng)官網(wǎng)申請，本文提供僅供學(xué)習(xí)使用
# 默認(rèn)輸入的服務(wù)器地址，測試時(shí)候使用，避免登錄總是輸入地址麻煩
default_server = "127.0.0.1:1"
# 定義服務(wù)器端口，一個(gè)端口一個(gè)房間
PORT = range(1, 3)
# 圖靈Tuling機(jī)器人還是ChatBot聊天機(jī)器人選擇
BOTS = ["TuLing", "ChatBot", "User"]
BOT = BOTS[2]
# 瀏覽器請求頭文件
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 '
? ? ? ? ? ? ? ? ? ? ? ? '(KHTML, like Gecko) Chrome/53.0.2785.143 Safari/537.36', }
headers2 = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 '
? ? ? ? ? ? ? ? ? ? ? ? ? '(KHTML, like Gecko)Chrome/62.0.3202.94 Safari/537.36'}
# 圖靈密匙，自動(dòng)回復(fù)地址，選擇的key不同，tuling機(jī)器人的回答也各不相同
tuling_app_key = "e5ccc9c7c8834ec3b08940e290ff1559"
tuling_app_key2 = "4bc32d41c10be18627438ae45eb839ac"
tuling_url = "http://www.tuling123.com/openapi/api"
# 語音保存播放開關(guān)
VOICE_SWITCH = True
# 百度文本轉(zhuǎn)語音地址和配置 tts地址
baidu_api_url = "http://tts.baidu.com/text2audio"
baidu_api_set = {"idx": 1, "cuid": "baidu_speech_demo", "cod": 2,
? ? ? ? ? ? ? ? "lan": "zh", "ctp": 1, "pdt": 1, "spd": 4, "per": 4, "vol": 5, "pit": 5}
# 百度文字轉(zhuǎn)語音 tsn地址
baidu_api_url2 = "http://tsn.baidu.com/text2audio?tex=%s&lan=zh&cuid=%s&ctp=1&tok=%s"
BaiDu_API_Key_GetVoi = "2NagVAULCYCnOnamrc8MNUPc"
BaiDu_Secret_Key_GetVoi = "af4860b64e77d187643db05ccdb060e4"
# 百度語音識(shí)別
BaiDu_App_ID = "10623076"
BaiDu_API_Key = "2NagVAULCYCnOnamrc8MNUPc"
BaiDu_Secret_Key = "af4860b64e77d187643db05ccdb060e4"
BaiDu_OpenApi_Url = "https://openapi.baidu.com/oauth/2.0/token" \
? ? ? ? ? ? ? ? ? ? "?grant_type=client_credentials&client_id=%&client_secret=%"
config.py
三、總結(jié)
此文在本地語音保存解析過程有時(shí)間差問題，讀者可自行優(yōu)化。
