Dependency inversion for s3 service
This commit is contained in:
@@ -1,87 +0,0 @@
|
||||
package s3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
)
|
||||
|
||||
type S3Service struct {
|
||||
client *s3.Client
|
||||
uploader *manager.Uploader
|
||||
bucketName string
|
||||
}
|
||||
|
||||
func NewS3Service() (*S3Service, error) {
|
||||
region := os.Getenv("AWS_REGION")
|
||||
accessKey := os.Getenv("AWS_ACCESS_KEY_ID")
|
||||
secretKey := os.Getenv("AWS_SECRET_ACCESS_KEY")
|
||||
bucketName := os.Getenv("S3_BUCKET_NAME")
|
||||
endpoint := os.Getenv("S3_ENDPOINT")
|
||||
|
||||
if region == "" || accessKey == "" || secretKey == "" || bucketName == "" {
|
||||
return nil, fmt.Errorf("missing required S3 environment variables")
|
||||
}
|
||||
|
||||
// Создаем конфигурацию
|
||||
cfg, err := config.LoadDefaultConfig(context.TODO(),
|
||||
config.WithRegion(region),
|
||||
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to load AWS config: %w", err)
|
||||
}
|
||||
|
||||
// Создаем клиент S3
|
||||
var client *s3.Client
|
||||
if endpoint != "" {
|
||||
// Кастомный endpoint (например, для MinIO)
|
||||
client = s3.NewFromConfig(cfg, func(o *s3.Options) {
|
||||
o.BaseEndpoint = aws.String(endpoint)
|
||||
o.UsePathStyle = true
|
||||
})
|
||||
} else {
|
||||
// Стандартный AWS S3
|
||||
client = s3.NewFromConfig(cfg)
|
||||
}
|
||||
|
||||
uploader := manager.NewUploader(client)
|
||||
|
||||
return &S3Service{
|
||||
client: client,
|
||||
uploader: uploader,
|
||||
bucketName: bucketName,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *S3Service) UploadFile(filePath, fileName string) error {
|
||||
file, err := os.Open(filePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open file %s: %w", filePath, err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
_, err = s.uploader.Upload(context.TODO(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(s.bucketName),
|
||||
Key: aws.String(fileName),
|
||||
Body: file,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to upload file to S3: %w", err)
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
@@ -1,181 +0,0 @@
|
||||
package speechkit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/metadata"
|
||||
|
||||
stt "github.com/yandex-cloud/go-genproto/yandex/cloud/ai/stt/v3"
|
||||
"github.com/yandex-cloud/go-genproto/yandex/cloud/operation"
|
||||
)
|
||||
|
||||
const (
|
||||
SpeechKitEndpoint = "stt.api.cloud.yandex.net:443"
|
||||
OperationEndpoint = "operation.api.cloud.yandex.net:443"
|
||||
|
||||
RecognitionModel = "deferred-general"
|
||||
)
|
||||
|
||||
type SpeechKitService struct {
|
||||
sttConn *grpc.ClientConn
|
||||
opConn *grpc.ClientConn
|
||||
sttClient stt.AsyncRecognizerClient
|
||||
opClient operation.OperationServiceClient
|
||||
apiKey string
|
||||
folderID string
|
||||
}
|
||||
|
||||
func NewSpeechKitService() (*SpeechKitService, error) {
|
||||
apiKey := os.Getenv("YANDEX_CLOUD_API_KEY")
|
||||
folderID := os.Getenv("YANDEX_CLOUD_FOLDER_ID")
|
||||
|
||||
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 {
|
||||
sttConn.Close()
|
||||
return nil, fmt.Errorf("failed to connect to Operations API: %w", err)
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
if err1 != nil {
|
||||
return err1
|
||||
}
|
||||
return err2
|
||||
}
|
||||
|
||||
// RecognizeFileFromS3 запускает асинхронное распознавание файла из S3
|
||||
func (s *SpeechKitService) RecognizeFileFromS3(s3URI string) (string, error) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Добавляем авторизацию и 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
|
||||
}
|
||||
|
||||
// GetRecognitionResult получает результат распознавания по ID операции
|
||||
func (s *SpeechKitService) GetRecognitionText(operationID string) (string, error) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Добавляем авторизацию и folder_id в контекст
|
||||
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 "", fmt.Errorf("failed to get recognition stream: %w", err)
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
for {
|
||||
resp, err := stream.Recv()
|
||||
if err != nil {
|
||||
if err.Error() == "EOF" {
|
||||
break
|
||||
}
|
||||
return "", fmt.Errorf("failed to receive recognition response: %w", err)
|
||||
}
|
||||
if refinement := resp.GetFinalRefinement(); refinement != nil {
|
||||
if text := refinement.GetNormalizedText(); text != nil {
|
||||
for _, alt := range text.Alternatives {
|
||||
sb.WriteString(alt.Text)
|
||||
sb.WriteString(" ")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return sb.String(), nil
|
||||
}
|
||||
|
||||
// CheckOperationStatus проверяет статус операции распознавания
|
||||
func (s *SpeechKitService) CheckOperationStatus(operationID string) (*operation.Operation, error) {
|
||||
ctx := context.Background()
|
||||
|
||||
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
|
||||
}
|
||||
@@ -8,10 +8,9 @@ import (
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"git.vakhrushev.me/av/transcriber/internal/adapter/speechkit"
|
||||
"git.vakhrushev.me/av/transcriber/internal/contract"
|
||||
"git.vakhrushev.me/av/transcriber/internal/entity"
|
||||
"git.vakhrushev.me/av/transcriber/internal/service/s3"
|
||||
"git.vakhrushev.me/av/transcriber/internal/service/speechkit"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
@@ -21,13 +20,20 @@ type TranscribeService struct {
|
||||
jobRepo contract.TranscriptJobRepository
|
||||
fileRepo contract.FileRepository
|
||||
converter contract.AudioFileConverter
|
||||
s3Service contract.YandexS3Uploader
|
||||
}
|
||||
|
||||
func NewTranscribeService(jobRepo contract.TranscriptJobRepository, fileRepo contract.FileRepository, converter contract.AudioFileConverter) *TranscribeService {
|
||||
func NewTranscribeService(
|
||||
jobRepo contract.TranscriptJobRepository,
|
||||
fileRepo contract.FileRepository,
|
||||
converter contract.AudioFileConverter,
|
||||
s3Service contract.YandexS3Uploader,
|
||||
) *TranscribeService {
|
||||
return &TranscribeService{
|
||||
jobRepo: jobRepo,
|
||||
fileRepo: fileRepo,
|
||||
converter: converter,
|
||||
s3Service: s3Service,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -173,14 +179,8 @@ func (s *TranscribeService) FindAndRunTranscribeJob() error {
|
||||
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)
|
||||
err = s.s3Service.UploadFile(filePath, destFileRecord.FileName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -192,7 +192,7 @@ func (s *TranscribeService) FindAndRunTranscribeJob() error {
|
||||
}
|
||||
|
||||
// Формируем S3 URI для файла
|
||||
s3URI := s3Service.FileUrl(destFileRecord.FileName)
|
||||
s3URI := s.s3Service.FileUrl(destFileRecord.FileName)
|
||||
|
||||
// Запускаем асинхронное распознавание
|
||||
operationID, err := speechKitService.RecognizeFileFromS3(s3URI)
|
||||
|
||||
Reference in New Issue
Block a user