[codex] fix(subtitles): stream live transcribe_translate cues (#177)

* fix(subtitles): stream live transcribe_translate cues

* fix(subtitles): harden live AI transcription
This commit is contained in:
Quick
2026-06-18 12:16:27 -04:00
committed by GitHub
parent 6e1f79e8a9
commit b3198276f7
10 changed files with 1590 additions and 109 deletions
@@ -0,0 +1,311 @@
# Plan: interleave `transcribe_translate` so live AI subtitles follow the playhead
Issue: https://github.com/Silo-Server/silo-server/issues/154
Area: `internal/subtitles/ai` (+ `internal/playback` extraction, `internal/config`)
Type: bug fix to an existing (shipped) capability — makes the chained AI-subtitle
path deliver its designed live behavior. No new client-facing contract.
Commands assume the repository root is the cwd.
## Problem (confirmed)
`Service.runTranscribe` (`internal/subtitles/ai/service.go`) runs the chained
`transcribe_translate` kind as two **global, sequential** stages:
1. Transcribe the **entire** file (progress 5%→70%). Transcript cues are not
streamed for the chained kind —
`streamTranscript := streaming && job.Kind == JobKindTranscribe`
(`service.go:417`), so it is `false` for `transcribe_translate`.
2. Only then translate **all** cues (70%→95%), streaming each translated batch
as it lands (`service.go:482-493`).
Because translation cannot start until step 1 finishes for the whole file, the
first translated cue streams minutes into the job, long past the playhead.
### Two distinct bottlenecks (both must be addressed)
- **Transcription stage is global.** Translation waits for the whole-file ASR
pass. This is the dominant term for a feature film (~20 min ASR) and the
primary cause. Fixed by interleaving translation per chunk (Phase 1).
- **Extraction is global and precedes any callback.** `Transcribe` calls
`t.extract(...)` (`transcriber.go:124`) before the `onChunk` loop, and
`ExtractAudioChunks` blocks on a single whole-track ffmpeg pass
(`internal/playback/audio_extract.go:34,65`). So even with per-chunk
interleave and small chunks, the first cue still waits for full-file
extraction (~1–2 min for a 2h film). Fixed by playhead-first incremental
extraction (Phase 2).
The transcriber already exposes the per-chunk seam: `Transcribe` invokes
`onChunk(cues, done, total)` per chunk, playhead-first (`transcriber.go:111-174`),
and plain `transcribe` already streams through it (`service.go:426-429`).
Secondary defect found while confirming: `WhisperTranscriber.SetExtraction`
clamps any `chunkSeconds` outside `[60, 600]` to the **default 600**
(`transcriber.go:100-102`), so a small live chunk value silently becomes 10 min
instead of clamping to the floor.
## Goal & honest latency targets
Translated cues stream playhead-first for a session-attached
`transcribe_translate` job, instead of waiting for whole-file transcription.
Time-to-first-translated-cue, by phase:
- **Today:** `extract(full) + transcribe(full) + translate(first batch)` — many
minutes (≈ full ASR pass).
- **After Phase 1 (interleave, existing full extraction):**
`extract(full) + transcribe(first chunk) + translate(first chunk)` — removes
the whole-file ASR pass from the critical path; ≈ extraction time (~1–2 min for
a 2h film), then cues follow the playhead. Already usable for live after a
one-time wait.
- **After Phase 2 (playhead-first incremental extraction):**
`extract(one chunk near playhead) + transcribe + translate` — single-digit
seconds. This is what the issue means by "within seconds."
Phase 1 delivers most of the value with low risk; Phase 2 closes the remaining
extraction latency. Ship Phase 1 first; do not claim "within seconds" until
Phase 2 lands.
### Non-goals (this plan)
- Pacing/throttling to the playhead (rolling window). Optional in the issue;
follow-up below.
- Cross-chunk translation-context continuity beyond batch context neighbours.
## Decision: interleave only when streaming
**Session-attached (live) jobs** use the interleaved per-chunk path.
**Background (non-streaming) `transcribe_translate` jobs keep the current
whole-file transcribe-then-translate-at-end path unchanged.** Rationale: the
issue is live-only; this preserves background output byte-for-byte (full
cross-file translation context, no incremental-extraction cost) and keeps the
two behaviors cleanly separated by the existing `streaming` boolean. The
shared transcript-storage step is factored so it is not duplicated.
---
## Phase 1 — interleave translation per chunk (streaming jobs)
### 1a. `internal/subtitles/ai/service.go` — `runTranscribe`
For the chained kind **when `streaming`**, replace the kind-specific `onChunk`
body (`service.go:423-430`) with per-chunk translate-and-stream:
- Capture a translate error in a closure: `var translateErr error` and an
accumulator `var translatedAccum []SubtitleCue`.
- In `onChunk(chunk, done, total)`:
- If `translateErr != nil` or `len(chunk) == 0`, return (skip silence; stop
translating after a prior failure). Do **not** abort transcription.
- Else call `s.translator.Translate` on **that chunk's** cues. On error: set
`translateErr`, emit `TranslationFailed` once, and return (let transcription
continue). On success: stream translated cues via `TranslationCues` and
append to `translatedAccum`.
- Progress: single 5%→95% band keyed on chunk `done/total`
(message "Transcribing & translating").
- Keep `onChunk` signature unchanged (no error return). Aborting on translate
failure via the callback is intentionally avoided — see 1b.
After `Transcribe` returns:
- **Always store the transcript track** (unchanged, `service.go:447-466`,
including the transcript `SubtitleReady` broadcast). This must run regardless
of `translateErr`.
- If `translateErr != nil`: `finishWithError(ctx, job, translateErr)`. The
transcript is already persisted (cache preserved), and the live session
already received `TranslationFailed`.
- Else: assemble the translated track from `translatedAccum` (`sortCuesByStart`
+ `StoreSubtitle`, `service.go:499-514`), `CompleteJob`, and
`TranslationCompleted` (`service.go:515-522`).
- Remove the whole-file translate-at-end (`service.go:479-497`) for the
streaming path only.
Non-streaming chained jobs fall through to the existing whole-file path,
untouched.
### 1b. Transcript-on-translation-failure (resolves review P1)
Today the full transcript is stored before translation starts
(`service.go:447` then `:482`), so a translation failure still leaves a
transcript/cache row and an already-broadcast `SubtitleReady`. The interleaved
path preserves this exactly because:
- a per-chunk translate failure sets `translateErr` but lets the ASR pass run to
completion (it does **not** return an error from `onChunk`), so `Transcribe`
returns the full cue set normally;
- the transcript is then stored on the same post-transcribe path as today,
before the job is failed for the translated track.
An ASR failure or context cancellation still returns an error from `Transcribe`
and yields no transcript — identical to today.
### 1c. Live progress semantics (resolves review P2)
`SubtitleTranslationCuesPayload.Done/Total` is documented as overall progress
(`internal/playback/realtime.go:172-181`). For live `transcribe_translate`:
- `TranslationCues` carries **`done = chunks completed, total = total chunk
count`** — the same chunk-granular convention plain `transcribe` already emits
(`service.go:427-428`). Monotonic and stable across chunks.
- `TranslationStarted.TotalCues = 0` (indeterminate), consistent with the
transcribe path today (`service.go:412`), since the cue total is unknown before
ASR completes.
- Cues carry absolute Start/End, so the client places them regardless of
arrival order; unpause is driven by cue arrival near the playhead, not by
`Done/Total`. Confirm with client teams that `Done/Total` is treated as a
fraction, not absolute cue indices (clients already receive chunk counts from
the plain `transcribe` path, so this should hold).
### 1d. Small live chunks + clamp fix
- Add `TranscribeJobRequest.ChunkSeconds int` (0 = configured default). In
`Transcribe`, use `req.ChunkSeconds` when `> 0`, else the atomic default
(`transcriber.go:121-122`); validate/clamp identically to `SetExtraction`.
- Fix the clamp in `SetExtraction` (`transcriber.go:99-105`): clamp
out-of-range to the nearest bound, not to the 600 default.
- Lower `minASRChunkSeconds` from 60 to 15 (`transcriber.go:26`); update the
doc comment about request count / boundary word-clip tradeoff.
- Config: add `subtitle_ai.live_asr_chunk_seconds` (default 30):
- `internal/config/config.go`: new field beside `ASRChunkSeconds`
(`config.go:274-277`).
- `internal/config/db_loader.go`: load it next to `asr_chunk_seconds`
(`db_loader.go:472-476`).
- `internal/subtitles/ai/config.go`: add `LiveASRChunkSeconds int` to the
service `Config` so the service can set `req.ChunkSeconds` for streaming jobs.
- `internal/api/router.go`: plumb through `effectiveSubtitleAIConfig` and the
`OnConfigChange` `UpdateConfig` (`router.go:1024-1055`).
### 1e. Phase 1 tests
- `service_test.go`: fake `Transcriber` emitting several chunks playhead-first
via `onChunk`; fake `Translator` recording calls. For a streaming
`transcribe_translate` job assert: one `Translate` per non-empty chunk; first
`TranslationCues` streamed before the last chunk is transcribed; empty chunks
skipped; final stored translated track = all translated cues sorted by start;
`Done/Total` is chunk-granular and monotonic.
- Failure case: a translate error on chunk N sets the failure but transcription
completes, the transcript track **is** stored, `TranslationFailed` is emitted,
and the job ends via `finishWithError`.
- Non-streaming `transcribe_translate`: unchanged whole-file path; correct final
track; no `TranslationCues`.
- `transcriber_test.go`: `SetExtraction` clamp fix and `ChunkSeconds` override
(incl. the new 15s floor).
---
## Phase 2 — playhead-first incremental extraction (true seconds latency)
Phase 1 still waits on full-file extraction before the first cue. Phase 2
removes that by extracting and consuming chunks incrementally, seeked to the
playhead.
### 2a. New incremental extractor in `internal/playback`
Add a streaming, seek-based extractor alongside `ExtractAudioChunks`:
```
ExtractAudioChunksFrom(ctx, filePath, audioTrackIndex, dir, ffmpegPath,
startSec float64, chunkSeconds int,
onSegment func(AudioChunk) error) error
```
- Runs one ffmpeg pass with `-ss <startSec> -i <file> ... -f segment
-segment_list segments.csv` covering `startSec`→end.
- Consumes segments **as they complete**: poll the segment-list CSV (ffmpeg
appends `filename,start,end` when it closes each segment) on a short interval;
for each new row, invoke `onSegment(AudioChunk{Path, Start: startSec + csvStart})`.
Reconcile any trailing rows after `cmd.Wait()` so the final segment is not
missed.
- Honors `ctx` (CommandContext kills ffmpeg on cancel) and `onSegment` errors
(stop the pass). Caller owns `dir`; segments are deleted by the consumer after
ASR to cap disk at one extraction.
- Preserves exact per-segment starts from the CSV, so timing does not accumulate
drift within a pass (only the initial `-ss` seek is packet-accurate, sub-second,
and is further corrected by the existing `ProbeAudioStartOffset`).
Rejected alternative — **per-chunk seek extraction** (`-ss s -t d` per chunk):
simpler control flow but O(n) ffmpeg spawns and packet-aligned per-chunk seeks
that can clip words at every seam. The single-pass streaming consumer keeps the
existing segment-muxer boundary quality and far fewer process spawns.
### 2b. Transcriber incremental mode
- Add `TranscribeJobRequest.Incremental bool`, set by the service for streaming
jobs.
- When `Incremental`, `Transcribe` drives playhead-first in two streaming passes
(reproducing today's `chunkOrderForPosition` order): pass A `startSec =
playhead → end`, then pass B `0 → playhead`. Each completed segment is fed
straight into the existing per-chunk ASR loop body (transcribe → `onChunk`),
which Phase 1 already wires to translate+stream. The full cue set is still
accumulated and returned for transcript storage.
- When not `Incremental` (background, and plain transcribe unless we opt it in),
keep the existing whole-file `ExtractAudioChunks` + `chunkOrderForPosition`
path.
Note: timing composition (`-ss` reset PTS + segment CSV start + probe offset)
must be verified with a fixture; see tests.
### 2c. Phase 2 tests
- `audio_extract` test (or transcriber test with an injected extractor) covering
incremental emission order, exact-start mapping with a `startSec` offset, and
ctx-cancel mid-pass.
- Transcriber test asserting `Incremental` yields playhead-first chunk order
across the two passes and returns the complete cue set for transcript storage.
- Confirm first `onChunk` fires after ~one chunk of audio, not after full-file
extraction (timing/ordering assertion via a fake extractor, not wall-clock).
---
## Multi-repo / client impact
None expected to the contract. Reuses the existing notifier protocol
(`TranslationStarted` / `TranslationCues` / `TranslationCompleted` /
`TranslationFailed`, `notifier.go`) under the same translated-track `trackKey`.
Translated cues simply begin arriving far earlier, and `Done/Total` keeps the
chunk-granular meaning plain `transcribe` already uses (1c). Worth a smoke test
on one `silo-android` / `silo-apple` client to confirm the live track renders
cues arriving before the transcript track completes, and that `Done/Total` is
treated as a fraction.
## Risks & tradeoffs
- **Translation context at chunk boundaries (live only).** Per-chunk translation
resets the batch context-neighbour window at each ~30s boundary — accepted
quality tradeoff for live; background jobs are unchanged.
- **More/smaller ASR requests for live jobs.** 30s chunks mean more requests;
per-chunk timeout (`chunkSeconds * asrChunkTimeoutFactor`) and per-spawn
overhead scale; the 15s floor bounds the worst case.
- **Incremental extraction reliability (Phase 2).** Segment-list polling + the
`-ss` seek path need fixtures for boundary timing, the final-segment race, and
cancel/cleanup. Single ffmpeg pass per direction keeps cost low.
- **Boundary word-clips.** Smaller live chunks add more fixed-length boundaries
where a straddling word can clip — already a documented v1 limitation; the
silence-aligned follow-up would address it.
## Out of scope / follow-ups
- **Pace to playback (issue item 3):** rolling window ahead of the playhead to
bound work; separate change once interleave + incremental extraction land.
- **Cross-chunk translation context:** feed the previous chunk's tail source
cues as untranslated context into each chunk's translate call.
- **Silence-aligned chunk boundaries:** reduce boundary word-clips that smaller
live chunks make more frequent.
## v1 scope note
`docs/architecture/v1-scope.md` is **NOT LOCKED**; the issue carries
`v1-proposed`. This is a **bug fix** to an already-shipped capability (no new
user-facing capability, endpoint, or response field — only an internal config
key and corrected streaming behavior), so it proceeds under the "bug fixes
proceed normally" clause. Link the PR to the AI-subtitle capability item per the
v1 PR requirements.
## Pre-push checklist
- `make lint`
- `cd web && pnpm run lint && pnpm run format:check` (run only if web touched —
none expected)
- `make verify-local-paths`
- Go tests for `internal/subtitles/ai` and `internal/playback` in a
libvips-capable container (a bare-host `go test ./...` silently skips CGO
packages).
+1
View File
@@ -2915,6 +2915,7 @@ func effectiveSubtitleAIConfig(cfg *config.Config) (subtitleai.Config, string) {
ASRModel: cfg.AI.ASRModel,
BatchSize: cfg.SubtitleAI.BatchSize,
ContextNeighbors: cfg.SubtitleAI.ContextNeighbors,
LiveASRChunkSeconds: cfg.SubtitleAI.LiveASRChunkSeconds,
TranscribeQuotaJobs: cfg.SubtitleAI.TranscribeQuotaJobs,
TranscribeQuotaPeriod: cfg.SubtitleAI.TranscribeQuotaPeriod,
}, disabledGateway
+4 -1
View File
@@ -272,9 +272,12 @@ type SubtitleAIConfig struct {
BatchSize int `yaml:"-"`
ContextNeighbors int `yaml:"-"`
// ASRChunkSeconds is the audio chunk length per transcription request
// (60..600). Shorter chunks bound Whisper timestamp drift on long files;
// (15..600). Shorter chunks bound Whisper timestamp drift on long files;
// longer chunks mean fewer requests and fewer boundary word-clips.
ASRChunkSeconds int `yaml:"-"`
// LiveASRChunkSeconds is the smaller per-request chunk length for
// session-attached live subtitle jobs (15..600).
LiveASRChunkSeconds int `yaml:"-"`
// TranscribeQuotaJobs caps how many transcription jobs each non-admin user
// may start per rolling quota period (0 = unlimited).
TranscribeQuotaJobs int `yaml:"-"`
+5
View File
@@ -474,6 +474,11 @@ func LoadFromDB(m map[string]string) (*Config, error) {
return nil, err
}
cfg.SubtitleAI.ASRChunkSeconds = subtitleAIChunkSeconds
subtitleAILiveChunkSeconds, err := intOr(m, "subtitle_ai.live_asr_chunk_seconds", 30)
if err != nil {
return nil, err
}
cfg.SubtitleAI.LiveASRChunkSeconds = subtitleAILiveChunkSeconds
// A bad quota row must not block startup: fall back to "no quota" /
// the daily window with a warning instead of failing the config load.
transcribeQuotaJobs, err := intOr(m, "subtitle_ai.transcribe_quota_jobs", 0)
+235 -18
View File
@@ -6,6 +6,7 @@ import (
"encoding/csv"
"encoding/json"
"fmt"
"io"
"math"
"os"
"os/exec"
@@ -13,6 +14,7 @@ import (
"sort"
"strconv"
"strings"
"time"
)
// AudioChunk is one extracted piece of an audio track. Start is the chunk's
@@ -40,25 +42,10 @@ func ExtractAudioChunks(ctx context.Context, filePath string, audioTrackIndex in
}
listPath := filepath.Join(dir, "segments.csv")
args := []string{
"-i", filePath,
"-vn", "-sn", "-dn",
"-map", fmt.Sprintf("0:a:%d", audioTrackIndex),
"-ac", "1",
"-ar", "16000",
"-c:a", "pcm_s16le",
"-f", "segment",
"-segment_time", strconv.Itoa(chunkSeconds),
"-segment_list", listPath,
"-segment_list_type", "csv",
"-y", filepath.Join(dir, "chunk%05d.wav"),
}
args := audioChunkExtractionArgs(filePath, audioTrackIndex, listPath,
filepath.Join(dir, "chunk%05d.wav"), 0, chunkSeconds)
ffmpeg := "ffmpeg"
if ffmpegPath != "" {
ffmpeg = ffmpegPath
}
cmd := exec.CommandContext(ctx, ffmpeg, args...)
cmd := exec.CommandContext(ctx, audioExtractionFFmpegBinary(ffmpegPath), args...)
var stderr bytes.Buffer
cmd.Stderr = &stderr
@@ -96,6 +83,126 @@ func ExtractAudioChunks(ctx context.Context, filePath string, audioTrackIndex in
return chunks, nil
}
// ExtractAudioChunksFrom extracts one audio track starting at startSec and calls
// onSegment as each WAV segment is closed by ffmpeg. onSegment is synchronous:
// if it blocks on ASR/translation, ffmpeg may continue extracting later chunks
// into dir, so callers should remove processed chunks promptly. The caller owns
// dir and cleanup. Segment starts are absolute media positions.
func ExtractAudioChunksFrom(
ctx context.Context,
filePath string,
audioTrackIndex int,
dir, ffmpegPath string,
startSec float64,
chunkSeconds int,
onSegment func(AudioChunk) error,
) error {
if chunkSeconds <= 0 {
chunkSeconds = 600
}
if audioTrackIndex < 0 {
audioTrackIndex = 0
}
if startSec < 0 {
startSec = 0
}
listPath := filepath.Join(dir, "segments.csv")
args := audioChunkExtractionArgs(filePath, audioTrackIndex, listPath,
filepath.Join(dir, "chunk%05d.wav"), startSec, chunkSeconds)
cmdCtx, cancel := context.WithCancel(ctx)
defer cancel()
cmd := exec.CommandContext(cmdCtx, audioExtractionFFmpegBinary(ffmpegPath), args...)
var stderr bytes.Buffer
cmd.Stderr = &stderr
if err := cmd.Start(); err != nil {
return fmt.Errorf("start ffmpeg audio extraction: %w", err)
}
waitCh := make(chan error, 1)
go func() {
waitCh <- cmd.Wait()
}()
seen := map[string]bool{}
tailer := &segmentListTailer{}
emit := func(flush bool) error {
return emitNewSegmentListEntries(listPath, dir, startSec, seen, tailer, flush, onSegment)
}
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
for {
select {
case waitErr := <-waitCh:
segmentErr := emit(true)
if segmentErr != nil {
return segmentErr
}
if waitErr != nil {
if ctx.Err() != nil {
return ctx.Err()
}
return fmt.Errorf("ffmpeg audio extraction failed: %w (stderr: %s)",
waitErr, truncateStderr(stderr.String()))
}
if len(seen) == 0 {
return fmt.Errorf("ffmpeg produced no audio chunks for track %d", audioTrackIndex)
}
return nil
case <-ticker.C:
if err := emit(false); err != nil {
cancel()
<-waitCh
return err
}
case <-ctx.Done():
cancel()
<-waitCh
return ctx.Err()
}
}
}
func audioChunkExtractionArgs(
filePath string,
audioTrackIndex int,
listPath string,
outPattern string,
startSec float64,
chunkSeconds int,
) []string {
args := []string{}
if startSec > 0 {
args = append(args, "-ss", strconv.FormatFloat(startSec, 'f', 3, 64))
}
args = append(args,
"-i", filePath,
"-vn", "-sn", "-dn",
"-map", fmt.Sprintf("0:a:%d", audioTrackIndex),
"-ac", "1",
"-ar", "16000",
"-c:a", "pcm_s16le",
"-f", "segment",
"-segment_time", strconv.Itoa(chunkSeconds),
"-segment_list", listPath,
"-segment_list_type", "csv",
"-y", outPattern,
)
return args
}
func audioExtractionFFmpegBinary(ffmpegPath string) string {
if ffmpegPath != "" {
return ffmpegPath
}
return "ffmpeg"
}
// parseSegmentList reads ffmpeg's CSV segment list (filename,start,end per
// line) into a filename → start map. Best effort: a missing or malformed list
// yields an empty map and callers fall back to nominal chunk starts.
@@ -123,6 +230,116 @@ func parseSegmentList(listPath string) map[string]float64 {
return starts
}
func emitNewSegmentListEntries(
listPath, dir string,
startOffset float64,
seen map[string]bool,
tailer *segmentListTailer,
flush bool,
onSegment func(AudioChunk) error,
) error {
entries := tailer.read(listPath, flush)
for _, entry := range entries {
key := filepath.Base(entry.name)
if seen[key] {
continue
}
seen[key] = true
path := entry.name
if !filepath.IsAbs(path) {
path = filepath.Join(dir, path)
}
if onSegment != nil {
if err := onSegment(AudioChunk{Path: path, Start: startOffset + entry.start}); err != nil {
return err
}
}
}
return nil
}
type segmentListEntry struct {
name string
start float64
}
type segmentListTailer struct {
offset int64
pending string
}
func (t *segmentListTailer) read(listPath string, flush bool) []segmentListEntry {
f, err := os.Open(listPath)
if err != nil {
return nil
}
defer f.Close()
info, err := f.Stat()
if err == nil && info.Size() < t.offset {
t.offset = 0
t.pending = ""
}
if _, err := f.Seek(t.offset, io.SeekStart); err != nil {
return nil
}
data, err := io.ReadAll(f)
if err != nil {
return nil
}
t.offset += int64(len(data))
text := t.pending + string(data)
if text == "" {
return nil
}
lineEnd := strings.LastIndexByte(text, '\n')
if lineEnd < 0 {
if !flush {
t.pending = text
return nil
}
t.pending = ""
return parseSegmentListEntries(strings.NewReader(text))
}
complete := text[:lineEnd+1]
t.pending = text[lineEnd+1:]
if flush && t.pending != "" {
complete += t.pending
t.pending = ""
}
return parseSegmentListEntries(strings.NewReader(complete))
}
func parseSegmentListEntries(r io.Reader) []segmentListEntry {
reader := csv.NewReader(r)
var out []segmentListEntry
for {
row, err := reader.Read()
if err == io.EOF {
break
}
if err != nil {
// Keep completed rows; malformed rows are ignored best-effort.
break
}
if len(row) < 2 {
continue
}
start, err := strconv.ParseFloat(strings.TrimSpace(row[1]), 64)
if err != nil || start < 0 {
continue
}
name := strings.TrimSpace(row[0])
if name == "" {
continue
}
out = append(out, segmentListEntry{name: name, start: start})
}
return out
}
// maxPlausibleAudioDelay caps the probed audio/container start delta; beyond
// this it is far more likely a probing artifact than a real stream delay.
const maxPlausibleAudioDelay = 30.0
+6 -5
View File
@@ -12,11 +12,12 @@ type Config struct {
TranslateEnabled bool
// TranscribeEnabled gates Whisper ASR generation (transcribe /
// transcribe_translate jobs).
TranscribeEnabled bool
ChatModel string // chat-completions model used for translation
ASRModel string // audio-transcription model used for ASR
BatchSize int // cues per translation request
ContextNeighbors int // preceding source cues sent as untranslated context
TranscribeEnabled bool
ChatModel string // chat-completions model used for translation
ASRModel string // audio-transcription model used for ASR
BatchSize int // cues per translation request
ContextNeighbors int // preceding source cues sent as untranslated context
LiveASRChunkSeconds int // smaller ASR chunks for session-attached live jobs
// TranscribeQuotaJobs caps how many transcription jobs (transcribe /
// transcribe_translate) each non-admin user may start per rolling quota
// period; 0 means unlimited.
+110 -39
View File
@@ -309,7 +309,7 @@ func (s *Service) run(ctx context.Context, job *Job) {
TargetLanguage: job.TargetLanguage,
}, func(batch []SubtitleCue, done, total int) {
// Map cue progress into the 5%..95% band; push the batch live.
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.05+0.9*float64(done)/float64(total), "Translating")
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.05, 0.9, done, total), "Translating")
if streaming {
s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
toStreamCues(batch), done, total)
@@ -325,34 +325,9 @@ func (s *Service) run(ctx context.Context, job *Job) {
// Persist in chronological order regardless of the playhead-first order used
// for translation/streaming.
sortCuesByStart(translated)
sub, err := s.store.StoreSubtitle(storeCtx, subtitles.StoreSubtitleRequest{
MediaFileID: job.MediaFileID,
UserID: job.RequestedBy,
Provider: providerTranslated,
Language: job.TargetLanguage,
Format: subtitles.FormatSRT,
ReleaseName: releaseName,
Data: SerializeSRT(translated),
})
if err != nil {
s.finishWithError(ctx, job, fmt.Errorf("store translated subtitle: %w", err))
if !s.finishTranslatedTrack(ctx, storeCtx, job, translated, releaseName, trackKey, streaming) {
return
}
if err := s.repo.CompleteJob(storeCtx, job.ID, sub.ID); err != nil {
s.logger.Warn("failed to mark subtitle ai job complete", "job", job.ID, "error", err)
}
if s.notifier != nil {
if streaming {
s.notifier.TranslationCompleted(storeCtx, job.SessionID, job.MediaFileID, job.ID, trackKey,
sub.ID, job.TargetLanguage, releaseName)
}
// Broadcast to other sessions watching this file (a no-op for the
// streaming requester, which already tracks its live track).
s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, releaseName)
}
}
// runTranscribe executes transcribe / transcribe_translate jobs: extract the
@@ -412,17 +387,37 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) {
s.notifier.TranslationStarted(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, liveLang, liveLabel, 0)
}
// Only a plain transcribe streams transcript cues — the chained kind
// streams the translated cues from its translation stage instead.
cfg := s.config()
// Plain transcribe streams transcript cues. A live chained job translates
// each non-empty ASR chunk immediately and streams translated cues from the
// same callback, while background chained jobs keep the whole-file path.
streamTranscript := streaming && job.Kind == JobKindTranscribe
streamTranslated := streaming && job.Kind == JobKindTranscribeTranslate
liveChunkSeconds := 0
if streaming {
liveChunkSeconds = cfg.LiveASRChunkSeconds
}
liveTranslate := liveTranslateState{sourceLanguage: hint}
cues, detected, err := s.transcriber.Transcribe(ctx, TranscribeJobRequest{
FilePath: file.FilePath,
AudioTrackIndex: audioIdx,
LanguageHint: hint,
StartPosition: job.StartPosition,
}, func(chunk []SubtitleCue, done, total int) {
DurationSeconds: float64(file.Duration),
ChunkSeconds: liveChunkSeconds,
Incremental: streaming,
}, func(chunk []SubtitleCue, chunkLanguage string, done, total int) {
if liveTranslate.sourceLanguage == "" && chunkLanguage != "" {
liveTranslate.sourceLanguage = chunkLanguage
}
if streamTranslated {
s.translateLiveChunk(ctx, job, trackKey, chunk, done, total, &liveTranslate)
return
}
// Transcription occupies the 5%..70% progress band (chunk granularity).
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.05+0.65*float64(done)/float64(total), "Transcribing")
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.05, 0.65, done, total), "Transcribing")
if streamTranscript {
s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
toStreamCues(chunk), done, total)
@@ -476,6 +471,16 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) {
return
}
if streamTranslated {
if liveTranslate.err != nil {
s.finishWithErrorNotify(ctx, job, liveTranslate.err, !liveTranslate.failedNotified)
return
}
translatedLabel := translatedReleaseName(language, job.TargetLanguage)
s.finishTranslatedTrack(ctx, storeCtx, job, liveTranslate.cues, translatedLabel, trackKey, true)
return
}
// transcribe_translate: run the transcript through the regular translator.
// Cues already arrive playhead-first from chunk ordering, so the viewer's
// region translates (and streams) first here too.
@@ -485,7 +490,7 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) {
TargetLanguage: job.TargetLanguage,
}, func(batch []SubtitleCue, done, total int) {
// Translation occupies the 70%..95% band.
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.7+0.25*float64(done)/float64(total), "Translating")
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.7, 0.25, done, total), "Translating")
if streaming {
s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
toStreamCues(batch), done, total)
@@ -497,31 +502,86 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) {
}
storeCtx = context.WithoutCancel(ctx)
sortCuesByStart(translated)
translatedLabel := translatedReleaseName(language, job.TargetLanguage)
s.finishTranslatedTrack(ctx, storeCtx, job, translated, translatedLabel, trackKey, streaming)
}
type liveTranslateState struct {
sourceLanguage string
err error
failedNotified bool
cues []SubtitleCue
}
func (s *Service) translateLiveChunk(
ctx context.Context,
job *Job,
trackKey string,
chunk []SubtitleCue,
done, total int,
state *liveTranslateState,
) {
_ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning,
progressInBand(0.05, 0.9, done, total), "Transcribing & translating")
if state.err != nil || len(chunk) == 0 {
return
}
translatedChunk, err := s.translator.Translate(ctx, TranslateRequest{
Cues: chunk,
SourceLanguage: state.sourceLanguage,
TargetLanguage: job.TargetLanguage,
}, func(batch []SubtitleCue, _, _ int) {
s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
toStreamCues(batch), done, total)
})
if err != nil {
state.err = err
if !state.failedNotified {
s.notifier.TranslationFailed(ctx, job.SessionID, job.MediaFileID, job.ID,
trackKey, llm.Truncate(err.Error(), 500))
state.failedNotified = true
}
return
}
state.cues = append(state.cues, translatedChunk...)
}
func (s *Service) finishTranslatedTrack(
ctx context.Context,
storeCtx context.Context,
job *Job,
cues []SubtitleCue,
releaseName string,
trackKey string,
streaming bool,
) bool {
sortCuesByStart(cues)
sub, err := s.store.StoreSubtitle(storeCtx, subtitles.StoreSubtitleRequest{
MediaFileID: job.MediaFileID,
UserID: job.RequestedBy,
Provider: providerTranslated,
Language: job.TargetLanguage,
Format: subtitles.FormatSRT,
ReleaseName: translatedLabel,
Data: SerializeSRT(translated),
ReleaseName: releaseName,
Data: SerializeSRT(cues),
})
if err != nil {
s.finishWithError(ctx, job, fmt.Errorf("store translated subtitle: %w", err))
return
return false
}
if err := s.repo.CompleteJob(storeCtx, job.ID, sub.ID); err != nil {
s.logger.Warn("failed to mark subtitle ai job complete", "job", job.ID, "error", err)
}
if s.notifier != nil {
if streaming {
s.notifier.TranslationCompleted(storeCtx, job.SessionID, job.MediaFileID, job.ID, trackKey,
sub.ID, job.TargetLanguage, translatedLabel)
sub.ID, job.TargetLanguage, releaseName)
}
s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, translatedLabel)
s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, releaseName)
}
return true
}
// defaultAudioTrackIndex returns the index of the default-flagged audio
@@ -544,6 +604,10 @@ func transcribedReleaseName(lang string) string {
}
func (s *Service) finishWithError(ctx context.Context, job *Job, err error) {
s.finishWithErrorNotify(ctx, job, err, true)
}
func (s *Service) finishWithErrorNotify(ctx context.Context, job *Job, err error, notify bool) {
status := JobStatusFailed
msg := llm.Truncate(err.Error(), 500)
// Only a genuine cancellation (user cancel, or shutdown via cancel) becomes
@@ -555,7 +619,7 @@ func (s *Service) finishWithError(ctx context.Context, job *Job, err error) {
if dbErr := s.repo.FailJob(context.WithoutCancel(ctx), job.ID, status, msg); dbErr != nil {
s.logger.Warn("failed to record subtitle ai job failure", "job", job.ID, "error", dbErr)
}
if job.SessionID != "" && s.notifier != nil {
if notify && job.SessionID != "" && s.notifier != nil {
s.notifier.TranslationFailed(context.WithoutCancel(ctx), job.SessionID, job.MediaFileID, job.ID,
liveTrackKey(job.ID), msg)
}
@@ -564,6 +628,13 @@ func (s *Service) finishWithError(ctx context.Context, job *Job, err error) {
}
}
func progressInBand(start, width float64, done, total int) float64 {
if total <= 0 {
return start
}
return start + width*float64(done)/float64(total)
}
// loadSource resolves the combined player subtitle index to translatable cues,
// mirroring the index space used by the playback subtitle endpoints
// (external → embedded → downloaded). Embedded tracks are extracted to SRT via
+437 -5
View File
@@ -3,12 +3,16 @@ package ai
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
"github.com/Silo-Server/silo-server/internal/ai/jobrunner"
"github.com/Silo-Server/silo-server/internal/models"
"github.com/Silo-Server/silo-server/internal/playback"
"github.com/Silo-Server/silo-server/internal/subtitles"
)
// recordingRepo is a JobRepository that records ResetStaleJobs calls and
@@ -21,6 +25,20 @@ type recordingRepo struct {
inserts int
quotaUsed int // returned by CountTranscribeJobsByUserSince
lastJob Job
completed []int
failures []recordedFailure
progress []recordedProgress
}
type recordedFailure struct {
id int64
status JobStatus
message string
}
type recordedProgress struct {
progress float64
message string
}
// InsertJob mirrors the Postgres repo's atomic quota guard: a non-nil quota
@@ -47,12 +65,25 @@ func (r *recordingRepo) CountTranscribeJobsByUserSince(context.Context, int, tim
defer r.mu.Unlock()
return r.quotaUsed, nil
}
func (r *recordingRepo) UpdateProgress(context.Context, int64, JobStatus, float64, string) error {
func (r *recordingRepo) UpdateProgress(_ context.Context, _ int64, _ JobStatus, progress float64, message string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.progress = append(r.progress, recordedProgress{progress: progress, message: message})
return nil
}
func (r *recordingRepo) CompleteJob(context.Context, int64, int) error { return nil }
func (r *recordingRepo) FailJob(context.Context, int64, JobStatus, string) error { return nil }
func (r *recordingRepo) Heartbeat(context.Context, int64) error { return nil }
func (r *recordingRepo) CompleteJob(_ context.Context, _ int64, subtitleID int) error {
r.mu.Lock()
defer r.mu.Unlock()
r.completed = append(r.completed, subtitleID)
return nil
}
func (r *recordingRepo) FailJob(_ context.Context, id int64, status JobStatus, message string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.failures = append(r.failures, recordedFailure{id: id, status: status, message: message})
return nil
}
func (r *recordingRepo) Heartbeat(context.Context, int64) error { return nil }
func (r *recordingRepo) ResetStaleJobs(_ context.Context, before time.Time, _ string) (int64, error) {
r.mu.Lock()
@@ -73,7 +104,7 @@ func (r *recordingRepo) snapshot() (int, time.Time) {
type stubTranscriber struct{}
func (stubTranscriber) Transcribe(context.Context, TranscribeJobRequest,
func([]SubtitleCue, int, int)) ([]SubtitleCue, string, error) {
TranscribeChunkCallback) ([]SubtitleCue, string, error) {
return nil, "", nil
}
@@ -221,3 +252,404 @@ func TestRecoverReapsStaleJobsImmediately(t *testing.T) {
t.Errorf("stale cutoff = %v, want ~%v (now-staleJobThreshold)", before, want)
}
}
type runTranscribeFileResolver struct {
file *models.MediaFile
}
func (r runTranscribeFileResolver) GetByID(context.Context, int) (*models.MediaFile, error) {
return r.file, nil
}
type chunkedTranscriber struct {
chunks [][]SubtitleCue
detected string
trace *[]string
lastReq TranscribeJobRequest
}
func (t *chunkedTranscriber) Transcribe(_ context.Context, req TranscribeJobRequest,
onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error) {
t.lastReq = req
var all []SubtitleCue
for i, chunk := range t.chunks {
appendTrace(t.trace, fmt.Sprintf("transcribe:%d", i))
if onChunk != nil {
total := len(t.chunks)
if req.Incremental {
total = progressTotal(i+1, estimatedChunkTotal(req.DurationSeconds, req.ChunkSeconds))
}
onChunk(chunk, t.detected, i+1, total)
}
all = append(all, chunk...)
}
return all, t.detected, nil
}
type recordingTranslator struct {
calls []TranslateRequest
errOnCall int
trace *[]string
}
func (t *recordingTranslator) Translate(_ context.Context, req TranslateRequest,
onBatch func([]SubtitleCue, int, int)) ([]SubtitleCue, error) {
t.calls = append(t.calls, TranslateRequest{
Cues: cloneTestCues(req.Cues),
SourceLanguage: req.SourceLanguage,
TargetLanguage: req.TargetLanguage,
})
call := len(t.calls)
if t.errOnCall == call {
return nil, errors.New("translator chunk failed")
}
out := cloneTestCues(req.Cues)
for i := range out {
for j, line := range out[i].Lines {
out[i].Lines[j] = req.TargetLanguage + ":" + line
}
}
if len(out) > 0 {
appendTrace(t.trace, "translate:"+strings.Join(out[0].Lines, " "))
}
if onBatch != nil {
onBatch(out, len(out), len(out))
}
return out, nil
}
type recordingSubtitleStore struct {
stored []subtitles.StoreSubtitleRequest
}
func (s *recordingSubtitleStore) StoreSubtitle(_ context.Context, req subtitles.StoreSubtitleRequest) (*subtitles.DownloadedSubtitle, error) {
s.stored = append(s.stored, req)
return &subtitles.DownloadedSubtitle{
ID: len(s.stored),
MediaFileID: req.MediaFileID,
Provider: req.Provider,
Language: req.Language,
Format: req.Format,
ReleaseName: req.ReleaseName,
}, nil
}
func (s *recordingSubtitleStore) GetSubtitleContent(context.Context, int) (*subtitles.DownloadedSubtitle, []byte, error) {
return nil, nil, errors.New("not implemented")
}
type notifierEvent struct {
kind string
cues []playback.StreamCue
done int
total int
message string
}
type recordingNotifier struct {
events []notifierEvent
trace *[]string
}
func (n *recordingNotifier) SubtitleReady(_ context.Context, _ int, subtitleID int, language, _ string) {
n.events = append(n.events, notifierEvent{kind: fmt.Sprintf("ready:%d:%s", subtitleID, language)})
}
func (n *recordingNotifier) TranslationStarted(context.Context, string, int, int64, string, string, string, int) {
n.events = append(n.events, notifierEvent{kind: "started"})
}
func (n *recordingNotifier) TranslationCues(_ context.Context, _ string, _ int, _ int64, _ string,
cues []playback.StreamCue, done, total int) {
n.events = append(n.events, notifierEvent{kind: "cues", cues: cues, done: done, total: total})
if len(cues) > 0 {
appendTrace(n.trace, "stream:"+cues[0].Text)
}
}
func (n *recordingNotifier) TranslationCompleted(context.Context, string, int, int64, string, int, string, string) {
n.events = append(n.events, notifierEvent{kind: "completed"})
}
func (n *recordingNotifier) TranslationFailed(_ context.Context, _ string, _ int, _ int64, _, message string) {
n.events = append(n.events, notifierEvent{kind: "failed", message: message})
}
func TestRunTranscribeTranslateStreamingInterleavesTranslationPerChunk(t *testing.T) {
var trace []string
transcriber := &chunkedTranscriber{
detected: "en",
trace: &trace,
chunks: [][]SubtitleCue{
{testCue(60, "second")},
{},
{testCue(0, "first")},
},
}
translator := &recordingTranslator{trace: &trace}
store := &recordingSubtitleStore{}
notifier := &recordingNotifier{trace: &trace}
repo := &recordingRepo{}
svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30})
svc.runTranscribe(context.Background(), &Job{
ID: 7,
MediaFileID: 10,
Kind: JobKindTranscribeTranslate,
SourceIndex: -1,
TargetLanguage: "es",
SessionID: "session-1",
StartPosition: 60,
})
if !transcriber.lastReq.Incremental {
t.Fatalf("streaming request did not enable incremental extraction")
}
if transcriber.lastReq.ChunkSeconds != 30 {
t.Fatalf("chunk override = %d, want 30", transcriber.lastReq.ChunkSeconds)
}
if got := len(translator.calls); got != 2 {
t.Fatalf("Translate calls = %d, want one per non-empty chunk", got)
}
if strings.Join(translator.calls[0].Cues[0].Lines, " ") != "second" {
t.Errorf("first translated chunk = %q, want playhead chunk", translator.calls[0].Cues[0].Lines)
}
if streamIdx, lastChunkIdx := traceIndex(trace, "stream:es:second"), traceIndex(trace, "transcribe:2"); streamIdx < 0 || lastChunkIdx < 0 || streamIdx > lastChunkIdx {
t.Fatalf("first translated stream did not precede final ASR chunk: %v", trace)
}
cueEvents := cueNotifierEvents(notifier.events)
if got := len(cueEvents); got != 2 {
t.Fatalf("TranslationCues events = %d, want 2", got)
}
if cueEvents[0].done != 1 || cueEvents[0].total != 6 || cueEvents[1].done != 3 || cueEvents[1].total != 6 {
t.Fatalf("chunk progress = (%d/%d, %d/%d), want 1/6 then 3/6",
cueEvents[0].done, cueEvents[0].total, cueEvents[1].done, cueEvents[1].total)
}
if len(store.stored) != 2 {
t.Fatalf("stored subtitles = %d, want transcript and translation", len(store.stored))
}
if store.stored[0].Provider != providerTranscribed || store.stored[1].Provider != providerTranslated {
t.Fatalf("stored providers = %q, %q", store.stored[0].Provider, store.stored[1].Provider)
}
translated, err := ParseCues(store.stored[1].Data)
if err != nil {
t.Fatalf("parse stored translation: %v", err)
}
if len(translated) != 2 || strings.Join(translated[0].Lines, " ") != "es:first" ||
strings.Join(translated[1].Lines, " ") != "es:second" {
t.Fatalf("stored translation not sorted/complete: %#v", translated)
}
if got := fmt.Sprint(repo.completed); got != "[2]" {
t.Fatalf("completed subtitle IDs = %s, want [2]", got)
}
}
func TestRunTranscribeTranslateStreamingFailureStoresTranscript(t *testing.T) {
var trace []string
transcriber := &chunkedTranscriber{
detected: "en",
trace: &trace,
chunks: [][]SubtitleCue{
{testCue(0, "one")},
{testCue(30, "two")},
{testCue(60, "three")},
},
}
translator := &recordingTranslator{errOnCall: 2, trace: &trace}
store := &recordingSubtitleStore{}
notifier := &recordingNotifier{trace: &trace}
repo := &recordingRepo{}
svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30})
svc.runTranscribe(context.Background(), &Job{
ID: 8,
MediaFileID: 10,
Kind: JobKindTranscribeTranslate,
SourceIndex: -1,
TargetLanguage: "es",
SessionID: "session-1",
})
if got := len(translator.calls); got != 2 {
t.Fatalf("Translate calls = %d, want calls stop after failure", got)
}
if traceIndex(trace, "transcribe:2") < 0 {
t.Fatalf("ASR did not continue through final chunk after translate failure: %v", trace)
}
if len(store.stored) != 1 || store.stored[0].Provider != providerTranscribed {
t.Fatalf("stored subtitles = %#v, want transcript only", store.stored)
}
if len(repo.failures) != 1 || repo.failures[0].status != JobStatusFailed {
t.Fatalf("failures = %#v, want one failed job", repo.failures)
}
if got := countNotifierKind(notifier.events, "failed"); got != 1 {
t.Fatalf("TranslationFailed events = %d, want 1", got)
}
if len(repo.completed) != 0 {
t.Fatalf("completed despite translation failure: %v", repo.completed)
}
}
func TestRunTranscribeTranslateNonStreamingKeepsWholeFileTranslation(t *testing.T) {
transcriber := &chunkedTranscriber{
detected: "en",
chunks: [][]SubtitleCue{
{testCue(30, "later")},
{testCue(0, "earlier")},
},
}
translator := &recordingTranslator{}
store := &recordingSubtitleStore{}
notifier := &recordingNotifier{}
repo := &recordingRepo{}
svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30})
svc.runTranscribe(context.Background(), &Job{
ID: 9,
MediaFileID: 10,
Kind: JobKindTranscribeTranslate,
SourceIndex: -1,
TargetLanguage: "es",
})
if transcriber.lastReq.Incremental {
t.Fatalf("background request unexpectedly enabled incremental extraction")
}
if transcriber.lastReq.ChunkSeconds != 0 {
t.Fatalf("background chunk override = %d, want 0", transcriber.lastReq.ChunkSeconds)
}
if got := len(translator.calls); got != 1 {
t.Fatalf("Translate calls = %d, want one whole-file call", got)
}
if got := len(translator.calls[0].Cues); got != 2 {
t.Fatalf("whole-file call cues = %d, want 2", got)
}
if got := countNotifierKind(notifier.events, "cues"); got != 0 {
t.Fatalf("non-streaming TranslationCues events = %d, want 0", got)
}
if len(store.stored) != 2 || store.stored[1].Provider != providerTranslated {
t.Fatalf("stored subtitles = %#v, want transcript and translated track", store.stored)
}
}
func TestRunTranscribeTranslateStreamingUsesDetectedLanguageForLiveChunks(t *testing.T) {
transcriber := &chunkedTranscriber{
detected: "ja",
chunks: [][]SubtitleCue{
{testCue(0, "konnichiwa")},
},
}
translator := &recordingTranslator{}
store := &recordingSubtitleStore{}
notifier := &recordingNotifier{}
repo := &recordingRepo{}
file := &models.MediaFile{
ID: 10,
FilePath: "/media/movie.mkv",
Duration: 180,
AudioTracks: []models.AudioTrack{{Default: true}},
}
svc := newRunTranscribeServiceWithFile(repo, translator, transcriber, store, notifier,
Config{LiveASRChunkSeconds: 30}, file)
svc.runTranscribe(context.Background(), &Job{
ID: 10,
MediaFileID: 10,
Kind: JobKindTranscribeTranslate,
SourceIndex: -1,
TargetLanguage: "es",
SessionID: "session-1",
})
if got := len(translator.calls); got != 1 {
t.Fatalf("Translate calls = %d, want one live chunk call", got)
}
if got := translator.calls[0].SourceLanguage; got != "ja" {
t.Fatalf("live chunk source language = %q, want detected ja", got)
}
}
func newRunTranscribeService(
repo *recordingRepo,
translator Translator,
transcriber Transcriber,
store *recordingSubtitleStore,
notifier Notifier,
cfg Config,
) *Service {
file := &models.MediaFile{
ID: 10,
FilePath: "/media/movie.mkv",
Duration: 180,
AudioTracks: []models.AudioTrack{{Language: "eng", Default: true}},
}
return newRunTranscribeServiceWithFile(repo, translator, transcriber, store, notifier, cfg, file)
}
func newRunTranscribeServiceWithFile(
repo *recordingRepo,
translator Translator,
transcriber Transcriber,
store *recordingSubtitleStore,
notifier Notifier,
cfg Config,
file *models.MediaFile,
) *Service {
return NewService(context.Background(), cfg, repo, translator, transcriber, store, nil,
runTranscribeFileResolver{file: file}, notifier, "", nil, nil)
}
func testCue(startSeconds float64, text string) SubtitleCue {
start := time.Duration(startSeconds * float64(time.Second))
return SubtitleCue{
Start: start,
End: start + 2*time.Second,
Lines: []string{text},
}
}
func cloneTestCues(cues []SubtitleCue) []SubtitleCue {
out := make([]SubtitleCue, len(cues))
for i, cue := range cues {
out[i] = cue
out[i].Lines = append([]string(nil), cue.Lines...)
}
return out
}
func appendTrace(trace *[]string, value string) {
if trace != nil {
*trace = append(*trace, value)
}
}
func traceIndex(trace []string, value string) int {
for i, got := range trace {
if got == value {
return i
}
}
return -1
}
func cueNotifierEvents(events []notifierEvent) []notifierEvent {
var out []notifierEvent
for _, event := range events {
if event.kind == "cues" {
out = append(out, event)
}
}
return out
}
func countNotifierKind(events []notifierEvent, kind string) int {
count := 0
for _, event := range events {
if event.kind == kind {
count++
}
}
return count
}
+201 -40
View File
@@ -2,7 +2,9 @@ package ai
import (
"context"
"errors"
"fmt"
"math"
"os"
"path/filepath"
"strings"
@@ -21,9 +23,11 @@ const (
// 10 minutes of 16 kHz mono WAV is ~19 MB — under typical 25 MB API
// limits. Shorter chunks bound Whisper's within-chunk timestamp drift at
// the cost of more requests and more boundary word-clips; the operator
// tunes this via subtitle_ai.asr_chunk_seconds.
// tunes this via subtitle_ai.asr_chunk_seconds. Live jobs may use smaller
// chunks through subtitle_ai.live_asr_chunk_seconds to reduce time to first
// cue; the 15s floor bounds request count and boundary word-clips.
defaultASRChunkSeconds = 600
minASRChunkSeconds = 60
minASRChunkSeconds = 15
// Per-chunk request timeout: 3× realtime accommodates local Whisper
// servers on modest hardware.
asrChunkTimeoutFactor = 3
@@ -46,6 +50,9 @@ type TranscribeJobRequest struct {
AudioTrackIndex int // resolved 0-based audio stream index
LanguageHint string // ISO 639-1; "" lets the model detect
StartPosition float64 // seconds; chunks are processed playhead-first
DurationSeconds float64 // optional media duration for incremental progress
ChunkSeconds int // optional per-job override; 0 uses configured default
Incremental bool // stream extraction chunks instead of extracting whole file first
}
// Transcriber converts an audio track into subtitle cues. The built-in
@@ -55,9 +62,14 @@ type TranscribeJobRequest struct {
// callers can report progress and stream cues live.
type Transcriber interface {
Transcribe(ctx context.Context, req TranscribeJobRequest,
onChunk func(cues []SubtitleCue, done, total int)) ([]SubtitleCue, string, error)
onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error)
}
// TranscribeChunkCallback receives each chunk's cues, the detected language for
// that chunk when the ASR endpoint reports one, and chunk progress. A total of
// 0 means the total is unknown/indeterminate.
type TranscribeChunkCallback func(cues []SubtitleCue, language string, done, total int)
// audioTranscriber is the slice of llm.Client the transcriber needs.
type audioTranscriber interface {
Transcribe(ctx context.Context, req llm.TranscribeRequest) (*llm.Transcription, error)
@@ -76,18 +88,20 @@ type WhisperTranscriber struct {
ffmpegPath atomic.Pointer[string]
chunkSeconds atomic.Int32
// extract and probeOffset are playback helpers, injectable for tests.
extract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, chunkSeconds int) ([]playback.AudioChunk, error)
probeOffset func(ctx context.Context, filePath string, audioTrackIndex int, ffmpegPath string) float64
extract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, chunkSeconds int) ([]playback.AudioChunk, error)
incrementalExtract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, startSec float64, chunkSeconds int, onSegment func(playback.AudioChunk) error) error
probeOffset func(ctx context.Context, filePath string, audioTrackIndex int, ffmpegPath string) float64
}
// NewWhisperTranscriber builds a transcriber backed by the shared AI client.
// chunkSeconds outside [minASRChunkSeconds, defaultASRChunkSeconds] falls
// back to the default (longer chunks would exceed upload limits).
// chunkSeconds outside [minASRChunkSeconds, defaultASRChunkSeconds] clamps
// to the nearest bound (longer chunks would exceed upload limits).
func NewWhisperTranscriber(client *llm.Client, ffmpegPath string, chunkSeconds int) *WhisperTranscriber {
t := &WhisperTranscriber{
client: client,
extract: playback.ExtractAudioChunks,
probeOffset: playback.ProbeAudioStartOffset,
client: client,
extract: playback.ExtractAudioChunks,
incrementalExtract: playback.ExtractAudioChunksFrom,
probeOffset: playback.ProbeAudioStartOffset,
}
t.SetExtraction(ffmpegPath, chunkSeconds)
return t
@@ -95,13 +109,10 @@ func NewWhisperTranscriber(client *llm.Client, ffmpegPath string, chunkSeconds i
// SetExtraction updates the ffmpeg path and ASR chunk duration used by jobs
// started afterwards. Safe for concurrent use; out-of-range chunk durations
// fall back to the default, matching construction.
// clamp to the nearest supported bound, matching per-request overrides.
func (t *WhisperTranscriber) SetExtraction(ffmpegPath string, chunkSeconds int) {
if chunkSeconds < minASRChunkSeconds || chunkSeconds > defaultASRChunkSeconds {
chunkSeconds = defaultASRChunkSeconds
}
t.ffmpegPath.Store(&ffmpegPath)
t.chunkSeconds.Store(int32(chunkSeconds))
t.chunkSeconds.Store(int32(clampASRChunkSeconds(chunkSeconds)))
}
// Transcribe implements Transcriber. The returned cues are NOT sorted (they
@@ -109,7 +120,7 @@ func (t *WhisperTranscriber) SetExtraction(ffmpegPath string, chunkSeconds int)
// endpoint's report for the first processed chunk, normalized to an ISO code
// where possible.
func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRequest,
onChunk func(cues []SubtitleCue, done, total int)) ([]SubtitleCue, string, error) {
onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error) {
dir, err := os.MkdirTemp("", "silo-asr-*")
if err != nil {
return nil, "", fmt.Errorf("create ASR temp dir: %w", err)
@@ -120,50 +131,41 @@ func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRe
// even if the config reloads mid-job.
ffmpegPath := *t.ffmpegPath.Load()
chunkSeconds := int(t.chunkSeconds.Load())
chunks, err := t.extract(ctx, req.FilePath, req.AudioTrackIndex, dir, ffmpegPath, chunkSeconds)
if err != nil {
return nil, "", err
if req.ChunkSeconds > 0 {
chunkSeconds = req.ChunkSeconds
}
chunkSeconds = clampASRChunkSeconds(chunkSeconds)
// Audio streams can start after the container's timeline origin (TS
// remuxes especially); Whisper times are relative to the first audio
// sample, so the delta is a constant sync error unless added back.
startOffset := t.probeOffset(ctx, req.FilePath, req.AudioTrackIndex, ffmpegPath)
if req.Incremental {
return t.transcribeIncremental(ctx, req, dir, ffmpegPath, chunkSeconds, startOffset, onChunk)
}
chunks, err := t.extract(ctx, req.FilePath, req.AudioTrackIndex, dir, ffmpegPath, chunkSeconds)
if err != nil {
return nil, "", err
}
order := chunkOrderForPosition(chunks, req.StartPosition)
timeout := time.Duration(chunkSeconds*asrChunkTimeoutFactor) * time.Second
var all []SubtitleCue
detected := ""
for done, idx := range order {
if err := ctx.Err(); err != nil {
return nil, "", err
}
data, err := os.ReadFile(chunks[idx].Path)
if err != nil {
return nil, "", fmt.Errorf("read audio chunk: %w", err)
}
tr, err := t.client.Transcribe(ctx, llm.TranscribeRequest{
Filename: filepath.Base(chunks[idx].Path),
Audio: data,
Language: req.LanguageHint,
Timeout: timeout,
})
cues, lang, err := t.transcribeChunk(ctx, chunks[idx], req.LanguageHint, timeout, startOffset)
if err != nil {
return nil, "", fmt.Errorf("transcribe chunk %d/%d: %w", idx+1, len(chunks), err)
}
// Each chunk is read exactly once; deleting it as we go caps disk usage
// at one extraction rather than extraction + retranscription leftovers.
_ = os.Remove(chunks[idx].Path)
if detected == "" {
detected = normalizeDetectedLanguage(tr.Language)
detected = lang
}
cues := cuesFromSegments(tr.Segments, chunks[idx].Start+startOffset)
all = append(all, cues...)
if onChunk != nil {
onChunk(cues, done+1, len(order))
onChunk(cues, lang, done+1, len(order))
}
}
@@ -173,6 +175,165 @@ func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRe
return all, detected, nil
}
var errStopIncrementalPass = errors.New("stop incremental ASR pass")
func (t *WhisperTranscriber) transcribeIncremental(
ctx context.Context,
req TranscribeJobRequest,
dir, ffmpegPath string,
chunkSeconds int,
startOffset float64,
onChunk TranscribeChunkCallback,
) ([]SubtitleCue, string, error) {
timeout := time.Duration(chunkSeconds*asrChunkTimeoutFactor) * time.Second
pivot := incrementalPivotStart(req.StartPosition, req.DurationSeconds, chunkSeconds)
total := estimatedChunkTotal(req.DurationSeconds, chunkSeconds)
var all []SubtitleCue
detected := ""
done := 0
process := func(chunk playback.AudioChunk, chunkOffset float64) error {
cues, lang, err := t.transcribeChunk(ctx, chunk, req.LanguageHint, timeout, chunkOffset)
if err != nil {
return fmt.Errorf("transcribe chunk at %.3fs: %w", chunk.Start, err)
}
if detected == "" {
detected = lang
}
all = append(all, cues...)
done++
if onChunk != nil {
onChunk(cues, lang, done, progressTotal(done, total))
}
return nil
}
firstPassDone := done
firstPassOffset := audioOffsetForIncrementalPass(pivot, startOffset)
if err := t.runIncrementalPass(ctx, req, dir, ffmpegPath, pivot, chunkSeconds, func(chunk playback.AudioChunk) error {
return process(chunk, firstPassOffset)
}); err != nil {
if ctx.Err() != nil || pivot == 0 || done != firstPassDone {
return nil, "", err
}
}
if pivot > 0 {
wrap := func(chunk playback.AudioChunk) error {
if chunk.Start >= pivot {
return errStopIncrementalPass
}
return process(chunk, startOffset)
}
if err := t.runIncrementalPass(ctx, req, dir, ffmpegPath, 0, chunkSeconds, wrap); err != nil {
if !errors.Is(err, errStopIncrementalPass) {
return nil, "", err
}
}
}
if len(all) == 0 {
return nil, detected, fmt.Errorf("no speech recognized in the audio track")
}
return all, detected, nil
}
func (t *WhisperTranscriber) runIncrementalPass(
ctx context.Context,
req TranscribeJobRequest,
dir, ffmpegPath string,
startSec float64,
chunkSeconds int,
onSegment func(playback.AudioChunk) error,
) error {
passDir := filepath.Join(dir, fmt.Sprintf("from-%.3f", startSec))
if err := os.MkdirAll(passDir, 0o755); err != nil {
return fmt.Errorf("create ASR incremental temp dir: %w", err)
}
return t.incrementalExtract(ctx, req.FilePath, req.AudioTrackIndex, passDir, ffmpegPath, startSec, chunkSeconds, onSegment)
}
func (t *WhisperTranscriber) transcribeChunk(
ctx context.Context,
chunk playback.AudioChunk,
languageHint string,
timeout time.Duration,
startOffset float64,
) ([]SubtitleCue, string, error) {
if err := ctx.Err(); err != nil {
return nil, "", err
}
data, err := os.ReadFile(chunk.Path)
if err != nil {
return nil, "", fmt.Errorf("read audio chunk: %w", err)
}
tr, err := t.client.Transcribe(ctx, llm.TranscribeRequest{
Filename: filepath.Base(chunk.Path),
Audio: data,
Language: languageHint,
Timeout: timeout,
})
if err != nil {
return nil, "", err
}
// Each chunk is read exactly once; deleting it as we go caps disk usage
// at one extraction rather than extraction + retranscription leftovers.
_ = os.Remove(chunk.Path)
return cuesFromSegments(tr.Segments, chunk.Start+startOffset), normalizeDetectedLanguage(tr.Language), nil
}
func clampASRChunkSeconds(chunkSeconds int) int {
switch {
case chunkSeconds < minASRChunkSeconds:
return minASRChunkSeconds
case chunkSeconds > defaultASRChunkSeconds:
return defaultASRChunkSeconds
default:
return chunkSeconds
}
}
func incrementalPivotStart(startSeconds, durationSeconds float64, chunkSeconds int) float64 {
if startSeconds <= 0 || chunkSeconds <= 0 {
return 0
}
if durationSeconds > 0 && startSeconds >= durationSeconds {
startSeconds = math.Max(0, durationSeconds-0.001)
}
pivot := math.Floor(startSeconds/float64(chunkSeconds)) * float64(chunkSeconds)
if pivot < 0 {
return 0
}
return pivot
}
func estimatedChunkTotal(durationSeconds float64, chunkSeconds int) int {
if durationSeconds <= 0 || chunkSeconds <= 0 {
return 0
}
return int(math.Ceil(durationSeconds / float64(chunkSeconds)))
}
func progressTotal(done, total int) int {
if total <= 0 {
return 0
}
if total < done {
return done
}
return total
}
func audioOffsetForIncrementalPass(passStartSec, audioStartOffset float64) float64 {
if passStartSec <= 0 {
return audioStartOffset
}
if audioStartOffset <= passStartSec {
return 0
}
return audioStartOffset - passStartSec
}
// chunkOrderForPosition orders chunk indexes so the chunk containing
// startSeconds is processed first, then forward, then wrapping to the start —
// the viewer's current region fills first, mirroring translation's
+280 -1
View File
@@ -117,7 +117,7 @@ func TestTranscribeProcessesChunksPlayheadFirst(t *testing.T) {
var chunkOrder []string
cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv", StartPosition: 1300, // inside chunk 2
}, func(chunk []SubtitleCue, done, total int) {
}, func(chunk []SubtitleCue, _ string, done, total int) {
if total != 3 {
t.Errorf("total = %d, want 3", total)
}
@@ -157,6 +157,67 @@ func TestTranscribePassesHintAndChunkSizedTimeout(t *testing.T) {
}
}
func TestSetExtractionClampsChunkSecondsToNearestBound(t *testing.T) {
tr := &WhisperTranscriber{}
tr.SetExtraction("", 1)
if got := int(tr.chunkSeconds.Load()); got != minASRChunkSeconds {
t.Fatalf("low clamp = %d, want %d", got, minASRChunkSeconds)
}
tr.SetExtraction("", defaultASRChunkSeconds+1)
if got := int(tr.chunkSeconds.Load()); got != defaultASRChunkSeconds {
t.Fatalf("high clamp = %d, want %d", got, defaultASRChunkSeconds)
}
tr.SetExtraction("", 45)
if got := int(tr.chunkSeconds.Load()); got != 45 {
t.Fatalf("in-range chunk seconds = %d, want 45", got)
}
}
func TestTranscribeUsesRequestChunkSecondsOverride(t *testing.T) {
var dir string
var extractedChunkSeconds int
client := &fakeASRClient{
language: "en",
perChunk: map[string][]llm.TranscriptionSegment{
"chunk00000.wav": {{Start: 0, End: 1, Text: "hi"}},
},
}
tr := &WhisperTranscriber{
client: client,
extract: func(_ context.Context, _ string, _ int, d, _ string, chunkSeconds int) ([]playback.AudioChunk, error) {
dir = d
extractedChunkSeconds = chunkSeconds
p := filepath.Join(d, "chunk00000.wav")
if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil {
return nil, err
}
return []playback.AudioChunk{{Path: p, Start: 0}}, nil
},
probeOffset: func(context.Context, string, int, string) float64 { return 0 },
}
tr.SetExtraction("", 600)
_, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv",
ChunkSeconds: 10,
}, nil)
if err != nil {
t.Fatalf("Transcribe: %v", err)
}
if extractedChunkSeconds != minASRChunkSeconds {
t.Fatalf("extract chunk seconds = %d, want clamped override %d", extractedChunkSeconds, minASRChunkSeconds)
}
if want := time.Duration(minASRChunkSeconds*asrChunkTimeoutFactor) * time.Second; client.requests[0].Timeout != want {
t.Fatalf("timeout = %v, want %v", client.requests[0].Timeout, want)
}
if _, statErr := os.Stat(dir); !os.IsNotExist(statErr) {
t.Errorf("temp dir %s not cleaned up", dir)
}
}
func TestTranscribeAllSilentChunksFailsClearly(t *testing.T) {
var dir string
client := &fakeASRClient{language: "en", perChunk: map[string][]llm.TranscriptionSegment{}}
@@ -326,6 +387,215 @@ func TestChunkOrderForPosition(t *testing.T) {
}
}
func TestTranscribeIncrementalStreamsPlayheadFirstAcrossTwoPasses(t *testing.T) {
var trace []string
var starts []float64
client := &fakeASRClient{
language: "en",
perChunk: map[string][]llm.TranscriptionSegment{
"inc1200.wav": {{Start: 0, End: 1, Text: "pivot"}},
"inc1800.wav": {{Start: 0, End: 1, Text: "tail"}},
"inc0.wav": {{Start: 0, End: 1, Text: "head"}},
"inc600.wav": {{Start: 0, End: 1, Text: "middle"}},
},
}
tr := &WhisperTranscriber{
client: client,
incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int,
onSegment func(playback.AudioChunk) error) error {
starts = append(starts, startSec)
segmentStarts := []float64{1200, 1800}
if startSec == 0 {
segmentStarts = []float64{0, 600, 1200}
}
for _, segStart := range segmentStarts {
name := fmt.Sprintf("inc%.0f.wav", segStart)
p := filepath.Join(dir, name)
if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil {
return err
}
trace = append(trace, fmt.Sprintf("segment:%.0f", segStart))
if err := onSegment(playback.AudioChunk{Path: p, Start: segStart}); err != nil {
return err
}
trace = append(trace, fmt.Sprintf("after:%.0f", segStart))
}
return nil
},
probeOffset: func(context.Context, string, int, string) float64 { return 0 },
}
tr.SetExtraction("", 600)
var progress []string
cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv",
StartPosition: 1300,
DurationSeconds: 2400,
Incremental: true,
}, func(chunk []SubtitleCue, _ string, done, total int) {
progress = append(progress, fmt.Sprintf("%d/%d", done, total))
trace = append(trace, "chunk:"+strings.Join(chunk[0].Lines, " "))
})
if err != nil {
t.Fatalf("Transcribe: %v", err)
}
if got := fmt.Sprint(starts); got != "[1200 0]" {
t.Fatalf("incremental pass starts = %s, want [1200 0]", got)
}
var requestOrder []string
for _, req := range client.requests {
requestOrder = append(requestOrder, req.Filename)
}
if got := strings.Join(requestOrder, ","); got != "inc1200.wav,inc1800.wav,inc0.wav,inc600.wav" {
t.Fatalf("request order = %s", got)
}
if got := strings.Join(progress, ","); got != "1/4,2/4,3/4,4/4" {
t.Fatalf("progress = %s, want chunk totals", got)
}
if chunkIdx, nextSegmentIdx := traceIndexString(trace, "chunk:pivot"), traceIndexString(trace, "segment:1800"); chunkIdx < 0 || nextSegmentIdx < 0 || chunkIdx > nextSegmentIdx {
t.Fatalf("first onChunk did not fire before the next extracted segment: %v", trace)
}
if len(cues) != 4 {
t.Fatalf("returned cues = %d, want complete transcript", len(cues))
}
}
func TestTranscribeIncrementalFallsBackToStartWhenSeekedPassProducesNoAudio(t *testing.T) {
client := &fakeASRClient{
language: "en",
perChunk: map[string][]llm.TranscriptionSegment{
"head.wav": {{Start: 1, End: 2, Text: "head"}},
},
}
tr := &WhisperTranscriber{
client: client,
incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int,
onSegment func(playback.AudioChunk) error) error {
if startSec > 0 {
return fmt.Errorf("ffmpeg produced no audio chunks for track 0")
}
p := filepath.Join(dir, "head.wav")
if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil {
return err
}
return onSegment(playback.AudioChunk{Path: p, Start: 0})
},
probeOffset: func(context.Context, string, int, string) float64 { return 0 },
}
tr.SetExtraction("", 600)
cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv",
StartPosition: 1300,
DurationSeconds: 1800,
Incremental: true,
}, nil)
if err != nil {
t.Fatalf("Transcribe: %v", err)
}
if len(cues) != 1 || strings.Join(cues[0].Lines, " ") != "head" {
t.Fatalf("fallback cues = %#v, want head cue from start pass", cues)
}
if len(client.requests) != 1 || client.requests[0].Filename != "head.wav" {
t.Fatalf("ASR requests = %#v, want only fallback chunk", client.requests)
}
}
func TestTranscribeIncrementalDoesNotDoubleApplyAudioOffsetAfterSeek(t *testing.T) {
client := &fakeASRClient{
language: "en",
perChunk: map[string][]llm.TranscriptionSegment{
"seek.wav": {{Start: 1, End: 2, Text: "seek"}},
"head.wav": {{Start: 1, End: 2, Text: "head"}},
},
}
tr := &WhisperTranscriber{
client: client,
incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int,
onSegment func(playback.AudioChunk) error) error {
name := "seek.wav"
chunkStart := 600.0
if startSec == 0 {
name = "head.wav"
chunkStart = 0
}
p := filepath.Join(dir, name)
if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil {
return err
}
if err := onSegment(playback.AudioChunk{Path: p, Start: chunkStart}); err != nil {
return err
}
if startSec == 0 {
stop := filepath.Join(dir, "seek.wav")
if err := os.WriteFile(stop, []byte("RIFF"), 0o644); err != nil {
return err
}
return onSegment(playback.AudioChunk{Path: stop, Start: 600})
}
return nil
},
probeOffset: func(context.Context, string, int, string) float64 { return 1.25 },
}
tr.SetExtraction("", 600)
cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv",
StartPosition: 600,
DurationSeconds: 1200,
Incremental: true,
}, nil)
if err != nil {
t.Fatalf("Transcribe: %v", err)
}
if len(cues) != 2 {
t.Fatalf("cues = %d, want seek and head", len(cues))
}
if want := 601 * time.Second; cues[0].Start != want {
t.Fatalf("seek cue start = %v, want %v without re-adding audio offset", cues[0].Start, want)
}
if want := time.Duration(2.25 * float64(time.Second)); cues[1].Start != want {
t.Fatalf("head cue start = %v, want %v with audio offset", cues[1].Start, want)
}
}
func TestTranscribeIncrementalKeepsProgressIndeterminateWhenDurationUnknown(t *testing.T) {
client := &fakeASRClient{
language: "en",
perChunk: map[string][]llm.TranscriptionSegment{
"head.wav": {{Start: 0, End: 1, Text: "head"}},
},
}
tr := &WhisperTranscriber{
client: client,
incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, _ float64, _ int,
onSegment func(playback.AudioChunk) error) error {
p := filepath.Join(dir, "head.wav")
if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil {
return err
}
return onSegment(playback.AudioChunk{Path: p, Start: 0})
},
probeOffset: func(context.Context, string, int, string) float64 { return 0 },
}
tr.SetExtraction("", 600)
var gotDone, gotTotal int
_, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{
FilePath: "/x.mkv",
Incremental: true,
}, func(_ []SubtitleCue, _ string, done, total int) {
gotDone = done
gotTotal = total
})
if err != nil {
t.Fatalf("Transcribe: %v", err)
}
if gotDone != 1 || gotTotal != 0 {
t.Fatalf("progress = %d/%d, want 1/0 for indeterminate total", gotDone, gotTotal)
}
}
// Cue timing must use the muxer-reported chunk start (not index*chunkSeconds)
// plus the probed audio start offset, or sync drifts on long files and
// delayed-audio containers.
@@ -358,6 +628,15 @@ func TestTranscribeUsesExactChunkStartsAndAudioOffset(t *testing.T) {
}
}
func traceIndexString(values []string, target string) int {
for i, value := range values {
if value == target {
return i
}
}
return -1
}
func TestNormalizeDetectedLanguage(t *testing.T) {
cases := map[string]string{
"english": "en",