Extract logic into transcribe service

internal/service/s3/s3.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
@@ -78,3 +79,9 @@ func (s *S3Service) UploadFile(filePath, fileName string) error {
 
 	return nil
 }
+
+func (s *S3Service) FileUrl(fileName string) string {
+	endpoint := strings.TrimRight(os.Getenv("S3_ENDPOINT"), "/")
+	bucketName := os.Getenv("S3_BUCKET_NAME")
+	return fmt.Sprintf("%s/%s/%s", endpoint, bucketName, fileName)
+}

internal/service/speechkit/speechkit.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"strings"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -17,6 +18,8 @@ import (
 const (
 	SpeechKitEndpoint = "stt.api.cloud.yandex.net:443"
 	OperationEndpoint = "operation.api.cloud.yandex.net:443"
+
+	RecognitionModel = "deferred-general"
 )
 
 type SpeechKitService struct {
@@ -91,7 +94,7 @@ func (s *SpeechKitService) RecognizeFileFromS3(s3URI string) (string, error) {
 			Uri: s3URI,
 		},
 		RecognitionModel: &stt.RecognitionModelOptions{
-			Model: "general", // Используем общую модель
+			Model: RecognitionModel,
 			AudioFormat: &stt.AudioFormatOptions{
 				AudioFormat: &stt.AudioFormatOptions_ContainerAudio{
 					ContainerAudio: &stt.ContainerAudio{
@@ -121,7 +124,7 @@ func (s *SpeechKitService) RecognizeFileFromS3(s3URI string) (string, error) {
 }
 
 // GetRecognitionResult получает результат распознавания по ID операции
-func (s *SpeechKitService) GetRecognitionResult(operationID string) ([]*stt.StreamingResponse, error) {
+func (s *SpeechKitService) GetRecognitionText(operationID string) (string, error) {
 	ctx := context.Background()
 
 	// Добавляем авторизацию и folder_id в контекст
@@ -134,22 +137,28 @@ func (s *SpeechKitService) GetRecognitionResult(operationID string) ([]*stt.Stre
 
 	stream, err := s.sttClient.GetRecognition(ctx, req)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get recognition stream: %w", err)
+		return "", fmt.Errorf("failed to get recognition stream: %w", err)
 	}
 
-	var responses []*stt.StreamingResponse
+	var sb strings.Builder
+
 	for {
 		resp, err := stream.Recv()
 		if err != nil {
 			if err.Error() == "EOF" {
 				break
 			}
-			return nil, fmt.Errorf("failed to receive recognition response: %w", err)
+			return "", fmt.Errorf("failed to receive recognition response: %w", err)
+		}
+		if final := resp.GetFinal(); final != nil {
+			for _, alt := range final.Alternatives {
+				sb.WriteString(alt.Text)
+				sb.WriteString(" ")
+			}
 		}
-		responses = append(responses, resp)
 	}
 
-	return responses, nil
+	return sb.String(), nil
 }
 
 // CheckOperationStatus проверяет статус операции распознавания
@@ -171,18 +180,3 @@ func (s *SpeechKitService) CheckOperationStatus(operationID string) (*operation.
 
 	return op, nil
 }
-
-// ExtractTranscriptionText извлекает текст из результатов распознавания
-func ExtractTranscriptionText(responses []*stt.StreamingResponse) string {
-	var fullText string
-
-	for _, resp := range responses {
-		if final := resp.GetFinal(); final != nil {
-			for _, alt := range final.Alternatives {
-				fullText += alt.Text + " "
-			}
-		}
-	}
-
-	return fullText
-}

internal/service/transcribe/transcribe.go

@@ -0,0 +1,277 @@
+package transcribe
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"git.vakhrushev.me/av/transcriber/internal/entity"
+	"git.vakhrushev.me/av/transcriber/internal/repo"
+	"git.vakhrushev.me/av/transcriber/internal/repo/ffmpeg"
+	"git.vakhrushev.me/av/transcriber/internal/service/s3"
+	"git.vakhrushev.me/av/transcriber/internal/service/speechkit"
+	"github.com/google/uuid"
+)
+
+const baseStorageDir = "data/files"
+
+type TranscribeService struct {
+	jobRepo  repo.TranscriptJobRepository
+	fileRepo repo.FileRepository
+}
+
+func NewTranscribeService(jobRepo repo.TranscriptJobRepository, fileRepo repo.FileRepository) *TranscribeService {
+	return &TranscribeService{jobRepo: jobRepo, fileRepo: fileRepo}
+}
+
+func (s *TranscribeService) CreateTranscribeJob(file io.Reader, fileName string) (*entity.TranscribeJob, error) {
+	// Генерируем UUID для файла
+	fileId := uuid.New().String()
+
+	// Определяем расширение файла
+	ext := filepath.Ext(fileName)
+	if ext == "" {
+		ext = ".audio" // fallback если расширение не определено
+	}
+
+	// Создаем путь для сохранения файла
+	storageFileName := fmt.Sprintf("%s%s", fileId, ext)
+	storageFilePath := filepath.Join(baseStorageDir, storageFileName)
+
+	// Создаем файл на диске
+	dst, err := os.Create(storageFilePath)
+	if err != nil {
+		return nil, err
+	}
+	defer dst.Close()
+
+	// Копируем содержимое загруженного файла
+	size, err := io.Copy(dst, file)
+	if err != nil {
+		return nil, err
+	}
+
+	// Создаем запись в таблице files
+	fileRecord := &entity.File{
+		Id:        fileId,
+		Storage:   entity.StorageLocal,
+		FileName:  storageFileName,
+		Size:      size,
+		CreatedAt: time.Now(),
+	}
+
+	if err := s.fileRepo.Create(fileRecord); err != nil {
+		// Удаляем файл если не удалось создать запись в БД
+		os.Remove(storageFilePath)
+		return nil, err
+	}
+
+	jobId := uuid.NewString()
+	now := time.Now()
+
+	// Создаем запись в таблице transcribe_jobs
+	job := &entity.TranscribeJob{
+		Id:        jobId,
+		State:     entity.StateCreated,
+		FileID:    &fileId,
+		IsError:   false,
+		CreatedAt: now,
+		UpdatedAt: now,
+	}
+
+	if err := s.jobRepo.Create(job); err != nil {
+		return nil, err
+	}
+
+	return job, nil
+}
+
+func (s *TranscribeService) FindAndRunConversionJob() error {
+	acquisitionId := uuid.NewString()
+	rottingTime := time.Now().Add(-1 * time.Hour)
+
+	job, err := s.jobRepo.FindAndAcquire(entity.StateCreated, acquisitionId, rottingTime)
+	if err != nil {
+		return err
+	}
+
+	srcFile, err := s.fileRepo.GetByID(*job.FileID)
+	if err != nil {
+		return err
+	}
+
+	srcFilePath := filepath.Join(baseStorageDir, srcFile.FileName)
+
+	destFileId := uuid.NewString()
+	destFileName := fmt.Sprintf("%s%s", destFileId, ".ogg")
+	destFilePath := filepath.Join(baseStorageDir, destFileName)
+
+	conv := ffmpeg.NewFileConverter()
+	err = conv.Convert(srcFilePath, destFilePath)
+	if err != nil {
+		return err
+	}
+
+	stat, err := os.Stat(destFilePath)
+	if err != nil {
+		return err
+	}
+
+	// Создаем запись в таблице files
+	destFileRecord := &entity.File{
+		Id:        destFileId,
+		Storage:   entity.StorageLocal,
+		FileName:  destFileName,
+		Size:      stat.Size(),
+		CreatedAt: time.Now(),
+	}
+
+	job.FileID = &destFileId
+	job.MoveToState(entity.StateConverted)
+
+	err = s.fileRepo.Create(destFileRecord)
+	if err != nil {
+		return err
+	}
+
+	err = s.jobRepo.Save(job)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *TranscribeService) FindAndRunTranscribeJob() error {
+	acquisitionId := uuid.NewString()
+	rottingTime := time.Now().Add(-1 * time.Hour)
+
+	jobRecord, err := s.jobRepo.FindAndAcquire(entity.StateConverted, acquisitionId, rottingTime)
+	if err != nil {
+		return err
+	}
+
+	fileRecord, err := s.fileRepo.GetByID(*jobRecord.FileID)
+	if err != nil {
+		return err
+	}
+
+	filePath := filepath.Join(baseStorageDir, fileRecord.FileName)
+
+	destFileId := uuid.NewString()
+	destFileRecord := fileRecord.CopyWithStorage(destFileId, entity.StorageS3)
+
+	// Создаем S3 сервис
+	s3Service, err := s3.NewS3Service()
+	if err != nil {
+		return err
+	}
+
+	// Загружаем файл на S3
+	err = s3Service.UploadFile(filePath, destFileRecord.FileName)
+	if err != nil {
+		return err
+	}
+
+	// Создаем SpeechKit сервис
+	speechKitService, err := speechkit.NewSpeechKitService()
+	if err != nil {
+		return err
+	}
+
+	// Формируем S3 URI для файла
+	s3URI := s3Service.FileUrl(destFileRecord.FileName)
+
+	// Запускаем асинхронное распознавание
+	operationID, err := speechKitService.RecognizeFileFromS3(s3URI)
+	if err != nil {
+		return err
+	}
+
+	// Обновляем задачу с ID операции распознавания
+	jobRecord.FileID = &destFileId
+	jobRecord.RecognitionOpID = &operationID
+	delayTime := time.Now().Add(time.Minute)
+	jobRecord.MoveToStateAndDelay(entity.StateTranscribe, &delayTime)
+
+	err = s.fileRepo.Create(destFileRecord)
+	if err != nil {
+		return err
+	}
+
+	err = s.jobRepo.Save(jobRecord)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *TranscribeService) FindAndRunTranscribeCheckJob() error {
+	acquisitionId := uuid.NewString()
+	rottingTime := time.Now().Add(-24 * time.Hour)
+
+	job, err := s.jobRepo.FindAndAcquire(entity.StateTranscribe, acquisitionId, rottingTime)
+	if err != nil {
+		return err
+	}
+
+	if job.RecognitionOpID == nil {
+		return fmt.Errorf("recogniton opId not found for job: %s", job.Id)
+	}
+
+	// Создаем SpeechKit сервис
+	speechKitService, err := speechkit.NewSpeechKitService()
+	if err != nil {
+		return err
+	}
+	defer speechKitService.Close()
+
+	// Проверяем статус операции
+	operation, err := speechKitService.CheckOperationStatus(*job.RecognitionOpID)
+	if err != nil {
+		return err
+	}
+
+	if !operation.Done {
+		// Операция еще не завершена, оставляем в статусе обработки
+		delayTime := time.Now().Add(10 * time.Second)
+		job.MoveToStateAndDelay(entity.StateTranscribe, &delayTime)
+		err := s.jobRepo.Save(job)
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	if opErr := operation.GetError(); opErr != nil {
+		job.IsError = true
+		errorText := fmt.Sprintf("Operation failed: code %d, message: %s", opErr.Code, opErr.Message)
+		job.ErrorText = &errorText
+		job.MoveToState(entity.StateFailed)
+		err := s.jobRepo.Save(job)
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	// Операция завершена, получаем результат
+	transcriptionText, err := speechKitService.GetRecognitionText(*job.RecognitionOpID)
+	if err != nil {
+		return err
+	}
+
+	// Обновляем задачу с результатом
+	job.TranscriptionText = &transcriptionText
+	job.MoveToState(entity.StateDone)
+
+	err = s.jobRepo.Save(job)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}