Telegramm-Sprachverarbeitung mit Yandex SpeechKit Cloud

Wie alles begann


In diesem Sommer war ich an der Entwicklung des Datatron- Bots beteiligt , der den Zugriff auf offene Finanzdaten der Russischen Föderation ermöglicht. Irgendwann wollte ich, dass der Bot Sprachanfragen verarbeiten kann, und entschied mich für Yandex-Entwicklungen.


Nach einer langen Suche nach mindestens einigen nützlichen Informationen zu diesem Thema traf ich endlich eine Person, die voiceru_bot schrieb und mir half, dieses Thema zu klären (die Quellen enthalten einen Link zu seinem Repository). Jetzt möchte ich dieses Wissen mit Ihnen teilen.


Vom Wort zur Praxis


Die Fragmente unten zeigen den Code, der vollständig einsatzbereit ist und praktisch einfach kopiert und in Ihr Projekt eingefügt werden kann.


Schritt 1. Wo soll ich anfangen?


Eröffnen Sie ein Konto bei Yandex (falls Sie noch keines haben). Dann lesen Sie die Nutzungsbedingungen SpeechKit Cloud API. Kurz gesagt, für nichtkommerzielle Projekte, bei denen die Anzahl der Anfragen nicht mehr als 1000 pro Tag beträgt, ist die Nutzung kostenlos. Gehen Sie danach zum Developer Cabinet und bestellen Sie einen Schlüssel für den gewünschten Service. Die Aktivierung erfolgt normalerweise innerhalb von 3 Werktagen (obwohl einer meiner Schlüssel nach einer Woche aktiviert wurde). Lesen Sie abschließend die Dokumentation .


Schritt 2: Speichern Sie die gesendete Sprachaufnahme


Bevor Sie eine Anforderung an die API senden, müssen Sie die Sprachnachricht selbst abrufen. Im folgenden Code erhalten wir in wenigen Zeilen ein Objekt, das alle Daten zur Sprachnachricht speichert.


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

Nachdem wir das Objekt in der Dateivariablen gespeichert haben, interessieren wir uns hauptsächlich für das Inhaltsfeld, in dem der Bytedatensatz der gesendeten Sprachnachricht gespeichert wird. Wir brauchen es für die weitere Arbeit.


Schritt 3. Rekodierung


Die Sprachnachricht des Telegramms wird im OGG-Format mit dem Opus-Audiocodec gespeichert. SpeechKit kann OGG-Audio mit dem Speex-Audiocodec verarbeiten. Daher ist es notwendig, die Datei, am besten auf PCM 16000 Hz 16 Bit, zu konvertieren, da dieses Format laut Dokumentation die beste Erkennungsqualität bietet. FFmpeg ist großartig dafür . Laden Sie es herunter und speichern Sie es im Projektverzeichnis, wobei nur der Ordner bin und sein Inhalt übrig bleiben. Unten ist eine Funktion, die mit Hilfe von FFmpeg einen Strom von Bytes in das gewünschte Format codiert.


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

Schritt 4. Übertragen einer Audioaufnahme in Teilen


Die SpeechKit Cloud-API akzeptiert eine Datei mit einer Größe von bis zu 1 MB. Die Größe muss separat angegeben werden (in Inhaltslänge). Es ist jedoch besser, die Dateiübertragung in Teilen zu implementieren (nicht größer als 1 MB bei Verwendung der Option "Transfer-Encoding: Chunked Header"). Dies ist sicherer und die Texterkennung wird schneller.


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

Schritt 5. Senden einer Anfrage an die Yandex-API und Analysieren einer Antwort


Schließlich besteht der letzte Schritt darin, eine einzelne Funktion zu schreiben, die als "API" für dieses Modul dient. Dies bedeutet, dass zuerst die Methoden aufgerufen werden, die für das Konvertieren und Lesen von Bytes in Blöcken verantwortlich sind, und dann die SpeechKit Cloud aufgerufen und die Antwort gelesen werden. Standardmäßig ist für Anfragen das Thema auf Notizen festgelegt, und die Sprache ist Russisch.


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

Schritt 6. Verwenden des geschriebenen Moduls


Nun fügen wir die Hauptmethode hinzu, von der aus wir die Funktion speech_to_text aufrufen. Darin müssen Sie nur die Verarbeitung des Falls hinzufügen, wenn der Benutzer eine Sprachnachricht sendet, in der keine Töne oder erkennbarer Text vorhanden sind. Denken Sie daran, die Funktion speech_to_text und gegebenenfalls die SpeechException-Klasse zu importieren.


@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:
        # Бизнес-логика

Das ist alles. Jetzt können Sie die Sprachverarbeitung einfach in Ihre Projekte implementieren. Und das nicht nur in Telegram, sondern auch auf anderen Plattformen, basierend auf diesem Artikel!


Quellen:


"@Voiceru_bot: https://github.com/just806me/voiceru_bot
" Die Telebot- Bibliothek wurde verwendet, um mit der Telegramm-API in Python zu arbeiten


Jetzt auch beliebt: