transcriber/internal/adapter/recognizer/yandex/speechkit.go

package yandex

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/proto"

	stt "github.com/yandex-cloud/go-genproto/yandex/cloud/ai/stt/v3"
	"github.com/yandex-cloud/go-genproto/yandex/cloud/operation"

	"git.vakhrushev.me/av/transcriber/internal/entity"
)

const (
	SpeechKitEndpoint = "stt.api.cloud.yandex.net:443"
	OperationEndpoint = "operation.api.cloud.yandex.net:443"

	RecognitionModel = "deferred-general"
)

type speechKitConfig struct {
	ApiKey   string
	FolderID string
}

type speechKitService struct {
	sttConn   *grpc.ClientConn
	opConn    *grpc.ClientConn
	sttClient stt.AsyncRecognizerClient
	opClient  operation.OperationServiceClient
	apiKey    string
	folderID  string
}

func newSpeechKitService(cfg speechKitConfig) (*speechKitService, error) {
	apiKey := cfg.ApiKey
	folderID := cfg.FolderID

	if apiKey == "" || folderID == "" {
		return nil, fmt.Errorf("missing required Yandex Cloud environment variables")
	}

	// Создаем защищенное соединение для SpeechKit
	creds := credentials.NewTLS(nil)
	sttConn, err := grpc.NewClient(SpeechKitEndpoint, grpc.WithTransportCredentials(creds))
	if err != nil {
		return nil, fmt.Errorf("failed to connect to SpeechKit: %w", err)
	}

	// Создаем защищенное соединение для Operations API
	opConn, err := grpc.NewClient(OperationEndpoint, grpc.WithTransportCredentials(creds))
	if err != nil {
		// Отказы независимы, и второй не теряется. На сегодняшнем клиенте он
		// почти наверняка не наступит: grpc.NewClient ленив, соединение к этому
		// моменту не открыто, и Close вернёт отказ только при повторном
		// закрытии — то есть сообщит о нашей ошибке, а не о Yandex. Сборка
		// оставлена как защита от смены реализации клиента; nil от закрытия
		// errors.Join отбрасывает, и форма ошибки в обычном случае не меняется.
		return nil, errors.Join(
			fmt.Errorf("failed to connect to Operations API: %w", err),
			sttConn.Close(),
		)
	}

	sttClient := stt.NewAsyncRecognizerClient(sttConn)
	opClient := operation.NewOperationServiceClient(opConn)

	return &speechKitService{
		sttConn:   sttConn,
		opConn:    opConn,
		sttClient: sttClient,
		opClient:  opClient,
		apiKey:    apiKey,
		folderID:  folderID,
	}, nil
}

func (s *speechKitService) Close() error {
	var err1, err2 error
	if s.sttConn != nil {
		err1 = s.sttConn.Close()
	}
	if s.opConn != nil {
		err2 = s.opConn.Close()
	}
	// Отказы двух соединений независимы, и вернуть только первый — значит
	// потерять половину причины: журнал пишется при остановке процесса, и
	// восстановить утраченное будет уже негде.
	return errors.Join(err1, err2)
}

// recognizeFileFromS3 запускает асинхронное распознавание файла из S3
func (s *speechKitService) recognizeFileFromS3(ctx context.Context, s3URI string) (string, error) {
	// Добавляем авторизацию и folder_id в контекст
	ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Api-Key "+s.apiKey)
	ctx = metadata.AppendToOutgoingContext(ctx, "x-folder-id", s.folderID)

	// Создаем запрос на распознавание
	req := &stt.RecognizeFileRequest{
		AudioSource: &stt.RecognizeFileRequest_Uri{
			Uri: s3URI,
		},
		RecognitionModel: &stt.RecognitionModelOptions{
			Model: RecognitionModel,
			AudioFormat: &stt.AudioFormatOptions{
				AudioFormat: &stt.AudioFormatOptions_ContainerAudio{
					ContainerAudio: &stt.ContainerAudio{
						ContainerAudioType: stt.ContainerAudio_OGG_OPUS,
					},
				},
			},
			TextNormalization: &stt.TextNormalizationOptions{
				TextNormalization: stt.TextNormalizationOptions_TEXT_NORMALIZATION_ENABLED,
				ProfanityFilter:   false,
				LiteratureText:    true,
			},
			AudioProcessingType: stt.RecognitionModelOptions_FULL_DATA,
		},
		SpeakerLabeling: &stt.SpeakerLabelingOptions{
			SpeakerLabeling: stt.SpeakerLabelingOptions_SPEAKER_LABELING_ENABLED,
		},
	}

	// Отправляем запрос
	op, err := s.sttClient.RecognizeFile(ctx, req)
	if err != nil {
		return "", fmt.Errorf("failed to start recognition: %w", err)
	}

	return op.Id, nil
}

