用Python为你念一句诗

看完网络编程的知识,觉得客户端和服务器之间消息传递太无聊了,于是调用了一个外部API和语音合成服务,只要在客户端输入任意内容,服务器返回一句诗并读给你听。

知识回顾

回顾《Python核心编程》第二章:网络编程的基础知识。

c/s架构

客户端/服务器架构,也称为c/s架构,即客户端发出请求,服务器接受请求并提供服务。举个例子,银行的柜员可以看作是服务器,而排队的客户则可以视为一个一个的客户端,柜员不断接受并处理来自客户的请求,然后将结果回复给客户。

image

c/s架构通常分为软件和硬件两种:

  • 硬件:打印机、文件服务器
  • 软件:web服务器

套接字

  • 一种计算机网络数据结构,开始任何通信之前必须创建套接字
    • 套接字家族(address family):AF_UNIX, AF_NETLINK, AF_TIPC, AF_INET(最常用)
  • 主机-端口对:“区号-电话号码”,有效端口范围为0~65535
  • 套接字类型:
    • 面向连接:TCP(传输控制协议),SOCK_STREAM
    • 无连接套接字:UDP(用户数据报协议),SOCK_DGRAM

实现步骤

实现的步骤较为简单,代码见github:https://github.com/swordspoet/ReadPoem

创建TCP服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
#!/usr/bin/env python
# coding: utf-8

from socket import *
from poetry_request import poetry_query


host = 'localhost'
port = 21765
buff_size = 1024
address = (host, port) # 主机-地址对

server_sock = socket(AF_INET, SOCK_STREAM) # 创建服务器套接字
server_sock.bind(address) # 套接字与主机-地址对绑定
server_sock.listen(5) # 监听连接


while 1: # 服务器无线循环
cli_sock, addr = server_sock.accept() # 接受客户端连接,开启单线程服务器

while 1: # 通信无限循环
query = cli_sock.recv(buff_size).decode() # 接受客户端消息
if not query:
break
query = poetry_query()
print(query.get('result').get('content')) # 从API返回结果中提取诗句内容并发送至客户端
content = query.get('result').get('content')
cli_sock.send(bytes(content.encode('utf8')))

# auth.save_voice(content.encode('utf8'), 'test')
# player.play_voice_by('test')

cli_sock.close()
server_sock.close()

创建TCP客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
#!/usr/bin/bin python
# coding: utf-8

from socket import *
from voice import Voice
from player import Player


host = 'localhost'
port = 21765
buff_size = 1024
address = (host, port)

cli_sock = socket(AF_INET, SOCK_STREAM)
cli_sock.connect(address)
# 阿里云语音合成配置(秘钥可以找我索取)
secret_id = ''
secret_key = ''
auth = Voice(secret_id, secret_key)
player = Player()


while 1:
query = input("念诗一首: ")
if not query:
break
cli_sock.send(query.encode())
data = cli_sock.recv(buff_size).decode() # 客户端接收来自服务器的消息
print(data)

auth.save_voice(data.encode('utf8'), 'test') # 将服务器返回的诗句合成为MP3文件
player.play_voice_by('test') # 调用系统播放器播放MP3文件
if not data:
break

cli_sock.close()

调用诗词API接口

接口的地址是:https://api.apiopen.top/singlePoetry,直接访问便可返回json字段的诗词信息,如:

1
{"code":200,"message":"成功!","result":{"author":"白居易","origin":"空闺怨","category":"古诗文-抒情-闺怨","content":"秋霜欲下手先知,灯底裁缝剪刀冷。"}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#!/usr/bin/env python
# coding: utf-8

import certifi
import urllib3
import json

http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED',
ca_certs=certifi.where())


def poetry_query():
request_url = 'https://api.apiopen.top/singlePoetry'
response_text = http.request('GET', request_url)
response = json.loads(response_text.data.decode('utf-8'))
return response

诗词转换为MP3文件

文本转为语音,即TTS,bat公司都有开放的语音合成服务,因为之前用过阿里的,所以直接调用(引用了https://pypi.org/project/aliyun-voice/1.0.2/)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
#!/usr/bin/env python
# coding: utf-8

import requests
import time
import urllib
import hmac, hashlib, base64


class Voice(object):
def __init__(self, access_id, access_key):
'''
用阿里云access_id,access_key初始化
'''
self.access_id = access_id
self.access_key = access_key
self.tts_params = {
"encode_type": "mp3",
"voice_name": "xiaoyun",
"volume": 40,
"sample_rate": 16000,
"speech_rate": 0,
"pitch_rate": 0,
"tss_nus": 1,
"background_music_id": -1,
"background_music_offset": 0,
"background_music_volume": 50,
}
self.__API = "https://nlsapi.aliyun.com/speak?%s"

def __get_tts_auth(self, text, date):
md5 = hashlib.md5()
md5.update(text)
body_md5 = base64.b64encode(md5.digest())
feature = "%s\n%s\n%s\n%s\n%s" % (
"POST", "audio/%s,application/json" % self.tts_params["encode_type"], body_md5.decode('utf-8'), "text/plain", date)
return base64.b64encode(hmac.new(bytes(self.access_key.encode('utf-8')), bytes(feature.encode('utf-8')), hashlib.sha1).digest()).decode('utf-8')

def __get_tts_params(self):
return urllib.parse.urlencode(self.tts_params)

def get_voice(self, text, **kw):
'''
获取文本的声音文件
@params<string> 文本
@params<dict> 设置参数,参照aliyun tts的配置。 【注意,文件类型只需传入 [mp3, wav, ...]即可】
@return<raw> 返回声音的二进制
@throw error 当参数不符合阿里云要求时会抛出错误
'''
for key, value in kw.items():
self.tts_params[key] = value
date = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
headers = {
"Authorization": "Dataplus %s:%s" % (self.access_id, self.__get_tts_auth(text, date)),
"Content-Type": "text/plain",
"accept": "audio/%s,application/json" % self.tts_params["encode_type"],
"date": date
}
urlstr = self.__API % self.__get_tts_params()
r = requests.post(urlstr, headers=headers, data=text, stream=True)
content_type = r.headers['Content-Type']
if content_type.find('json') != -1:
raise StandardError(r.json())
else:
return r.content
# return r.content

def save_voice(self, text, dist, **kw):
'''
存储文本的声音文件
@params<string> 文本
@params<string> 声音文件存储路径
@params<dict> 设置参数,参照aliyun tts的配置。 【注意,文件类型只需传入 [mp3, wav, ...]即可】
@throw error 当参数不符合阿里云要求时会抛出错误
'''
content = self.get_voice(text, **kw)
with open(dist, 'wb') as fd:
fd.write(content)

调用播放器播放MP3文件

经过文字转语音,还需要调用播放器来播放MP3文件。

1
2
3
4
5
6
7
8
9
import subprocess


class Player(object):
def play_voice_by(self, voice_file_path):
'''- 调用 mac 系统播放器 afplay 播放 MP3 文件
- linux 下安装 sudo apt install mplayer,调用方法为:subprocess(['mplayer', voice_file_path])
'''
subprocess.call(['mplayer', voice_file_path])
觉得还不错?赞助一下~
0%