package whisperx import ( "context" "encoding/json" "errors" "fmt" "io" "net" "net/http" "net/url" "os" "strings" "time" "gitea.maximumdirect.net/eric/narratio/internal/fileops" ) const ( defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024 ) // HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings. type HTTPClientConfig struct { TranscribeURL string Language string Timeout time.Duration Retries int RetryDelay time.Duration HTTPClient *http.Client MaxResponseBytes int64 } // HTTPClient is a real WhisperX HTTP adapter implementation. type HTTPClient struct { url *url.URL language string timeout time.Duration retries int retryDelay time.Duration httpClient *http.Client maxResponseBytes int64 openAudio func(string) (io.ReadCloser, error) } // NewHTTPClientFromConfigValues builds a client from config values and parses durations once. func NewHTTPClientFromConfigValues(transcribeURL, language, timeout, retryDelay string, retries int) (*HTTPClient, error) { if strings.TrimSpace(timeout) == "" { return nil, fmt.Errorf("whisperx timeout is required") } parsedTimeout, err := time.ParseDuration(timeout) if err != nil { return nil, fmt.Errorf("parse whisperx timeout %q: %w", timeout, err) } if strings.TrimSpace(retryDelay) == "" { return nil, fmt.Errorf("whisperx retry_delay is required") } parsedRetryDelay, err := time.ParseDuration(retryDelay) if err != nil { return nil, fmt.Errorf("parse whisperx retry_delay %q: %w", retryDelay, err) } return NewHTTPClient(HTTPClientConfig{ TranscribeURL: transcribeURL, Language: language, Timeout: parsedTimeout, Retries: retries, RetryDelay: parsedRetryDelay, }) } // NewHTTPClient constructs an HTTP WhisperX client with validated settings. func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) { if strings.TrimSpace(cfg.TranscribeURL) == "" { return nil, fmt.Errorf("whisperx transcribe_url is required") } u, err := url.Parse(cfg.TranscribeURL) if err != nil || !u.IsAbs() || u.Host == "" || !isHTTPURLScheme(u.Scheme) { if err != nil { return nil, fmt.Errorf("invalid whisperx transcribe_url %q: %w", cfg.TranscribeURL, err) } return nil, fmt.Errorf("invalid whisperx transcribe_url %q: must be an absolute http or https URL", cfg.TranscribeURL) } if cfg.Timeout <= 0 { return nil, fmt.Errorf("whisperx timeout must be > 0") } if cfg.Retries < 0 { return nil, fmt.Errorf("whisperx retries must be >= 0") } if cfg.RetryDelay < 0 { return nil, fmt.Errorf("whisperx retry_delay must be >= 0") } if strings.TrimSpace(cfg.Language) == "" { return nil, fmt.Errorf("whisperx language is required") } client := cfg.HTTPClient if client == nil { transport := http.DefaultTransport.(*http.Transport).Clone() transport.ExpectContinueTimeout = 100 * time.Millisecond client = &http.Client{Transport: transport} } maxBytes := cfg.MaxResponseBytes if maxBytes <= 0 { maxBytes = defaultMaxWhisperXResponseBytes } return &HTTPClient{ url: u, language: cfg.Language, timeout: cfg.Timeout, retries: cfg.Retries, retryDelay: cfg.RetryDelay, httpClient: client, maxResponseBytes: maxBytes, openAudio: func(path string) (io.ReadCloser, error) { return os.Open(path) }, }, nil } // Transcribe submits one audio file to WhisperX and writes validated JSON output atomically. func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (TranscribeResult, error) { if c == nil { return TranscribeResult{}, fmt.Errorf("whisperx http client is nil") } if strings.TrimSpace(req.AudioPath) == "" { return TranscribeResult{}, fmt.Errorf("whisperx transcribe request audio path is required") } if strings.TrimSpace(req.OutputRawTranscriptPath) == "" { return TranscribeResult{}, fmt.Errorf("whisperx transcribe request output path is required") } start := time.Now() result := TranscribeResult{ OutputRawTranscriptPath: req.OutputRawTranscriptPath, } attempts := c.retries + 1 var lastErr error for attempt := 1; attempt <= attempts; attempt++ { result.Attempts = attempt attemptCtx, cancel := context.WithTimeout(ctx, c.timeout) status, body, err := c.doTranscribeAttempt(attemptCtx, req.AudioPath) cancel() if status > 0 { result.HTTPStatus = status } if err == nil { if err := c.validateJSON(body); err != nil { result.Duration = time.Since(start) return result, fmt.Errorf("whisperx attempt %d returned invalid json: %w", attempt, err) } if err := writeFileAtomic(req.OutputRawTranscriptPath, body, fileops.WorkspaceFileMode); err != nil { result.Duration = time.Since(start) return result, fmt.Errorf("whisperx write transcript output %q: %w", req.OutputRawTranscriptPath, err) } result.Duration = time.Since(start) result.Metadata = map[string]any{ "adapter": "whisperx_http", } return result, nil } lastErr = err if !c.shouldRetry(ctx, err, status) || attempt == attempts { break } if c.retryDelay > 0 { timer := time.NewTimer(c.retryDelay) select { case <-ctx.Done(): timer.Stop() result.Duration = time.Since(start) return result, fmt.Errorf("whisperx attempt %d aborted during retry delay: %w", attempt, ctx.Err()) case <-timer.C: } } } result.Duration = time.Since(start) if lastErr == nil { lastErr = fmt.Errorf("whisperx transcription failed without explicit error") } return result, fmt.Errorf("whisperx transcription failed after %d attempt(s): %w", result.Attempts, lastErr) } func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) { upload, err := newMultipartBodyFactory(ctx, audioPath, c.language, c.openAudio) if err != nil { return 0, nil, fmt.Errorf("prepare whisperx request body: %w", err) } defer func() { _ = upload.Close() }() body, err := upload.NewBody() if err != nil { return 0, nil, fmt.Errorf("open whisperx request body: %w", err) } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), body) if err != nil { return 0, nil, fmt.Errorf("build whisperx request: %w", err) } req.Header.Set("Content-Type", upload.contentType) req.Header.Set("Expect", "100-continue") req.GetBody = upload.NewBody resp, err := c.httpClient.Do(req) if err != nil { _ = upload.Close() if sourceErr := upload.SourceError(true); sourceErr != nil { return 0, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) } if ctxErr := ctx.Err(); ctxErr != nil { return 0, nil, fmt.Errorf("perform whisperx request: %w", ctxErr) } return 0, nil, fmt.Errorf("perform whisperx request: %w", err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { _ = upload.Close() if sourceErr := upload.SourceError(false); sourceErr != nil { return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) } if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil { return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) } return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode) } data, responseErr := readWhisperXResponse(resp.Body, c.maxResponseBytes) _ = upload.Close() if sourceErr := upload.SourceError(false); sourceErr != nil { if responseErr != nil { return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", errors.Join(sourceErr, responseErr)) } return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) } if err := ctx.Err(); err != nil { return resp.StatusCode, nil, fmt.Errorf("complete whisperx request: %w", err) } if responseErr != nil { return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", responseErr) } return resp.StatusCode, data, nil } func (c *HTTPClient) validateJSON(data []byte) error { var v any if err := json.Unmarshal(data, &v); err != nil { return err } return nil } func (c *HTTPClient) shouldRetry(parent context.Context, err error, status int) bool { if parent.Err() != nil { return false } if status == http.StatusTooManyRequests || status >= 500 { return true } if status >= 400 && status < 500 { return false } if errors.Is(err, context.Canceled) { return false } if errors.Is(err, context.DeadlineExceeded) { return true } var nerr net.Error if errors.As(err, &nerr) { return true } return false } func readWhisperXResponse(r io.Reader, maxBytes int64) ([]byte, error) { limited := io.LimitReader(r, maxBytes+1) data, err := io.ReadAll(limited) if err != nil { return nil, err } if int64(len(data)) > maxBytes { return nil, fmt.Errorf("whisperx response exceeds configured limit of %d bytes", maxBytes) } return data, nil } func isHTTPURLScheme(scheme string) bool { switch strings.ToLower(scheme) { case "http", "https": return true default: return false } } func writeFileAtomic(path string, data []byte, perm os.FileMode) error { if strings.TrimSpace(path) == "" { return fmt.Errorf("path is required") } if err := fileops.WriteFileAtomic(path, data, perm); err != nil { return fmt.Errorf("write file %q: %w", path, err) } return nil }