// fetchRecognition забирает результат операции целиком и отдаёт его доменным,
// вместе с сырым ответом на хранение.
//
// Ответ сохраняется потому, что **результат операции у провайдера не
// переспрашивается**: связь реплики с говорящим сервис строить пока не умеет, и
// когда научится, архив пересчитается из сохранённого без единого рубля.
func (s *speechKitService) fetchRecognition(ctx context.Context, operationID string) (*entity.RecognitionOutcome, error) {
	ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Api-Key "+s.apiKey)
	ctx = metadata.AppendToOutgoingContext(ctx, "x-folder-id", s.folderID)

	req := &stt.GetRecognitionRequest{
		OperationId: operationID,
	}

	stream, err := s.sttClient.GetRecognition(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("failed to get recognition stream: %w", err)
	}

	var responses []*stt.StreamingResponse
	for {
		resp, err := stream.Recv()
		if err != nil {
			// Конец потока библиотека отдаёт ровно `io.EOF`. Прежде он узнавался
			// сравнением текста сообщения: так же выглядел бы и настоящий отказ
			// с текстом «EOF», и распознавание молча вернуло бы половину текста.
			if errors.Is(err, io.EOF) {
				break
			}
			return nil, fmt.Errorf("failed to receive recognition response: %w", err)
		}
		responses = append(responses, resp)
	}

	raw, err := encodeResponses(responses)
	if err != nil {
		return nil, err
	}

	outcome := outcomeFromResponses(responses)
	outcome.Raw = raw
	return outcome, nil
}

// checkOperationStatus проверяет статус операции распознавания
func (s *speechKitService) checkOperationStatus(ctx context.Context, operationID string) (*operation.Operation, error) {
	ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Api-Key "+s.apiKey)
	ctx = metadata.AppendToOutgoingContext(ctx, "x-folder-id", s.folderID)

	op, err := s.opClient.Get(ctx, &operation.GetOperationRequest{
		OperationId: operationID,
	})

	if err != nil {
		return nil, fmt.Errorf("failed to get operation status: %w", err)
	}

	return op, nil
}

// encodeResponses укладывает ответ провайдера целиком, в том виде, в каком он
// пришёл: сообщения потока подряд, каждое со своей длиной впереди.
//
// Форма **двоичная**, а не текстовая, и это несущее решение. Текстовое
// представление собирается по нашей скомпилированной схеме и молча выбрасывает
// поля, которых в ней нет, — а провайдер добавляет их без предупреждения.
// Двоичная форма неизвестные поля переносит: они переживают запись и чтение и
// станут читаемыми, когда мы обновим схему. Ради этого архив и заводился —
// результат операции у провайдера не переспрашивается, и повторное
// распознавание стоит денег.
//
// Цена названа прямо: сохранённое не читается глазами и не разбирается ничем,
// кроме нашего же кода.
func encodeResponses(responses []*stt.StreamingResponse) ([]byte, error) {
	var raw []byte
	for _, resp := range responses {
		encoded, err := proto.Marshal(resp)
		if err != nil {
			// Текст расшифровки наружу не выходит даже отказом: сообщение
			// провайдера несёт её целиком.
			return nil, errors.New("failed to encode provider response")
		}
		raw = binary.AppendUvarint(raw, uint64(len(encoded)))
		raw = append(raw, encoded...)
	}
	return raw, nil
}

// decodeResponses читает сохранённый ответ провайдера обратно.
func decodeResponses(raw []byte) ([]*stt.StreamingResponse, error) {
	var responses []*stt.StreamingResponse

	for len(raw) > 0 {
		size, read := binary.Uvarint(raw)
		if read <= 0 || uint64(len(raw)-read) < size {
			return nil, errors.New("stored provider payload is truncated")
		}
		raw = raw[read:]

		var resp stt.StreamingResponse
		if err := proto.Unmarshal(raw[:size], &resp); err != nil {
			return nil, errors.New("failed to decode provider response")
		}
		responses = append(responses, &resp)
		raw = raw[size:]
	}

	return responses, nil
}

// outcomeFromResponses строит доменный результат: реплики со временем и плоский
// текст. Формата провайдера отсюда наружу не выходит ничего.
//
// Говорящие не размечаются: связь реплики с разбором говорящего у провайдера не
// выяснена. Структура при этом строится из сохранённого ответа, поэтому разметка
// станет возможной без повторной оплаты.
func outcomeFromResponses(responses []*stt.StreamingResponse) *entity.RecognitionOutcome {
	outcome := &entity.RecognitionOutcome{}

	var plain []byte
	for _, resp := range responses {
		refinement := resp.GetFinalRefinement()
		if refinement == nil {
			continue
		}
		text := refinement.GetNormalizedText()
		if text == nil {
			continue
		}
		for _, alt := range text.GetAlternatives() {
			if alt.GetText() == "" {
				continue
			}
			outcome.Replicas = append(outcome.Replicas, entity.Replica{
				StartMs: alt.GetStartTimeMs(),
				EndMs:   alt.GetEndTimeMs(),
				Text:    alt.GetText(),
			})
			if len(plain) > 0 {
				plain = append(plain, ' ')
			}
			plain = append(plain, alt.GetText()...)
		}
	}

	outcome.PlainText = string(plain)
	return outcome
}