Yandex SpeechKit Cloudによるテレグラム音声処理

それがすべて始まった方法

この夏、私はロシア連邦のオープンな財務データへのアクセスを提供するDatatronボットの開発に参加しました。 ある時点で、ボットが音声要求を処理できるようにしたかったため、このタスクのためにYandex開発を使用することにしました。

このトピックに関する少なくともいくつかの有用な情報を長時間検索した後、私はついにvoiceru_botを書いた人に会い、このトピックを整理するのを手伝いました（ソースは彼のリポジトリへのリンクを提供します）。 この知識をあなたと共有したいと思います。

言葉から練習へ

以下のフラグメントには、完全に使用可能なコードが表示されます。これは、実際に簡単にコピーしてプロジェクトに貼り付けることができます。

ステップ1.どこから始めますか？

Yandexでアカウントを取得します（アカウントがない場合）。 次に、SpeechKit Cloud API の利用規約をお読みください。 要するに、1日あたりのリクエスト数が1000以下の非営利プロジェクトの場合、使用は無料です。 その後、 開発者キャビネットに移動して、必要なサービスのキーを注文します。 通常、3営業日以内にアクティベートされます（ただし、1週間後にキーの1つがアクティベートされました）。 最後に、 ドキュメントを調べます 。

ステップ2：送信した音声録音を保存する

APIにリクエストを送信する前に、音声メッセージ自体を取得する必要があります。 次のコードでは、数行で、音声メッセージに関するすべてのデータを格納するオブジェクトを取得しています。

import requests @bot.message_handler(content_types=['voice']) def voice_processing(message): file_info = bot.get_file(message.voice.file_id) file = requests.get('https://api.telegram.org/file/bot{0}/{1}'.format(TELEGRAM_API_TOKEN, file_info.file_path))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

オブジェクトをファイル変数に保存したら、主に、送信されたボイスメッセージのバイトレコードを格納するコンテンツフィールドに関心があります。 今後の作業に必要です。

ステップ3.コーディング

テレグラムの音声メッセージはOpusオーディオコーデックを使用してOGG形式で保存されます。 SpeechKitは、Speexオーディオコーデックを使用してOGGオーディオを処理できます。 したがって、ドキュメントによれば、このフォーマットは最高の認識品質を提供するため、ファイルを、何よりもPCM 16000 Hz 16ビットに変換する必要があります。 FFmpegはこれに最適です。 ダウンロードしてプロジェクトディレクトリに保存し、binフォルダーとその内容のみを残します。 以下は、FFmpegの助けを借りて、バイトストリームを目的の形式にエンコードする関数です。

 import subprocess import tempfile import os def convert_to_pcm16b16000r(in_filename=None, in_bytes=None): with tempfile.TemporaryFile() as temp_out_file: temp_in_file = None if in_bytes: temp_in_file = tempfile.NamedTemporaryFile(delete=False) temp_in_file.write(in_bytes) in_filename = temp_in_file.name temp_in_file.close() if not in_filename: raise Exception('Neither input file name nor input bytes is specified.') #        FFmpeg command = [ r'Project\ffmpeg\bin\ffmpeg.exe', #   ffmpeg.exe '-i', in_filename, '-f', 's16le', '-acodec', 'pcm_s16le', '-ar', '16000', '-' ] proc = subprocess.Popen(command, stdout=temp_out_file, stderr=subprocess.DEVNULL) proc.wait() if temp_in_file: os.remove(in_filename) temp_out_file.seek(0) return temp_out_file.read()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ステップ4.録音した音声を部分的に転送する

SpeechKit Cloud APIは最大1 MBのサイズのファイルを受け入れ、そのサイズは（Content-Lengthで）個別に指定する必要があります。 ただし、ファイル転送を部分的に実行することをお勧めします（Transfer-Encoding：chunkedヘッダーを使用して、サイズが1 MB以下）。 これはより安全であり、テキスト認識はより高速になります。

 def read_chunks(chunk_size, bytes): while True: chunk = bytes[:chunk_size] bytes = bytes[chunk_size:] yield chunk if not bytes: break
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ステップ5.要求をYandex APIに送信し、応答を解析する

最後に、最後のステップは、このモジュールの「API」として機能する単一の関数を作成することです。 つまり、最初に、ブロック内のバイトの変換と読み取りを行うメソッドを呼び出してから、SpeechKit Cloudに移動して応答を読み取ります。 デフォルトでは、リクエストの場合、トピックはメモに設定され、言語はロシア語です。

 import xml.etree.ElementTree as XmlElementTree import httplib2 import uuid from config import YANDEX_API_KEY YANDEX_ASR_HOST = 'asr.yandex.net' YANDEX_ASR_PATH = '/asr_xml' CHUNK_SIZE = 1024 ** 2 def speech_to_text(filename=None, bytes=None, request_id=uuid.uuid4().hex, topic='notes', lang='ru-RU', key=YANDEX_API_KEY): #    if filename: with open(filename, 'br') as file: bytes = file.read() if not bytes: raise Exception('Neither file name nor bytes provided.') #     bytes = convert_to_pcm16b16000r(in_bytes=bytes) #     Yandex API url = YANDEX_ASR_PATH + '?uuid=%s&key=%s&topic=%s&lang=%s' % ( request_id, key, topic, lang ) #    chunks = read_chunks(CHUNK_SIZE, bytes) #      connection = httplib2.HTTPConnectionWithTimeout(YANDEX_ASR_HOST) connection.connect() connection.putrequest('POST', url) connection.putheader('Transfer-Encoding', 'chunked') connection.putheader('Content-Type', 'audio/x-pcm;bit=16;rate=16000') connection.endheaders() #    for chunk in chunks: connection.send(('%s\r\n' % hex(len(chunk))[2:]).encode()) connection.send(chunk) connection.send('\r\n'.encode()) connection.send('0\r\n\r\n'.encode()) response = connection.getresponse() #    if response.code == 200: response_text = response.read() xml = XmlElementTree.fromstring(response_text) if int(xml.attrib['success']) == 1: max_confidence = - float("inf") text = '' for child in xml: if float(child.attrib['confidence']) > max_confidence: text = child.text max_confidence = float(child.attrib['confidence']) if max_confidence != - float("inf"): return text else: #      - -    raise SpeechException('No text found.\n\nResponse:\n%s' % (response_text)) else: raise SpeechException('No text found.\n\nResponse:\n%s' % (response_text)) else: raise SpeechException('Unknown error.\nCode: %s\n\n%s' % (response.code, response.read())) #    lass SpeechException(Exception): pass
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ステップ6.書かれたモジュールを使用する

次に、speech_to_text関数を呼び出すmainメソッドを追加します。 その中に、ユーザーが音声または認識可能なテキストがない音声メッセージを送信する場合にのみ、ケースの処理を追加する必要があります。 必要に応じてspeech_to_text関数とSpeechExceptionクラスを忘れずにインポートしてください。

 @bot.message_handler(content_types=['voice']) def voice_processing(message): file_info = bot.get_file(message.voice.file_id) file = requests.get( 'https://api.telegram.org/file/bot{0}/{1}'.format(API_TOKEN, file_info.file_path)) try: #      text = speech_to_text(bytes=file.content) except SpeechException: #  ,     else: # -
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

以上です。 これで、プロジェクトに音声処理を簡単に実装できます。 そして、この記事を基礎として、Telegramだけでなく、他のプラットフォームでも！

ソース：

「@Voiceru_bot： https : //github.com/just806me/voiceru_bot

» テレボットライブラリは、PythonでTelegram APIを操作するために使用されました