Download S3 audio during prepare
This commit is contained in:
10
README.md
10
README.md
@@ -49,7 +49,7 @@ YAML decoding is strict (`KnownFields(true)`), so unknown fields fail fast.
|
|||||||
|
|
||||||
## Storage And Archive Foundations
|
## Storage And Archive Foundations
|
||||||
|
|
||||||
Narratio now includes configuration and path-model foundations for future S3 audio input and archive support.
|
Narratio now includes configuration and path-model foundations for archive support, plus implemented prepare-stage S3 audio input.
|
||||||
|
|
||||||
Implemented foundations:
|
Implemented foundations:
|
||||||
|
|
||||||
@@ -58,7 +58,7 @@ Implemented foundations:
|
|||||||
- `pipeline.archive` config shape (`enabled`, `upload_run`, `promote_artifacts`)
|
- `pipeline.archive` config shape (`enabled`, `upload_run`, `promote_artifacts`)
|
||||||
- promotion-rule validation (`from`/`to` required, relative-only paths, traversal rejected)
|
- promotion-rule validation (`from`/`to` required, relative-only paths, traversal rejected)
|
||||||
- `session.campaign` requirement for campaign-aware path construction
|
- `session.campaign` requirement for campaign-aware path construction
|
||||||
- optional `session.inputs.audio_s3.prefix` modeling (not executed yet)
|
- optional `session.inputs.audio_s3.prefix` modeling and prepare-stage S3 audio download
|
||||||
- run ID generation and S3/local path helper foundations
|
- run ID generation and S3/local path helper foundations
|
||||||
- manifest run/path identity fields
|
- manifest run/path identity fields
|
||||||
|
|
||||||
@@ -79,10 +79,11 @@ Current boundaries:
|
|||||||
- `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive
|
- `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive
|
||||||
- real S3-compatible backend now exists in the storage adapter package
|
- real S3-compatible backend now exists in the storage adapter package
|
||||||
- storage backend tests use fake storage and do not require live S3
|
- storage backend tests use fake storage and do not require live S3
|
||||||
- no prepare-stage S3 list/download behavior yet
|
|
||||||
- no archive-stage S3 upload/promotion behavior yet
|
- no archive-stage S3 upload/promotion behavior yet
|
||||||
- no `current/manifest.json` or `current/run_id.txt` uploads yet
|
- no `current/manifest.json` or `current/run_id.txt` uploads yet
|
||||||
|
|
||||||
|
S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md).
|
||||||
|
|
||||||
## Remote Storage Backend
|
## Remote Storage Backend
|
||||||
|
|
||||||
Narratio includes an object-store backend layer for future prepare/archive work:
|
Narratio includes an object-store backend layer for future prepare/archive work:
|
||||||
@@ -104,7 +105,8 @@ Key invariant:
|
|||||||
|
|
||||||
Current boundary:
|
Current boundary:
|
||||||
|
|
||||||
- this backend layer is implemented but not yet wired into prepare-stage audio retrieval or archive-stage uploads
|
- `prepare` uses `List` + `Download` through the backend when `session.inputs.audio_s3` is configured
|
||||||
|
- archive-stage uploads/promotion are still not implemented
|
||||||
|
|
||||||
## Canonical Stage Order
|
## Canonical Stage Order
|
||||||
|
|
||||||
|
|||||||
@@ -41,7 +41,6 @@ Implemented:
|
|||||||
Still placeholder/future:
|
Still placeholder/future:
|
||||||
|
|
||||||
- `archive` stage behavior
|
- `archive` stage behavior
|
||||||
- prepare-stage S3 audio download behavior
|
|
||||||
- archive-stage S3 upload/promotion behavior
|
- archive-stage S3 upload/promotion behavior
|
||||||
- `notify` stage behavior
|
- `notify` stage behavior
|
||||||
- additional Scriptorium artifact types beyond `session_recap`
|
- additional Scriptorium artifact types beyond `session_recap`
|
||||||
@@ -152,9 +151,9 @@ Session input foundations:
|
|||||||
|
|
||||||
- `session.campaign` is required
|
- `session.campaign` is required
|
||||||
- local audio remains supported through `session.inputs.audio_dir` or `session.inputs.audio_files`
|
- local audio remains supported through `session.inputs.audio_dir` or `session.inputs.audio_files`
|
||||||
- optional S3 audio input shape is modeled as `session.inputs.audio_s3.prefix`
|
- optional S3 audio input shape is `session.inputs.audio_s3.prefix`
|
||||||
- `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive
|
- `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive
|
||||||
- S3 input execution (object listing/downloading) is not implemented yet
|
- when `audio_s3` is configured, `prepare` lists and downloads `.flac` objects through the object-store backend
|
||||||
|
|
||||||
Cross-config validation scope:
|
Cross-config validation scope:
|
||||||
|
|
||||||
@@ -168,7 +167,21 @@ Remote object-store backend scope:
|
|||||||
- S3 key/session path semantics remain outside the backend, with this invariant:
|
- S3 key/session path semantics remain outside the backend, with this invariant:
|
||||||
- callers pass full bucket-relative object keys
|
- callers pass full bucket-relative object keys
|
||||||
- backend methods do not prepend `root_prefix` or infer campaign/session/run paths
|
- backend methods do not prepend `root_prefix` or infer campaign/session/run paths
|
||||||
- the backend layer is available for future prepare/archive usage, but no stage currently invokes `List`/`Download`/`Upload`/`Exists`
|
- `prepare` now uses object-store `List` and `Download` for S3 audio input
|
||||||
|
- archive upload/promotion behavior using `Upload`/`Exists` remains future work
|
||||||
|
|
||||||
|
Prepare S3 audio behavior (implemented):
|
||||||
|
|
||||||
|
- compute session prefix as `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/`
|
||||||
|
- resolve `session.inputs.audio_s3.prefix` under that session prefix
|
||||||
|
- list objects under the computed audio prefix and filter `.flac` keys
|
||||||
|
- fail clearly when no `.flac` objects are found
|
||||||
|
- download selected objects to spool audio path:
|
||||||
|
- `{spool.root}/{campaign}/{session_id}/{run_id}/audio/`
|
||||||
|
- materialize audio files into workdir audio path:
|
||||||
|
- `{workspace.root}/work/{campaign}/{session_id}/{run_id}/audio/`
|
||||||
|
- record S3 provenance in manifest input records (bucket/key/metadata/local paths/checksum)
|
||||||
|
- no AWS SDK types are used in stage code; storage implementation details stay in storage adapter packages
|
||||||
|
|
||||||
`pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work.
|
`pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work.
|
||||||
|
|
||||||
|
|||||||
@@ -103,7 +103,6 @@ Implemented in repository:
|
|||||||
|
|
||||||
Not implemented yet:
|
Not implemented yet:
|
||||||
|
|
||||||
- prepare-stage S3 list/download behavior
|
|
||||||
- archive-stage S3 upload behavior
|
- archive-stage S3 upload behavior
|
||||||
- promotion uploads
|
- promotion uploads
|
||||||
- writing `current/manifest.json` and `current/run_id.txt` to S3
|
- writing `current/manifest.json` and `current/run_id.txt` to S3
|
||||||
@@ -314,7 +313,7 @@ work/audio
|
|||||||
transcribe
|
transcribe
|
||||||
```
|
```
|
||||||
|
|
||||||
The `prepare` stage should:
|
Implemented `prepare` behavior:
|
||||||
|
|
||||||
1. list `.flac` objects under the configured S3 audio prefix,
|
1. list `.flac` objects under the configured S3 audio prefix,
|
||||||
2. fail clearly if none are found,
|
2. fail clearly if none are found,
|
||||||
@@ -331,7 +330,11 @@ spool:
|
|||||||
delete_audio_after_archive: true
|
delete_audio_after_archive: true
|
||||||
```
|
```
|
||||||
|
|
||||||
For v1, prefer retaining local workdir audio until the run has successfully archived. The source of truth remains S3, but local diagnostics are valuable during development.
|
Current boundary:
|
||||||
|
|
||||||
|
- downloaded audio is retained in spool/workdir
|
||||||
|
- spool cleanup policy remains future work
|
||||||
|
- archive does not upload source audio by default
|
||||||
|
|
||||||
## 6. Configuration Design
|
## 6. Configuration Design
|
||||||
|
|
||||||
|
|||||||
82
docs/s3-audio-input.md
Normal file
82
docs/s3-audio-input.md
Normal file
@@ -0,0 +1,82 @@
|
|||||||
|
# S3 Audio Input
|
||||||
|
|
||||||
|
This document describes implemented S3 audio input behavior in `prepare`.
|
||||||
|
|
||||||
|
## Scope
|
||||||
|
|
||||||
|
Implemented:
|
||||||
|
|
||||||
|
- `prepare` can acquire source audio from S3 when `session.inputs.audio_s3.prefix` is configured.
|
||||||
|
- object listing and download go through the storage backend abstraction.
|
||||||
|
- tests use fake storage; no live S3 service is required for test runs.
|
||||||
|
|
||||||
|
Not implemented:
|
||||||
|
|
||||||
|
- archive uploads (run uploads, promotion uploads, current pointer writes)
|
||||||
|
- spool cleanup/deletion behavior
|
||||||
|
- uploads of failed runs
|
||||||
|
|
||||||
|
## Required Configuration
|
||||||
|
|
||||||
|
`pipeline.yml`:
|
||||||
|
|
||||||
|
- `storage.s3.bucket` must be set when S3 audio input is used.
|
||||||
|
- `storage.s3.root_prefix` defaults to `dnd`.
|
||||||
|
- `spool.root` defaults to `/var/spool/narratio`.
|
||||||
|
|
||||||
|
`session.yml`:
|
||||||
|
|
||||||
|
- configure `session.campaign` and `session.session_id`.
|
||||||
|
- configure `session.inputs.audio_s3.prefix` for S3 audio input.
|
||||||
|
- do not configure `inputs.audio_dir` or `inputs.audio_files` at the same time as `inputs.audio_s3`.
|
||||||
|
|
||||||
|
## Prefix Shape
|
||||||
|
|
||||||
|
Session S3 root:
|
||||||
|
|
||||||
|
`{root_prefix}/campaigns/{campaign}/sessions/{session_id}/`
|
||||||
|
|
||||||
|
Audio prefix:
|
||||||
|
|
||||||
|
`{session_root}/{audio_s3.prefix}`
|
||||||
|
|
||||||
|
Example:
|
||||||
|
|
||||||
|
`dnd/campaigns/forsaken/sessions/2026-04-19/audio/`
|
||||||
|
|
||||||
|
Audio files must already exist in S3 before running Narratio.
|
||||||
|
|
||||||
|
## Prepare Behavior
|
||||||
|
|
||||||
|
When `inputs.audio_s3.prefix` is configured, `prepare`:
|
||||||
|
|
||||||
|
1. lists objects under the computed S3 audio prefix
|
||||||
|
2. filters to `.flac` objects
|
||||||
|
3. fails when no `.flac` objects are found
|
||||||
|
4. downloads selected objects to spool audio:
|
||||||
|
- `{spool.root}/{campaign}/{session_id}/{run_id}/audio/`
|
||||||
|
5. materializes audio into workdir audio:
|
||||||
|
- `{workspace.root}/work/{campaign}/{session_id}/{run_id}/audio/`
|
||||||
|
6. records input provenance in the manifest (bucket, key, metadata, local paths, checksum)
|
||||||
|
|
||||||
|
Notes:
|
||||||
|
|
||||||
|
- `.flac` filtering is case-insensitive.
|
||||||
|
- ETag is recorded as provider metadata only and is not treated as a checksum.
|
||||||
|
|
||||||
|
## Local Audio Development
|
||||||
|
|
||||||
|
Local audio workflows remain supported:
|
||||||
|
|
||||||
|
- `inputs.audio_dir`
|
||||||
|
- `inputs.audio_files`
|
||||||
|
|
||||||
|
These options are mutually exclusive with `inputs.audio_s3`.
|
||||||
|
|
||||||
|
## Archive Boundary
|
||||||
|
|
||||||
|
Current archive behavior remains unimplemented in this area:
|
||||||
|
|
||||||
|
- no upload of source audio by default
|
||||||
|
- no upload of failed runs
|
||||||
|
- no `current/manifest.json` or `current/run_id.txt` writes
|
||||||
@@ -5,6 +5,7 @@ title: Sample Session
|
|||||||
inputs:
|
inputs:
|
||||||
audio_dir: ./audio
|
audio_dir: ./audio
|
||||||
# Optional S3 input alternative. Do not configure with audio_dir/audio_files.
|
# Optional S3 input alternative. Do not configure with audio_dir/audio_files.
|
||||||
|
# Narratio prepare lists this prefix and downloads .flac files.
|
||||||
# audio_s3:
|
# audio_s3:
|
||||||
# prefix: "audio/"
|
# prefix: "audio/"
|
||||||
speakers_file: ./speakers.yml
|
speakers_file: ./speakers.yml
|
||||||
|
|||||||
@@ -81,6 +81,13 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
if env.Storage == nil {
|
if env.Storage == nil {
|
||||||
env.Storage = &storage.NoopBackend{}
|
env.Storage = &storage.NoopBackend{}
|
||||||
}
|
}
|
||||||
|
if env.ObjectStore == nil && env.Config != nil && env.Config.Session != nil && env.Config.Session.Inputs.AudioS3 != nil {
|
||||||
|
objectStore, err := storage.NewObjectStoreFromConfig(ctx, env.Config)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("initialize object store backend: %w", err)
|
||||||
|
}
|
||||||
|
env.ObjectStore = objectStore
|
||||||
|
}
|
||||||
if env.Notifier == nil {
|
if env.Notifier == nil {
|
||||||
env.Notifier = ¬ify.NoopSender{}
|
env.Notifier = ¬ify.NoopSender{}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,9 +14,15 @@ type ErrorRecord struct {
|
|||||||
|
|
||||||
// InputRecord captures one resolved input and optional checksum.
|
// InputRecord captures one resolved input and optional checksum.
|
||||||
type InputRecord struct {
|
type InputRecord struct {
|
||||||
Kind string `json:"kind"`
|
Kind string `json:"kind"`
|
||||||
Path string `json:"path"`
|
Path string `json:"path"`
|
||||||
Checksum string `json:"checksum,omitempty"`
|
Checksum string `json:"checksum,omitempty"`
|
||||||
|
Source string `json:"source,omitempty"`
|
||||||
|
S3Bucket string `json:"s3_bucket,omitempty"`
|
||||||
|
S3Key string `json:"s3_key,omitempty"`
|
||||||
|
S3Size int64 `json:"s3_size,omitempty"`
|
||||||
|
S3ETag string `json:"s3_etag,omitempty"`
|
||||||
|
SpoolPath string `json:"spool_path,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// ArtifactRecord captures one produced artifact and optional remote metadata.
|
// ArtifactRecord captures one produced artifact and optional remote metadata.
|
||||||
|
|||||||
@@ -6,10 +6,12 @@ import (
|
|||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
|
"path"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
@@ -33,7 +35,7 @@ func (prepareStage) Declares() IODecl {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
|
func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
|
||||||
if env == nil || env.Config == nil {
|
if env == nil || env.Config == nil {
|
||||||
return nil, fmt.Errorf("prepare: stage environment config is required")
|
return nil, fmt.Errorf("prepare: stage environment config is required")
|
||||||
}
|
}
|
||||||
@@ -89,13 +91,12 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
resolvedAudio, err := resolveAudioFiles(sessionDir, env.Config.Session.Inputs)
|
resolvedLocalAudio, useS3Audio, err := resolveAudioInputs(sessionDir, env.Config.Session.Inputs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("prepare: resolve audio inputs: %w", err)
|
return nil, fmt.Errorf("prepare: resolve audio inputs: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
copiedByDest := map[string]string{}
|
inputs := make([]manifest.InputRecord, 0, 5+len(resolvedLocalAudio))
|
||||||
inputs := make([]manifest.InputRecord, 0, 5+len(resolvedAudio))
|
|
||||||
registerInput := func(kind, path, checksum string) {
|
registerInput := func(kind, path, checksum string) {
|
||||||
inputs = append(inputs, manifest.InputRecord{Kind: kind, Path: path, Checksum: checksum})
|
inputs = append(inputs, manifest.InputRecord{Kind: kind, Path: path, Checksum: checksum})
|
||||||
}
|
}
|
||||||
@@ -134,19 +135,14 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta
|
|||||||
registerInput(cfgFile.kind, cfgFile.dst, checksum)
|
registerInput(cfgFile.kind, cfgFile.dst, checksum)
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, src := range resolvedAudio {
|
if useS3Audio {
|
||||||
base := filepath.Base(src)
|
if err := materializeS3AudioInputs(ctx, env, m, sessionID, &inputs); err != nil {
|
||||||
if prev, exists := copiedByDest[base]; exists && prev != src {
|
return nil, fmt.Errorf("prepare: materialize s3 audio: %w", err)
|
||||||
return nil, fmt.Errorf("prepare: duplicate audio basename %q from %q and %q", base, prev, src)
|
|
||||||
}
|
}
|
||||||
copiedByDest[base] = src
|
} else {
|
||||||
|
if err := materializeLocalAudioInputs(env, paths, resolvedLocalAudio, registerInput); err != nil {
|
||||||
dst := filepath.Join(paths.AudioDir, base)
|
return nil, fmt.Errorf("prepare: %w", err)
|
||||||
checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("prepare: materialize audio %q: %w", base, err)
|
|
||||||
}
|
}
|
||||||
registerInput("audio", dst, checksum)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
sort.Slice(inputs, func(i, j int) bool {
|
sort.Slice(inputs, func(i, j int) bool {
|
||||||
@@ -162,7 +158,7 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta
|
|||||||
"prepared": true,
|
"prepared": true,
|
||||||
"stage": "prepare",
|
"stage": "prepare",
|
||||||
"inputs_count": len(inputs),
|
"inputs_count": len(inputs),
|
||||||
"audio_files_resolved": len(resolvedAudio),
|
"audio_files_resolved": countAudioInputs(inputs),
|
||||||
},
|
},
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
@@ -171,7 +167,23 @@ func renderResolvedPipeline(cfg *config.PipelineConfig) ([]byte, error) {
|
|||||||
return yaml.Marshal(cfg)
|
return yaml.Marshal(cfg)
|
||||||
}
|
}
|
||||||
|
|
||||||
func resolveAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) {
|
func resolveAudioInputs(sessionDir string, inputs config.SessionInputsConfig) ([]string, bool, error) {
|
||||||
|
hasLocal := strings.TrimSpace(inputs.AudioDir) != "" || len(inputs.AudioFiles) > 0
|
||||||
|
if inputs.AudioS3 != nil {
|
||||||
|
if hasLocal {
|
||||||
|
return nil, false, fmt.Errorf("audio_dir/audio_files and audio_s3 are mutually exclusive")
|
||||||
|
}
|
||||||
|
return nil, true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
local, err := resolveLocalAudioFiles(sessionDir, inputs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, false, err
|
||||||
|
}
|
||||||
|
return local, false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func resolveLocalAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) {
|
||||||
if len(inputs.AudioFiles) > 0 {
|
if len(inputs.AudioFiles) > 0 {
|
||||||
out := make([]string, 0, len(inputs.AudioFiles))
|
out := make([]string, 0, len(inputs.AudioFiles))
|
||||||
for _, p := range inputs.AudioFiles {
|
for _, p := range inputs.AudioFiles {
|
||||||
@@ -223,6 +235,142 @@ func resolveAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func materializeLocalAudioInputs(env *Env, paths artifacts.SessionPaths, resolvedAudio []string, registerInput func(kind, path, checksum string)) error {
|
||||||
|
copiedByDest := map[string]string{}
|
||||||
|
for _, src := range resolvedAudio {
|
||||||
|
base := filepath.Base(src)
|
||||||
|
if prev, exists := copiedByDest[base]; exists && prev != src {
|
||||||
|
return fmt.Errorf("duplicate audio basename %q from %q and %q", base, prev, src)
|
||||||
|
}
|
||||||
|
copiedByDest[base] = src
|
||||||
|
|
||||||
|
dst := filepath.Join(paths.AudioDir, base)
|
||||||
|
checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("materialize audio %q: %w", base, err)
|
||||||
|
}
|
||||||
|
registerInput("audio", dst, checksum)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifest, sessionID string, inputs *[]manifest.InputRecord) error {
|
||||||
|
if env.ObjectStore == nil {
|
||||||
|
return fmt.Errorf("s3 audio input requires object store backend")
|
||||||
|
}
|
||||||
|
if env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil || env.Config.Pipeline.Storage.S3 == nil || env.Config.Session.Inputs.AudioS3 == nil {
|
||||||
|
return fmt.Errorf("s3 audio input requires pipeline.storage.s3 and session.inputs.audio_s3 configuration")
|
||||||
|
}
|
||||||
|
|
||||||
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
||||||
|
if campaign == "" {
|
||||||
|
return fmt.Errorf("session campaign is required for s3 audio input")
|
||||||
|
}
|
||||||
|
runID := strings.TrimSpace(m.RunID)
|
||||||
|
if runID == "" {
|
||||||
|
return fmt.Errorf("run id is required for s3 audio input")
|
||||||
|
}
|
||||||
|
|
||||||
|
sessionPrefix := artifacts.S3SessionPrefix(env.Config.Pipeline.Storage.S3.RootPrefix, campaign, sessionID)
|
||||||
|
audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, env.Config.Session.Inputs.AudioS3.Prefix)
|
||||||
|
objects, err := env.ObjectStore.List(ctx, audioPrefix)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("list s3 audio objects under %q: %w", audioPrefix, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
audioObjects := make([]storage.ObjectInfo, 0, len(objects))
|
||||||
|
for _, obj := range objects {
|
||||||
|
key := strings.TrimSpace(obj.Key)
|
||||||
|
if key == "" || strings.HasSuffix(key, "/") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if !isFlac(key) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
audioObjects = append(audioObjects, obj)
|
||||||
|
}
|
||||||
|
sort.Slice(audioObjects, func(i, j int) bool {
|
||||||
|
return audioObjects[i].Key < audioObjects[j].Key
|
||||||
|
})
|
||||||
|
if len(audioObjects) == 0 {
|
||||||
|
return fmt.Errorf("no .flac files found under s3 audio prefix %q", audioPrefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
spoolAudioDir := strings.TrimSpace(m.LocalSpoolDir)
|
||||||
|
if spoolAudioDir == "" {
|
||||||
|
spoolAudioDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, campaign, sessionID, runID)
|
||||||
|
}
|
||||||
|
workAudioDir := filepath.Join(pathsWorkDirForManifest(env, m, sessionID), "audio")
|
||||||
|
|
||||||
|
if err := os.MkdirAll(spoolAudioDir, 0o755); err != nil {
|
||||||
|
return fmt.Errorf("create spool audio directory %q: %w", spoolAudioDir, err)
|
||||||
|
}
|
||||||
|
if err := os.MkdirAll(workAudioDir, 0o755); err != nil {
|
||||||
|
return fmt.Errorf("create work audio directory %q: %w", workAudioDir, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
seenBase := map[string]string{}
|
||||||
|
for _, obj := range audioObjects {
|
||||||
|
base := path.Base(obj.Key)
|
||||||
|
if prev, exists := seenBase[base]; exists && prev != obj.Key {
|
||||||
|
return fmt.Errorf("duplicate s3 audio basename %q from %q and %q", base, prev, obj.Key)
|
||||||
|
}
|
||||||
|
seenBase[base] = obj.Key
|
||||||
|
|
||||||
|
spoolPath := filepath.Join(spoolAudioDir, base)
|
||||||
|
if err := env.ObjectStore.Download(ctx, obj.Key, spoolPath); err != nil {
|
||||||
|
return fmt.Errorf("download s3 audio object %q: %w", obj.Key, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
workPath := filepath.Join(workAudioDir, base)
|
||||||
|
checksum, err := copyFileIfChanged(env.ArtifactStore, spoolPath, workPath)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("materialize downloaded audio %q: %w", base, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
*inputs = append(*inputs, manifest.InputRecord{
|
||||||
|
Kind: "audio",
|
||||||
|
Path: workPath,
|
||||||
|
Checksum: checksum,
|
||||||
|
Source: "s3",
|
||||||
|
S3Bucket: strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket),
|
||||||
|
S3Key: obj.Key,
|
||||||
|
S3Size: obj.Size,
|
||||||
|
S3ETag: obj.ETag,
|
||||||
|
SpoolPath: spoolPath,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func countAudioInputs(inputs []manifest.InputRecord) int {
|
||||||
|
count := 0
|
||||||
|
for _, in := range inputs {
|
||||||
|
if in.Kind == "audio" {
|
||||||
|
count++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return count
|
||||||
|
}
|
||||||
|
|
||||||
|
func pathsWorkDirForManifest(env *Env, m *manifest.Manifest, sessionID string) string {
|
||||||
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
if m != nil && strings.TrimSpace(m.LocalWorkDir) != "" {
|
||||||
|
return strings.TrimSpace(m.LocalWorkDir)
|
||||||
|
}
|
||||||
|
runID := ""
|
||||||
|
if m != nil {
|
||||||
|
runID = strings.TrimSpace(m.RunID)
|
||||||
|
}
|
||||||
|
if runID != "" {
|
||||||
|
return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, sessionID, runID)
|
||||||
|
}
|
||||||
|
return artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
func resolvePath(baseDir, p string) (string, error) {
|
func resolvePath(baseDir, p string) (string, error) {
|
||||||
trimmed := strings.TrimSpace(p)
|
trimmed := strings.TrimSpace(p)
|
||||||
if trimmed == "" {
|
if trimmed == "" {
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
@@ -150,6 +151,139 @@ func TestPrepareStageIdempotent(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPrepareStageS3AudioDownloadAndMaterialization(t *testing.T) {
|
||||||
|
env, m := setupPrepareEnv(t)
|
||||||
|
env.Config.Session.Campaign = "forsaken"
|
||||||
|
env.Config.Session.Inputs.AudioDir = ""
|
||||||
|
env.Config.Session.Inputs.AudioFiles = nil
|
||||||
|
env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"}
|
||||||
|
env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")}
|
||||||
|
env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"}
|
||||||
|
m.RunID = "20260515T031522Z-a1b2c3d4"
|
||||||
|
m.LocalWorkDir = artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, "forsaken", m.SessionID, m.RunID)
|
||||||
|
m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m.SessionID, m.RunID)
|
||||||
|
|
||||||
|
fake := &storage.FakeBackend{}
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")})
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/bob.FLAC", Data: []byte("bob")})
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/ignore.txt", Data: []byte("x")})
|
||||||
|
env.ObjectStore = fake
|
||||||
|
|
||||||
|
result, err := (prepareStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("prepare.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if result == nil || result.Metadata["audio_files_resolved"] != 2 {
|
||||||
|
t.Fatalf("result metadata = %#v, want audio_files_resolved=2", result)
|
||||||
|
}
|
||||||
|
|
||||||
|
aliceWork := filepath.Join(m.LocalWorkDir, "audio", "alice.flac")
|
||||||
|
bobWork := filepath.Join(m.LocalWorkDir, "audio", "bob.FLAC")
|
||||||
|
aliceSpool := filepath.Join(m.LocalSpoolDir, "alice.flac")
|
||||||
|
bobSpool := filepath.Join(m.LocalSpoolDir, "bob.FLAC")
|
||||||
|
for _, p := range []string{aliceWork, bobWork, aliceSpool, bobSpool} {
|
||||||
|
if _, err := os.Stat(p); err != nil {
|
||||||
|
t.Fatalf("expected file %q: %v", p, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
audioInputs := 0
|
||||||
|
for _, in := range m.Inputs {
|
||||||
|
if in.Kind != "audio" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
audioInputs++
|
||||||
|
if in.Source != "s3" {
|
||||||
|
t.Fatalf("audio input source = %q, want s3", in.Source)
|
||||||
|
}
|
||||||
|
if in.S3Bucket != "my-dnd-archive" {
|
||||||
|
t.Fatalf("audio input bucket = %q", in.S3Bucket)
|
||||||
|
}
|
||||||
|
if in.S3Key == "" || in.SpoolPath == "" || in.Checksum == "" {
|
||||||
|
t.Fatalf("audio input missing provenance: %#v", in)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if audioInputs != 2 {
|
||||||
|
t.Fatalf("audio input count = %d, want 2", audioInputs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPrepareStageS3AudioFailures(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
setup func(env *Env, m *manifest.Manifest, fake *storage.FakeBackend)
|
||||||
|
wantError string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "no flac files",
|
||||||
|
setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) {
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/readme.txt", Data: []byte("x")})
|
||||||
|
},
|
||||||
|
wantError: "no .flac files found",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "list error",
|
||||||
|
setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) {
|
||||||
|
fake.ListErr = os.ErrPermission
|
||||||
|
},
|
||||||
|
wantError: "list s3 audio objects",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "download error",
|
||||||
|
setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) {
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")})
|
||||||
|
fake.DownloadErr = os.ErrPermission
|
||||||
|
},
|
||||||
|
wantError: "download s3 audio object",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "basename collision",
|
||||||
|
setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) {
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/a/alice.flac", Data: []byte("a")})
|
||||||
|
fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/b/alice.flac", Data: []byte("b")})
|
||||||
|
},
|
||||||
|
wantError: "duplicate s3 audio basename",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
env, m := setupPrepareEnv(t)
|
||||||
|
env.Config.Session.Campaign = "forsaken"
|
||||||
|
env.Config.Session.Inputs.AudioDir = ""
|
||||||
|
env.Config.Session.Inputs.AudioFiles = nil
|
||||||
|
env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"}
|
||||||
|
env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")}
|
||||||
|
env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"}
|
||||||
|
m.RunID = "20260515T031522Z-a1b2c3d4"
|
||||||
|
m.LocalWorkDir = artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, "forsaken", m.SessionID, m.RunID)
|
||||||
|
m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m.SessionID, m.RunID)
|
||||||
|
|
||||||
|
fake := &storage.FakeBackend{}
|
||||||
|
tt.setup(env, m, fake)
|
||||||
|
env.ObjectStore = fake
|
||||||
|
|
||||||
|
_, err := (prepareStage{}).Run(context.Background(), env, m)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), tt.wantError) {
|
||||||
|
t.Fatalf("error = %v, want %q", err, tt.wantError)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPrepareStageAudioSourceConflictFails(t *testing.T) {
|
||||||
|
env, m := setupPrepareEnv(t)
|
||||||
|
root := filepath.Dir(env.Config.SessionPath)
|
||||||
|
writeFile(t, filepath.Join(root, "audio", "a.flac"), "a")
|
||||||
|
env.Config.Session.Inputs.AudioDir = "./audio"
|
||||||
|
env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"}
|
||||||
|
|
||||||
|
_, err := (prepareStage{}).Run(context.Background(), env, m)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "mutually exclusive") {
|
||||||
|
t.Fatalf("error = %v, want mutually exclusive error", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) {
|
func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
workspace := t.TempDir()
|
workspace := t.TempDir()
|
||||||
@@ -170,6 +304,7 @@ func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) {
|
|||||||
PipelinePath: pipelinePath,
|
PipelinePath: pipelinePath,
|
||||||
Session: &config.SessionConfig{
|
Session: &config.SessionConfig{
|
||||||
SessionID: "2026-05-03",
|
SessionID: "2026-05-03",
|
||||||
|
Campaign: "sample-campaign",
|
||||||
Inputs: config.SessionInputsConfig{
|
Inputs: config.SessionInputsConfig{
|
||||||
AudioDir: "./audio",
|
AudioDir: "./audio",
|
||||||
SpeakersFile: "./speakers.yml",
|
SpeakersFile: "./speakers.yml",
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ type Env struct {
|
|||||||
Scriptorium scriptorium.Runner
|
Scriptorium scriptorium.Runner
|
||||||
Analyzer analyzer.Runner
|
Analyzer analyzer.Runner
|
||||||
Storage storage.Backend
|
Storage storage.Backend
|
||||||
|
ObjectStore storage.ObjectStore
|
||||||
Notifier notify.Sender
|
Notifier notify.Sender
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user