notarius/internal/modules/input/seriatim/adapter.go

package seriatim

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"math"
	"math/big"
	"strconv"
	"strings"

	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)

const Key = "seriatim"

const (
	DocumentKind = "transcript"
	UnitKind     = "transcript_segment"
	Format       = "application/vnd.seriatim+json"
)

var providedCapabilities = []string{
	"source.transcript",
	"transcript.speaker",
	"transcript.timestamps",
}

var _ contracts.InputAdapter = (*Adapter)(nil)

type Adapter struct{}

func New() *Adapter {
	return &Adapter{}
}

func (a *Adapter) Key() string {
	return Key
}

func (a *Adapter) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
	if ctx == nil {
		return nil, inputErrorf("context must not be nil")
	}
	if err := ctx.Err(); err != nil {
		return nil, inputErrorf("context error before parsing: %w", err)
	}
	if len(req.Raw) == 0 {
		return nil, inputErrorf("raw input must not be empty")
	}

	parsed, err := decodeTranscript(req.Raw)
	if err != nil {
		return nil, inputErrorf("parse JSON: %w", err)
	}
	if len(parsed.Segments) == 0 {
		return nil, inputErrorf("segments must not be empty")
	}

	rawDigest := digest(req.Raw)
	doc := &source.SourceDocument{
		ID:       documentID(req.SourceID, parsed.Metadata, rawDigest),
		Kind:     DocumentKind,
		Format:   Format,
		Digest:   rawDigest,
		Metadata: copyMetadata(parsed.Metadata),
	}

	seenSegmentIDs := make(map[string]struct{}, len(parsed.Segments))
	for i, segment := range parsed.Segments {
		unit, err := sourceUnit(segment, i, seenSegmentIDs)
		if err != nil {
			return nil, err
		}
		doc.Units = append(doc.Units, unit)
	}

	if err := source.ValidateDocument(doc); err != nil {
		return nil, inputErrorf("validate source document: %w", err)
	}
	return doc, nil
}

func ModuleSpec() pipeline.ModuleSpec {
	return pipeline.ModuleSpec{
		Key:      Key,
		Stage:    pipeline.StageInput,
		Provides: append([]string(nil), providedCapabilities...),
	}
}

func Register(registry *pipeline.InputAdapterRegistry) error {
	return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.InputAdapter, error) {
		return New(), nil
	})
}

func sourceUnit(segment segment, index int, seen map[string]struct{}) (source.SourceUnit, error) {
	segmentLabel := fmt.Sprintf("segment[%d]", index)
	segmentID := strings.TrimSpace(segment.ID)
	if segmentID == "" {
		return source.SourceUnit{}, inputErrorf("%s id must not be empty", segmentLabel)
	}
	if segmentID != segment.ID {
		return source.SourceUnit{}, inputErrorf("%s id %q must not contain leading or trailing whitespace", segmentLabel, segment.ID)
	}
	if _, ok := seen[segment.ID]; ok {
		return source.SourceUnit{}, inputErrorf("segment id %q is duplicated", segment.ID)
	}
	seen[segment.ID] = struct{}{}

	speaker := strings.TrimSpace(segment.Speaker)
	if speaker == "" {
		return source.SourceUnit{}, inputErrorf("segment %q speaker must not be empty", segment.ID)
	}

	start, err := validTimestamp(segment.Start, fmt.Sprintf("segment %q start", segment.ID))
	if err != nil {
		return source.SourceUnit{}, err
	}
	end, err := validTimestamp(segment.End, fmt.Sprintf("segment %q end", segment.ID))
	if err != nil {
		return source.SourceUnit{}, err
	}
	if end.Cmp(start) < 0 {
		return source.SourceUnit{}, inputErrorf("segment %q end must be greater than or equal to start", segment.ID)
	}

	if strings.TrimSpace(segment.Text) == "" {
		return source.SourceUnit{}, inputErrorf("segment %q text must not be empty", segment.ID)
	}

	return source.SourceUnit{
		ID:   segment.ID,
		Kind: UnitKind,
		Text: segment.Text,
		Metadata: map[string]any{
			MetadataSpeaker: segment.Speaker,
			MetadataStart:   segment.Start,
			MetadataEnd:     segment.End,
		},
	}, nil
}

func validTimestamp(value fmt.Stringer, label string) (*big.Rat, error) {
	raw := strings.TrimSpace(value.String())
	if raw == "" {
		return nil, inputErrorf("%s must not be empty", label)
	}
	parsed, err := strconv.ParseFloat(raw, 64)
	if err != nil {
		return nil, inputErrorf("%s must be a valid number: %w", label, err)
	}
	if math.IsInf(parsed, 0) || math.IsNaN(parsed) {
		return nil, inputErrorf("%s must be finite", label)
	}
	if parsed < 0 {
		return nil, inputErrorf("%s must not be negative", label)
	}
	rat, ok := new(big.Rat).SetString(raw)
	if !ok {
		return nil, inputErrorf("%s must be a valid number", label)
	}
	return rat, nil
}

func documentID(requestedID string, metadata map[string]any, rawDigest string) string {
	if id := strings.TrimSpace(requestedID); id != "" {
		return id
	}
	if id := stringMetadata(metadata, "id"); id != "" {
		return id
	}
	if id := stringMetadata(metadata, "source_id"); id != "" {
		return id
	}
	return "seriatim:" + strings.TrimPrefix(rawDigest, "sha256:")[:16]
}

func stringMetadata(metadata map[string]any, key string) string {
	value, ok := metadata[key].(string)
	if !ok {
		return ""
	}
	return strings.TrimSpace(value)
}

func copyMetadata(metadata map[string]any) map[string]any {
	if len(metadata) == 0 {
		return nil
	}
	copied := make(map[string]any, len(metadata))
	for key, value := range metadata {
		copied[key] = value
	}
	return copied
}

func digest(raw []byte) string {
	sum := sha256.Sum256(raw)
	return "sha256:" + hex.EncodeToString(sum[:])
}

func inputErrorf(format string, args ...any) error {
	return fmt.Errorf("seriatim input: "+format, args...)
}