用Python为你念一句诗

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

知识回顾

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

c/s架构

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

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服务器

#!/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客户端

#!/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字段的诗词信息，如：

{"code":200,"message":"成功!","result":{"author":"白居易","origin":"空闺怨","category":"古诗文-抒情-闺怨","content":"秋霜欲下手先知，灯底裁缝剪刀冷。"}}

#!/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/）。

#!/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文件。

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])