From 7010da885511e1cf232ff5c5b9b84b19911328e4 Mon Sep 17 00:00:00 2001 From: m1saka Date: Wed, 9 Sep 2026 16:04:10 +0800 Subject: [PATCH] add full source: transcribe pipeline, HTTP API, OpenAI endpoint, web UI, Docker --- .dockerignore | 14 + .gitignore | 17 + Dockerfile | 101 +++++ Makefile | 82 ++++ docker-compose.yml | 34 ++ go.mod | 21 ++ go.sum | 69 ++++ internal/audio/decode.go | 228 +++++++++++ internal/config/config.go | 197 ++++++++++ internal/models/manager.go | 272 +++++++++++++ internal/server/handlers.go | 205 ++++++++++ internal/server/openai.go | 162 ++++++++ internal/server/server.go | 195 ++++++++++ internal/server/ui.go | 25 ++ internal/server/ui/index.html | 650 ++++++++++++++++++++++++++++++++ internal/store/errors.go | 6 + internal/store/store.go | 386 +++++++++++++++++++ internal/transcribe/pipeline.go | 190 ++++++++++ main.go | 64 ++++ 19 files changed, 2918 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 Makefile create mode 100644 docker-compose.yml create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/audio/decode.go create mode 100644 internal/config/config.go create mode 100644 internal/models/manager.go create mode 100644 internal/server/handlers.go create mode 100644 internal/server/openai.go create mode 100644 internal/server/server.go create mode 100644 internal/server/ui.go create mode 100644 internal/server/ui/index.html create mode 100644 internal/store/errors.go create mode 100644 internal/store/store.go create mode 100644 internal/transcribe/pipeline.go create mode 100644 main.go diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..496468c --- /dev/null +++ b/.dockerignore @@ -0,0 +1,14 @@ +# Build context exclusions — keep the image lean and avoid copying local state. +.git +.gitignore +data/ +dist/ +scripts/ +*.log +*.md +!README.md +Dockerfile +.dockerignore +docker-compose.yml +Makefile +.mimocode/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..13f124f --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# build / dist artifacts +/dist/ +/bin/ + +# runtime data (tasks, uploads, downloaded models) +/data/ + +# test audio samples (re-fetchable from sherpa-onnx releases) +/test_wavs/ + +# logs +*.log + +# editor / os +.DS_Store +.idea/ +.vscode/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..1000593 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,101 @@ +# Multi-stage, multi-arch build for record-asr. +# +# docker build -t record-asr . +# docker buildx build --platform linux/amd64,linux/arm64 -t record-asr:latest --push . +# +# buildx builds each platform NATIVELY (the builder image matches the target +# arch), so cgo links the correct prebuilt .so for that arch and NO cross +# compiler is needed — unlike the bare `make build-arm64` path. + +# ---- builder ---- +FROM golang:1.22-bookworm AS builder + +WORKDIR /src +# Allow overriding the Go module proxy, e.g. for restricted networks: +# docker build --build-arg GOPROXY=https://goproxy.cn,direct -t record-asr . +ARG GOPROXY=https://proxy.golang.org,direct +ENV GOPROXY=${GOPROXY} +# Cache module downloads (incl. the large sherpa-onnx-go .so) separately from +# source so code changes don't re-download deps. +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +# Build, then copy the prebuilt .so that match THIS builder's arch into /out/lib. +# The glob avoids hard-coding the sherpa-onnx-go-linux version. +RUN set -eux; \ + CGO_ENABLED=1 go build -trimpath -o /out/record-asr .; \ + arch="$(go env GOARCH)"; \ + case "$arch" in \ + amd64) libd=x86_64-unknown-linux-gnu;; \ + arm64) libd=aarch64-unknown-linux-gnu;; \ + *) echo "unsupported arch: $arch" >&2; exit 1;; \ + esac; \ + mkdir -p /out/lib; \ + cp "$(go env GOMODCACHE)/github.com/k2-fsa/sherpa-onnx-go-linux@"*/lib/"$libd"/*.so /out/lib/ + +# ---- models: download SenseVoice int8 + silero VAD at BUILD time ---- +# These are arch-independent ONNX files, so the same download serves every +# platform. Baking them into the image means the container starts instantly +# and needs no network at runtime. (One-time ~230MB cost per build.) +FROM debian:bookworm-slim AS models +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates curl bzip2 \ + && rm -rf /var/lib/apt/lists/* +# Extract ONLY model.int8.onnx + tokens.txt from the release tarball (skip the +# test wavs / float model) to keep the layer lean. --strip-components=1 flattens +# the top archive dir so files land directly in /models/sensevoice/. +RUN mkdir -p /models/sensevoice \ + && curl -fsSL -o /tmp/sv.tar.bz2 \ + "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2024-07-17.tar.bz2" \ + && tar -xjf /tmp/sv.tar.bz2 -C /models/sensevoice \ + --strip-components=1 --wildcards --no-anchored 'model.int8.onnx' 'tokens.txt' \ + && rm -f /tmp/sv.tar.bz2 \ + && curl -fsSL -o /models/silero_vad.onnx \ + "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/silero_vad.onnx" \ + && test -s /models/sensevoice/model.int8.onnx \ + && test -s /models/sensevoice/tokens.txt \ + && test -s /models/silero_vad.onnx \ + && ls -lh /models/sensevoice /models/silero_vad.onnx + +# ---- runtime (glibc; sherpa-onnx .so are glibc-built, so NOT alpine/musl) ---- +FROM debian:bookworm-slim + +# APT_MIRROR: optionally swap deb.debian.org for a faster regional mirror, e.g. +# docker build --build-arg APT_MIRROR=mirrors.tuna.tsinghua.edu.cn -t record-asr . +# Defaults to the upstream deb.debian.org (no swap). ffmpeg (~250MB of codec +# libs) decodes m4a/aac/opus/wma — beep is pure-Go for wav/mp3/flac/ogg only. +ARG APT_MIRROR= +RUN set -eux; \ + if [ -n "$APT_MIRROR" ]; then \ + f=/etc/apt/sources.list.d/debian.sources; \ + if [ -f "$f" ]; then sed -i "s|deb.debian.org|$APT_MIRROR|g; s|security.debian.org|$APT_MIRROR|g" "$f"; \ + else sed -i "s|deb.debian.org|$APT_MIRROR|g; s|security.debian.org|$APT_MIRROR|g" /etc/apt/sources.list; fi; \ + fi; \ + apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates curl ffmpeg \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY --from=builder /out/record-asr /app/record-asr +COPY --from=builder /out/lib /app/lib +COPY --from=models /models /models + +# Defaults; override via -e / compose. Models are baked in at /models, so +# startup skips the download entirely. Only /app/data (tasks + uploads) is a +# volume; it never holds models. +ENV ASR_ADDR=:8080 \ + ASR_DATA_DIR=/app/data \ + ASR_MODEL_DIR=/models \ + ASR_DOWNLOAD_TIMEOUT=1800 \ + LD_LIBRARY_PATH=/app/lib + +VOLUME /app/data +EXPOSE 8080 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \ + CMD curl -fsS http://127.0.0.1:8080/healthz || exit 1 + +ENTRYPOINT ["/app/record-asr"] + diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b2d53db --- /dev/null +++ b/Makefile @@ -0,0 +1,82 @@ +# Makefile for record-asr — a Go HTTP transcription service built on +# sherpa-onnx (SenseVoiceSmall int8 + silero VAD). +# +# Targets: +# make build a *portable* native binary into dist//, +# with the sherpa/onnxruntime .so files copied next to it +# and an $ORIGIN/lib rpath baked in so it runs anywhere. +# make build-amd64 same, for linux/amd64 (native here). +# make build-arm64 cross-compile for linux/arm64 +# (needs: apt-get install gcc-aarch64-linux-gnu). +# make dist build both arches and produce tarballs in dist/. +# make run build native, then run with ./data and default env. +# make tidy/clean/check. +# +# All runtime behavior is controlled by ASR_* environment variables, not build +# flags — see README.md. + +VERSION ?= $(shell git describe --tags --always 2>/dev/null || echo dev) +MODCACHE := $(shell go env GOMODCACHE) +SHERPA_LINUX := $(MODCACHE)/github.com/k2-fsa/sherpa-onnx-go-linux@v1.13.7 + +# RPATH baked into every binary so it finds the .so files shipped in ./lib +# next to it, regardless of where the bundle is unpacked. $$ORIGIN -> $ORIGIN +# (a dynamic-loader token = directory of the executable). +ORIGIN_RPATH := '-Wl,-rpath,$$ORIGIN/lib' + +.PHONY: all build build-amd64 build-arm64 dist run tidy clean check + +all: build + +# Portable native build for whatever arch `go env GOARCH` reports. +build: + @arch=$$(go env GOARCH); \ + case $$arch in \ + amd64) lib=x86_64-unknown-linux-gnu;; \ + arm64) lib=aarch64-unknown-linux-gnu;; \ + arm) lib=arm-unknown-linux-gnueabihf;; \ + *) echo "unsupported arch: $$arch"; exit 1;; \ + esac; \ + dist=dist/$$arch; mkdir -p $$dist/lib; \ + echo ">> building $$dist/record-asr ($$arch)"; \ + CGO_ENABLED=1 CGO_LDFLAGS=$(ORIGIN_RPATH) go build -trimpath -o $$dist/record-asr .; \ + cp $(SHERPA_LINUX)/lib/$$lib/*.so $$dist/lib/; \ + echo ">> built $$dist/record-asr (run with: ./$$dist/record-asr)" + +# Explicit linux/amd64 (native toolchain). +build-amd64: + @dist=dist/amd64; mkdir -p $$dist/lib; \ + echo ">> building $$dist/record-asr (amd64)"; \ + CGO_ENABLED=1 CGO_LDFLAGS=$(ORIGIN_RPATH) go build -trimpath -o $$dist/record-asr .; \ + cp $(SHERPA_LINUX)/lib/x86_64-unknown-linux-gnu/*.so $$dist/lib/; \ + echo ">> built $$dist/record-asr" + +# Cross-compile to linux/arm64. Requires a C cross-compiler: +# sudo apt-get install gcc-aarch64-linux-gnu +build-arm64: + @dist=dist/arm64; mkdir -p $$dist/lib; \ + echo ">> building $$dist/record-asr (arm64)"; \ + CGO_ENABLED=1 CC=aarch64-linux-gnu-gcc GOOS=linux GOARCH=arm64 \ + CGO_LDFLAGS=$(ORIGIN_RPATH) go build -trimpath -o $$dist/record-asr .; \ + cp $(SHERPA_LINUX)/lib/aarch64-unknown-linux-gnu/*.so $$dist/lib/; \ + echo ">> built $$dist/record-asr (run on aarch64 with qemu or on-device)" + +# Both arches + tarballs. +dist: build-amd64 build-arm64 + @cd dist && \ + tar czf record-asr-$(VERSION)-linux-amd64.tar.gz amd64 && \ + tar czf record-asr-$(VERSION)-linux-arm64.tar.gz arm64 && \ + ls -lh *.tar.gz + +# Build native then run from the build dir with a local ./data. +run: build + @ASR_DATA_DIR=$$PWD/data ./dist/$$(go env GOARCH)/record-asr + +tidy: + go mod tidy + +clean: + rm -rf dist + +check: + go vet ./... diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0e0045c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,34 @@ +# docker compose up -d +# +# Set the password (and any other ASR_* knob) in .env or your shell, e.g.: +# ASR_PASSWORD=changeme docker compose up -d +# +# Models + task state persist in the named volume `asr-data` across restarts. + +services: + asr: + build: . + image: record-asr:latest + container_name: record-asr + ports: + - "8080:8080" + environment: + ASR_PASSWORD: ${ASR_PASSWORD:?ASR_PASSWORD must be set} + ASR_LANGUAGE: ${ASR_LANGUAGE:-auto} + ASR_USE_ITN: ${ASR_USE_ITN:-1} + ASR_MAX_RETRIES: ${ASR_MAX_RETRIES:-3} + ASR_NUM_THREADS: ${ASR_NUM_THREADS:-2} + ASR_WORKERS: ${ASR_WORKERS:-1} + ASR_MAX_UPLOAD_MB: ${ASR_MAX_UPLOAD_MB:-512} + volumes: + - asr-data:/app/data + restart: unless-stopped + healthcheck: + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:8080/healthz"] + interval: 30s + timeout: 5s + start_period: 60s + retries: 3 + +volumes: + asr-data: diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..6891de1 --- /dev/null +++ b/go.mod @@ -0,0 +1,21 @@ +module git.misaka.ren/m1saka/record-asr + +go 1.22 + +require ( + github.com/gopxl/beep/v2 v2.1.1 + github.com/k2-fsa/sherpa-onnx-go v1.13.7 +) + +require ( + github.com/hajimehoshi/go-mp3 v0.3.4 // indirect + github.com/icza/bitio v1.1.0 // indirect + github.com/jfreymuth/oggvorbis v1.0.5 // indirect + github.com/jfreymuth/vorbis v1.0.2 // indirect + github.com/k2-fsa/sherpa-onnx-go-linux v1.13.7 // indirect + github.com/k2-fsa/sherpa-onnx-go-macos v1.13.7 // indirect + github.com/k2-fsa/sherpa-onnx-go-windows v1.13.7 // indirect + github.com/mewkiz/flac v1.0.12 // indirect + github.com/mewkiz/pkg v0.0.0-20230226050401-4010bf0fec14 // indirect + github.com/pkg/errors v0.9.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..ac5aab0 --- /dev/null +++ b/go.sum @@ -0,0 +1,69 @@ +github.com/d4l3k/messagediff v1.2.2-0.20190829033028-7e0a312ae40b/go.mod h1:Oozbb1TVXFac9FtSIxHBMnBCq2qeH/2KkEQxENCrlLo= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gopxl/beep/v2 v2.1.1 h1:6FYIYMm2qPAdWkjX+7xwKrViS1x0Po5kDMdRkq8NVbU= +github.com/gopxl/beep/v2 v2.1.1/go.mod h1:ZAm9TGQ9lvpoiFLd4zf5B1IuyxZhgRACMId1XJbaW0E= +github.com/hajimehoshi/go-mp3 v0.3.4 h1:NUP7pBYH8OguP4diaTZ9wJbUbk3tC0KlfzsEpWmYj68= +github.com/hajimehoshi/go-mp3 v0.3.4/go.mod h1:fRtZraRFcWb0pu7ok0LqyFhCUrPeMsGRSVop0eemFmo= +github.com/hajimehoshi/oto/v2 v2.3.1/go.mod h1:seWLbgHH7AyUMYKfKYT9pg7PhUu9/SisyJvNTT+ASQo= +github.com/icza/bitio v1.1.0 h1:ysX4vtldjdi3Ygai5m1cWy4oLkhWTAi+SyO6HC8L9T0= +github.com/icza/bitio v1.1.0/go.mod h1:0jGnlLAx8MKMr9VGnn/4YrvZiprkvBelsVIbA9Jjr9A= +github.com/icza/mighty v0.0.0-20180919140131-cfd07d671de6 h1:8UsGZ2rr2ksmEru6lToqnXgA8Mz1DP11X4zSJ159C3k= +github.com/icza/mighty v0.0.0-20180919140131-cfd07d671de6/go.mod h1:xQig96I1VNBDIWGCdTt54nHt6EeI639SmHycLYL7FkA= +github.com/jfreymuth/oggvorbis v1.0.5 h1:u+Ck+R0eLSRhgq8WTmffYnrVtSztJcYrl588DM4e3kQ= +github.com/jfreymuth/oggvorbis v1.0.5/go.mod h1:1U4pqWmghcoVsCJJ4fRBKv9peUJMBHixthRlBeD6uII= +github.com/jfreymuth/vorbis v1.0.2 h1:m1xH6+ZI4thH927pgKD8JOH4eaGRm18rEE9/0WKjvNE= +github.com/jfreymuth/vorbis v1.0.2/go.mod h1:DoftRo4AznKnShRl1GxiTFCseHr4zR9BN3TWXyuzrqQ= +github.com/jszwec/csvutil v1.5.1/go.mod h1:Rpu7Uu9giO9subDyMCIQfHVDuLrcaC36UA4YcJjGBkg= +github.com/k2-fsa/sherpa-onnx-go v1.13.7 h1:QG5bjU5LQEz2nNDT9Rt0V1k3Pvr22OdQCVByChT/aPM= +github.com/k2-fsa/sherpa-onnx-go v1.13.7/go.mod h1:JC2kmiNHLMvY7zEbAVXduAymntUTi4QQLkmjnuYlDxo= +github.com/k2-fsa/sherpa-onnx-go-linux v1.13.7 h1:URypLy1vlSQOKLnuqODCUjvOoFPxY022NNbbirHwxWw= +github.com/k2-fsa/sherpa-onnx-go-linux v1.13.7/go.mod h1:NXEH2rsBgTdqY59YpPq6CtSBlBAXy/8a9FmpLERU97I= +github.com/k2-fsa/sherpa-onnx-go-macos v1.13.7 h1:miDo530lFRiiKxEP66PpPry+RHQuucYK+IrBe+wqCdI= +github.com/k2-fsa/sherpa-onnx-go-macos v1.13.7/go.mod h1:ZOhUAXC62Unj0ZNfu6zxSFKcW96aXf7P3BsqiUyOBbE= +github.com/k2-fsa/sherpa-onnx-go-windows v1.13.7 h1:XOgxcWUOBXjNZpdTIgDGur8v1+A+Tpm688XXYS1y6NI= +github.com/k2-fsa/sherpa-onnx-go-windows v1.13.7/go.mod h1:5AX7TU8+P/gInjglY1ijtWUM2b8iyR0QX4yEngzMe64= +github.com/mewkiz/flac v1.0.12 h1:5Y1BRlUebfiVXPmz7hDD7h3ceV2XNrGNMejNVjDpgPY= +github.com/mewkiz/flac v1.0.12/go.mod h1:1UeXlFRJp4ft2mfZnPLRpQTd7cSjb/s17o7JQzzyrCA= +github.com/mewkiz/pkg v0.0.0-20230226050401-4010bf0fec14 h1:tnAPMExbRERsyEYkmR1YjhTgDM0iqyiBYf8ojRXxdbA= +github.com/mewkiz/pkg v0.0.0-20230226050401-4010bf0fec14/go.mod h1:QYCFBiH5q6XTHEbWhR0uhR3M9qNPoD2CSQzr0g75kE4= +github.com/orcaman/writerseeker v0.0.0-20200621085525-1d3f536ff85e h1:s2RNOM/IGdY0Y6qfTeUKhDawdHDpK9RGBdx80qN4Ttw= +github.com/orcaman/writerseeker v0.0.0-20200621085525-1d3f536ff85e/go.mod h1:nBdnFKj15wFbf94Rwfq4m30eAcyY9V/IyKAGQFtqkW0= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/image v0.5.0/go.mod h1:FVC7BI/5Ym8R25iw5OLsgshdUBbT1h5jZTpA+mvAdZ4= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220712014510-0a85c31ab51e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/audio/decode.go b/internal/audio/decode.go new file mode 100644 index 0000000..5cdfb36 --- /dev/null +++ b/internal/audio/decode.go @@ -0,0 +1,228 @@ +// Package audio decodes uploaded audio files into the 16 kHz mono float32 +// sample slice that sherpa-onnx expects. +// +// Pure-Go beep decoders handle .wav/.mp3/.flac/.ogg. Everything else +// (notably .m4a/.aac — there is no robust pure-Go AAC decoder) is decoded by +// shelling out to ffmpeg, which converts to 16 kHz mono f32le PCM on stdout. +package audio + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "math" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/gopxl/beep/v2" + "github.com/gopxl/beep/v2/flac" + "github.com/gopxl/beep/v2/mp3" + "github.com/gopxl/beep/v2/vorbis" + "github.com/gopxl/beep/v2/wav" +) + +// TargetSampleRate is the rate sherpa-onnx models were trained on. +const TargetSampleRate = 16000 + +// Package-level decoder configuration set once by the server bootstrap. +var ( + ffmpegPath = "ffmpeg" + decTimeout = 30 * time.Minute +) + +// Configure sets the ffmpeg binary path and a per-file decode timeout. Call +// once at startup from the config. Safe to call multiple times. +func Configure(ffmpeg string, timeout time.Duration) { + if strings.TrimSpace(ffmpeg) != "" { + ffmpegPath = ffmpeg + } + if timeout > 0 { + decTimeout = timeout + } +} + +// beepExts are the formats beep decodes natively (pure Go, no external tool). +var beepExts = map[string]bool{ + ".wav": true, ".mp3": true, ".flac": true, ".ogg": true, ".oga": true, +} + +// Decode loads the file at path and returns mono float32 samples resampled to +// TargetSampleRate. Beep handles its native formats; ffmpeg handles the rest +// (m4a, aac, opus, wma, webm, aiff, …). If a beep-native file fails to decode, +// it transparently falls back to ffmpeg so a quirky/mis-named file still works. +func Decode(path string) (samples []float32, sampleRate int, err error) { + ext := strings.ToLower(filepath.Ext(path)) + + if beepExts[ext] { + if s, format, e := tryBeep(path); e == nil { + out, e2 := finalizeBeep(s, format) + if e2 == nil { + return out, TargetSampleRate, nil + } + // beep read failed mid-stream; fall through to ffmpeg. + } + // beep decode failed entirely; fall through to ffmpeg. + } + return decodeFFmpeg(path) +} + +// tryBeep opens the file and dispatches to the matching beep decoder. +func tryBeep(path string) (beep.StreamSeekCloser, beep.Format, error) { + f, err := os.Open(path) + if err != nil { + return nil, beep.Format{}, fmt.Errorf("open audio: %w", err) + } + switch strings.ToLower(filepath.Ext(path)) { + case ".wav": + return wav.Decode(f) + case ".mp3": + return mp3.Decode(f) + case ".flac": + return flac.Decode(f) + case ".ogg", ".oga": + return vorbis.Decode(f) + } + // Caller should not reach here for non-beep exts. + f.Close() + return nil, beep.Format{}, fmt.Errorf("not a beep-supported extension") +} + +// finalizeBeep drains the decoder into a mono float64 buffer at the native rate +// and resamples to 16 kHz mono float32. +func finalizeBeep(s beep.StreamSeekCloser, format beep.Format) ([]float32, error) { + defer s.Close() + const bufLen = 8192 + buf := make([][2]float64, bufLen) + mono := make([]float64, 0, 1<<20) + for { + n, ok := s.Stream(buf) + if n > 0 { + for i := 0; i < n; i++ { + mono = append(mono, (buf[i][0]+buf[i][1])/2.0) + } + } + if !ok { + break + } + } + if err := s.Err(); err != nil { + return nil, fmt.Errorf("read audio: %w", err) + } + if len(mono) == 0 { + return nil, fmt.Errorf("audio file has no samples") + } + return resampleMono(mono, int(format.SampleRate), TargetSampleRate), nil +} + +// decodeFFmpeg runs `ffmpeg -i path -f f32le -ar 16000 -ac 1 -` and reads the +// raw little-endian float32 PCM from stdout. This is the universal fallback +// for formats beep cannot decode (m4a/aac/opus/wma/…). +func decodeFFmpeg(path string) ([]float32, int, error) { + if _, err := exec.LookPath(ffmpegPath); err != nil { + return nil, 0, fmt.Errorf("ffmpeg not found (%s): install ffmpeg or set ASR_FFMPEG to decode this format", ffmpegPath) + } + ctx, cancel := context.WithTimeout(context.Background(), decTimeout) + defer cancel() + + var stderr bytes.Buffer + // -vn drops the video stream entirely: for video inputs (mp4/mkv/…) this + // skips video decoding so only the audio track is extracted; for + // audio-only inputs it's a harmless no-op. + cmd := exec.CommandContext(ctx, ffmpegPath, + "-nostdin", "-hide_banner", "-loglevel", "error", + "-i", path, + "-vn", + "-f", "f32le", "-acodec", "pcm_f32le", + "-ar", "16000", "-ac", "1", + "-", + ) + var stdout bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + if err := cmd.Run(); err != nil { + if ctx.Err() == context.DeadlineExceeded { + return nil, 0, fmt.Errorf("ffmpeg decode timed out after %s", decTimeout) + } + msg := strings.TrimSpace(stderr.String()) + if msg == "" { + msg = err.Error() + } + return nil, 0, fmt.Errorf("ffmpeg decode failed: %s", msg) + } + raw := stdout.Bytes() + if len(raw) == 0 { + return nil, 0, fmt.Errorf("ffmpeg produced no audio (empty or no audio stream in %s)", filepath.Base(path)) + } + if len(raw)%4 != 0 { + raw = raw[:len(raw)-len(raw)%4] // drop trailing partial sample + } + out := make([]float32, len(raw)/4) + for i := 0; i < len(out); i++ { + out[i] = math.Float32frombits(binary.LittleEndian.Uint32(raw[i*4 : i*4+4])) + } + return out, TargetSampleRate, nil +} + +// resampleMono converts a mono float64 slice from srcRate to dstRate using +// linear interpolation. Adequate for speech recognition at 16 kHz. +func resampleMono(in []float64, srcRate, dstRate int) []float32 { + if srcRate == dstRate { + out := make([]float32, len(in)) + for i, v := range in { + out[i] = float32(v) + } + return out + } + if srcRate <= 0 || dstRate <= 0 || len(in) < 2 { + out := make([]float32, len(in)) + for i, v := range in { + out[i] = float32(v) + } + return out + } + ratio := float64(dstRate) / float64(srcRate) + outLen := int(float64(len(in)-1) * ratio) + if outLen < 1 { + outLen = 1 + } + out := make([]float32, outLen) + for i := 0; i < outLen; i++ { + pos := float64(i) / ratio + idx := int(pos) + if idx >= len(in)-1 { + out[i] = float32(in[len(in)-1]) + continue + } + frac := pos - float64(idx) + v := in[idx]*(1-frac) + in[idx+1]*frac + out[i] = float32(v) + } + return out +} + +// acceptedExts is the allowlist the upload handler rejects against. Beep-native +// formats decode without ffmpeg; the rest need ffmpeg on PATH. +var acceptedExts = map[string]bool{ + ".wav": true, ".mp3": true, ".flac": true, ".ogg": true, ".oga": true, + ".m4a": true, ".m4b": true, ".aac": true, ".opus": true, + ".wma": true, ".webm": true, ".aiff": true, ".aif": true, + ".amr": true, ".au": true, ".snd": true, ".gsm": true, + // Video containers — the audio track is extracted via ffmpeg (-vn drops video). + ".mp4": true, ".m4v": true, ".mkv": true, ".mov": true, ".avi": true, + ".ts": true, ".m2ts": true, ".mts": true, ".wmv": true, ".flv": true, + ".3gp": true, ".3g2": true, ".mpg": true, ".mpeg": true, ".vob": true, +} + +// SupportedExtensions lists the file extensions Decode accepts. +func SupportedExtensions() string { + return ".wav, .mp3, .flac, .ogg, .m4a, .aac, .opus, .aiff, + 视频 .mp4/.mkv/.mov/.avi/.ts… (ffmpeg extracts audio)" +} + +// IsSupported reports whether the file at path has a decodable extension. +func IsSupported(path string) bool { + return acceptedExts[strings.ToLower(filepath.Ext(path))] +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..3fa9795 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,197 @@ +// Package config loads all runtime configuration from environment variables. +// +// Every knob the transcriber exposes is controlled by an ASR_* environment +// variable so the same binary can be reconfigured without a rebuild. +package config + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "time" +) + +// Config holds all runtime configuration derived from environment variables. +type Config struct { + // ASR_PASSWORD: shared secret required on every API request. If empty, + // auth is disabled and a loud warning is logged (intended for local use). + Password string + // ASR_ADDR: listen address (default ":8080"). + Addr string + // ASR_DATA_DIR: root for persisted state (default "./data"). + DataDir string + // ASR_MODEL_DIR: where models live / are auto-downloaded + // (default "/models"). + ModelDir string + // ASR_NUM_THREADS: ONNX compute threads (default: min(4, NumCPU)). + NumThreads int + // ASR_LANGUAGE: SenseVoice language hint. "auto" detects per segment. + Language string + // ASR_USE_ITN: enable inverse text normalization (numbers/punctuation). + UseITN bool + // ASR_MAX_RETRIES: automatic retry attempts before a task is marked failed. + MaxRetries int + // ASR_MAX_UPLOAD_MB: per-upload size cap (default 512). + MaxUploadBytes int64 + // ASR_WORKERS: concurrent transcription workers (default 1). + Workers int + // ASR_MAX_CONCURRENT: cap on simultaneously in-flight synchronous + // transcriptions on the OpenAI endpoint. Bounds onnxruntime thread + // contention under load. Defaults to ASR_NUM_THREADS. Excess requests + // wait for a slot (or get 503 if the client disconnects while waiting). + MaxConcurrent int + + // VAD (silero) tuning. + VadThreshold float32 + VadMinSpeechDuration float32 + VadMinSilenceDuration float32 + VadMaxSpeechDuration float32 + VadWindowSize int + VadBufferSeconds float32 + + // ASR_DOWNLOAD_TIMEOUT: timeout for the initial model download. + DownloadTimeout time.Duration + // ASR_FFMPEG: path to the ffmpeg binary used to decode formats beep + // can't (m4a/aac/opus/wma …). Defaults to "ffmpeg" on PATH. Only + // required when such formats are actually uploaded. + FFmpegPath string + // ASR_DECODE_TIMEOUT: per-file decode timeout (beep + ffmpeg fallback). + DecodeTimeout time.Duration +} + +// Load reads the environment and returns a validated Config. +func Load() (Config, error) { + c := Config{ + Password: strings.TrimSpace(os.Getenv("ASR_PASSWORD")), + Addr: envStr("ASR_ADDR", ":8080"), + DataDir: envStr("ASR_DATA_DIR", "./data"), + NumThreads: envInt("ASR_NUM_THREADS", defaultThreads()), + Language: envStr("ASR_LANGUAGE", "auto"), + UseITN: envBool("ASR_USE_ITN", true), + MaxRetries: envInt("ASR_MAX_RETRIES", 3), + MaxUploadBytes: int64(envInt("ASR_MAX_UPLOAD_MB", 512)) * 1024 * 1024, + Workers: envInt("ASR_WORKERS", 1), + MaxConcurrent: envInt("ASR_MAX_CONCURRENT", 0), + VadThreshold: float32(envFloat("ASR_VAD_THRESHOLD", 0.5)), + VadMinSpeechDuration: float32(envFloat("ASR_VAD_MIN_SPEECH", 0.5)), + VadMinSilenceDuration: float32(envFloat("ASR_VAD_MIN_SILENCE", 0.5)), + VadMaxSpeechDuration: float32(envFloat("ASR_VAD_MAX_SPEECH", 20)), + VadWindowSize: envInt("ASR_VAD_WINDOW", 512), + VadBufferSeconds: float32(envFloat("ASR_VAD_BUFFER_SECS", 60)), + DownloadTimeout: time.Duration(envInt("ASR_DOWNLOAD_TIMEOUT", 1800)) * time.Second, + FFmpegPath: envStr("ASR_FFMPEG", "ffmpeg"), + DecodeTimeout: time.Duration(envInt("ASR_DECODE_TIMEOUT", 1800)) * time.Second, + } + c.ModelDir = envStr("ASR_MODEL_DIR", filepath.Join(c.DataDir, "models")) + + if c.Workers < 1 { + c.Workers = 1 + } + if c.MaxConcurrent < 1 { + c.MaxConcurrent = c.NumThreads + } + if c.NumThreads < 1 { + c.NumThreads = 1 + } + if c.MaxRetries < 0 { + c.MaxRetries = 0 + } + if c.VadMaxSpeechDuration < 1 { + c.VadMaxSpeechDuration = 1 + } + if c.DecodeTimeout < time.Minute { + c.DecodeTimeout = 30 * time.Minute + } + if c.VadWindowSize <= 0 { + c.VadWindowSize = 512 + } + if c.MaxUploadBytes < 1<<20 { + c.MaxUploadBytes = 512 << 20 + } + if c.DownloadTimeout < time.Minute { + c.DownloadTimeout = 30 * time.Minute + } + + // Absolutize data dir so relative file paths in persisted state survive + // regardless of the process working directory at request time. + if abs, err := filepath.Abs(c.DataDir); err == nil { + c.DataDir = abs + } + if abs, err := filepath.Abs(c.ModelDir); err == nil { + c.ModelDir = abs + } + return c, nil +} + +// AuthEnabled reports whether API auth is enforced. +func (c Config) AuthEnabled() bool { return c.Password != "" } + +func (c Config) String() string { + return fmt.Sprintf("addr=%s data=%s models=%s threads=%d lang=%s itn=%t workers=%d maxconcurrent=%d retries=%d vad{thr=%.2f max=%.0fs win=%d}", + c.Addr, c.DataDir, c.ModelDir, c.NumThreads, c.Language, c.UseITN, c.Workers, + c.MaxConcurrent, c.MaxRetries, c.VadThreshold, c.VadMaxSpeechDuration, c.VadWindowSize) +} + +func defaultThreads() int { + n := runtime.NumCPU() + if n > 4 { + n = 4 + } + if n < 1 { + n = 1 + } + return n +} + +func envStr(key, def string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return def +} + +func envInt(key string, def int) int { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil { + return def + } + return n +} + +func envFloat(key string, def float64) float64 { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return def + } + f, err := strconv.ParseFloat(v, 64) + if err != nil { + return def + } + return f +} + +func envBool(key string, def bool) bool { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return def + } + b, err := strconv.ParseBool(v) + if err != nil { + // Accept "0"/"1" and "yes"/"no" too. + switch strings.ToLower(v) { + case "yes", "on", "true", "1": + return true + case "no", "off", "false", "0": + return false + } + return def + } + return b +} diff --git a/internal/models/manager.go b/internal/models/manager.go new file mode 100644 index 0000000..4bdf4d4 --- /dev/null +++ b/internal/models/manager.go @@ -0,0 +1,272 @@ +// Package models owns model lifecycle: it auto-downloads the SenseVoiceSmall +// int8 model and the silero VAD on first run, then builds and holds the +// sherpa-onnx recognizer and voice-activity detector for the transcriber. +package models + +import ( + "archive/tar" + "compress/bzip2" + "context" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" + + sherpa "github.com/k2-fsa/sherpa-onnx-go/sherpa_onnx" + + "git.misaka.ren/m1saka/record-asr/internal/config" +) + +// Source URLs for the prebuilt models. Pinning these to the k2-fsa +// "asr-models" release channel keeps the downloads stable and resumable. +const ( + senseVoiceTarballURL = "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2024-07-17.tar.bz2" + sileroVadURL = "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/silero_vad.onnx" +) + +// Manager holds the loaded recognizer and VAD. It is safe for concurrent use; +// the underlying sherpa-onnx objects are not re-entrant per stream, but the +// transcriber creates a fresh OfflineStream per segment under its own lock. +type Manager struct { + cfg config.Config + Recognizer *sherpa.OfflineRecognizer + VAD *sherpa.VoiceActivityDetector + + // vadMu serializes access to the stateful VAD across concurrent workers. + vadMu sync.Mutex + ready bool +} + +// VadMu returns the mutex guarding the VAD. The transcribe pipeline locks it +// for the duration of a single file's segmentation so VAD state never +// interleaves between files. +func (m *Manager) VadMu() *sync.Mutex { return &m.vadMu } + +// New ensures the model files are present (downloading if necessary) and then +// constructs the recognizer and VAD. It blocks until both are ready. +func New(ctx context.Context, cfg config.Config) (*Manager, error) { + m := &Manager{cfg: cfg} + + senseDir := filepath.Join(cfg.ModelDir, "sensevoice") + senseModel := filepath.Join(senseDir, "model.int8.onnx") + senseTokens := filepath.Join(senseDir, "tokens.txt") + sileroModel := filepath.Join(cfg.ModelDir, "silero_vad.onnx") + + if err := os.MkdirAll(senseDir, 0o755); err != nil { + return nil, fmt.Errorf("create model dir %s: %w", senseDir, err) + } + + // Ensure models are on disk before constructing anything. Downloading the + // ~230MB SenseVoice tarball can take a while on first run; log progress. + if !fileOK(senseModel) || !fileOK(senseTokens) { + fmt.Println("model: downloading SenseVoiceSmall int8 (first run)...") + if err := downloadSenseVoice(ctx, cfg.DownloadTimeout, senseVoiceTarballURL, senseDir); err != nil { + return nil, fmt.Errorf("download SenseVoice: %w", err) + } + fmt.Println("model: SenseVoice ready") + } + if !fileOK(sileroModel) { + fmt.Println("model: downloading silero VAD...") + if err := downloadFile(ctx, cfg.DownloadTimeout, sileroVadURL, sileroModel); err != nil { + return nil, fmt.Errorf("download silero VAD: %w", err) + } + fmt.Println("model: silero VAD ready") + } + if !fileOK(senseModel) || !fileOK(senseTokens) || !fileOK(sileroModel) { + return nil, fmt.Errorf("model files missing after download: %s,%s,%s", senseModel, senseTokens, sileroModel) + } + + itn := 0 + if cfg.UseITN { + itn = 1 + } + recCfg := &sherpa.OfflineRecognizerConfig{ + FeatConfig: sherpa.FeatureConfig{SampleRate: 16000, FeatureDim: 80}, + ModelConfig: sherpa.OfflineModelConfig{ + SenseVoice: sherpa.OfflineSenseVoiceModelConfig{ + Model: senseModel, + Language: cfg.Language, + UseInverseTextNormalization: itn, + }, + Tokens: senseTokens, + NumThreads: cfg.NumThreads, + Provider: "cpu", + Debug: 0, + }, + DecodingMethod: "greedy_search", + } + rec := sherpa.NewOfflineRecognizer(recCfg) + if rec == nil { + return nil, fmt.Errorf("failed to create recognizer (model=%s tokens=%s)", senseModel, senseTokens) + } + + vadCfg := &sherpa.VadModelConfig{ + SileroVad: sherpa.SileroVadModelConfig{ + Model: sileroModel, + Threshold: cfg.VadThreshold, + MinSilenceDuration: cfg.VadMinSilenceDuration, + MinSpeechDuration: cfg.VadMinSpeechDuration, + WindowSize: cfg.VadWindowSize, + MaxSpeechDuration: cfg.VadMaxSpeechDuration, + }, + SampleRate: 16000, + NumThreads: cfg.NumThreads, + Provider: "cpu", + } + // bufferSizeInSeconds: how much audio the VAD buffers internally. Sized to + // the largest expected utterance so we never drop speech waiting on a flush. + vad := sherpa.NewVoiceActivityDetector(vadCfg, cfg.VadBufferSeconds) + if vad == nil { + sherpa.DeleteOfflineRecognizer(rec) + return nil, fmt.Errorf("failed to create VAD (model=%s)", sileroModel) + } + + m.Recognizer = rec + m.VAD = vad + m.ready = true + fmt.Printf("model: loaded (sherpa-onnx %s, onnxruntime %s)\n", sherpa.GetVersion(), sherpa.GetOnnxruntimeVersion()) + return m, nil +} + +// Close releases the native resources backing the recognizer and VAD. +func (m *Manager) Close() { + m.vadMu.Lock() + defer m.vadMu.Unlock() + if m.VAD != nil { + sherpa.DeleteVoiceActivityDetector(m.VAD) + m.VAD = nil + } + if m.Recognizer != nil { + sherpa.DeleteOfflineRecognizer(m.Recognizer) + m.Recognizer = nil + } + m.ready = false +} + +// fileOK reports whether path exists, is a regular file, and is non-empty. +func fileOK(path string) bool { + st, err := os.Stat(path) + return err == nil && !st.IsDir() && st.Size() > 0 +} + +// downloadFile fetches a single file from url into dest via a temp file + +// atomic rename, so a partial download never replaces a good file. +func downloadFile(ctx context.Context, timeout time.Duration, url, dest string) error { + dctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + req, err := http.NewRequestWithContext(dctx, http.MethodGet, url, nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("HTTP %d for %s", resp.StatusCode, url) + } + tmp := dest + ".tmp" + f, err := os.Create(tmp) + if err != nil { + return err + } + if _, err := io.Copy(f, resp.Body); err != nil { + f.Close() + os.Remove(tmp) + return err + } + if err := f.Close(); err != nil { + os.Remove(tmp) + return err + } + return os.Rename(tmp, dest) +} + +// downloadSenseVoice streams the .tar.bz2 release directly through a bzip2 +// reader into a tar reader, extracting only the two files we need +// (model.int8.onnx, tokens.txt) into destDir. This avoids materializing the +// full ~230MB archive on disk. +func downloadSenseVoice(ctx context.Context, timeout time.Duration, url, destDir string) error { + dctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + req, err := http.NewRequestWithContext(dctx, http.MethodGet, url, nil) + if err != nil { + return err + } + // GitHub releases 302 to a CDN; the default client follows redirects. + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("HTTP %d for %s", resp.StatusCode, url) + } + + bz := bzip2.NewReader(resp.Body) + tr := tar.NewReader(bz) + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return fmt.Errorf("read tar: %w", err) + } + if hdr.Typeflag != tar.TypeReg && hdr.Typeflag != tar.TypeRegA { + continue + } + base := filepath.Base(hdr.Name) + if base != "model.int8.onnx" && base != "tokens.txt" { + continue + } + dst := filepath.Join(destDir, base) + tmp := dst + ".tmp" + f, err := os.Create(tmp) + if err != nil { + return err + } + if _, err := io.Copy(f, tr); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("extract %s: %w", base, err) + } + if err := f.Close(); err != nil { + os.Remove(tmp) + return err + } + if err := os.Rename(tmp, dst); err != nil { + return err + } + } + // Guard against a successful walk that found nothing matching. + if !fileOK(filepath.Join(destDir, "model.int8.onnx")) || !fileOK(filepath.Join(destDir, "tokens.txt")) { + return fmt.Errorf("tarball did not contain model.int8.onnx/tokens.txt") + } + return nil +} + +// ModelDir returns the configured model directory (used by the server for +// diagnostics). +func (m *Manager) ModelDir() string { return m.cfg.ModelDir } + +// Ready reports whether the recognizer and VAD are loaded. +func (m *Manager) Ready() bool { + m.vadMu.Lock() + defer m.vadMu.Unlock() + return m.ready +} + +// VersionInfo returns sherpa-onnx + onnxruntime build versions for /healthz. +func (m *Manager) VersionInfo() string { + return fmt.Sprintf("sherpa-onnx %s, onnxruntime %s", + strings.TrimSpace(sherpa.GetVersion()), + strings.TrimSpace(sherpa.GetOnnxruntimeVersion())) +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go new file mode 100644 index 0000000..4ca11d3 --- /dev/null +++ b/internal/server/handlers.go @@ -0,0 +1,205 @@ +package server + +import ( + "encoding/json" + "net/http" + "strings" + + "git.misaka.ren/m1saka/record-asr/internal/audio" + "git.misaka.ren/m1saka/record-asr/internal/store" +) + +// writeJSON marshals v as JSON with the given status. +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +// handleHealthz reports liveness and model readiness (public). +func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{ + "status": "ok", + "models": s.manager != nil && s.manager.Ready(), + "version": s.modelVersion(), + }) +} + +// handleRoot returns a tiny service descriptor (public). +func (s *Server) handleRoot(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "name": "record-asr", + "engine": "SenseVoiceSmall (int8) + silero VAD", + "version": s.modelVersion(), + "endpoints": []string{ + "POST /api/tasks (multipart: file) -> enqueue", + "GET /api/tasks -> list", + "GET /api/tasks/{id} -> detail", + "DELETE /api/tasks/{id} -> delete", + "POST /api/tasks/{id}/retry -> re-run failed task", + "POST /api/tasks/{id}/skip -> abandon failed task", + "GET /api/tasks/{id}/transcript -> transcript", + }, + }) +} + +// handleCreateTask accepts a multipart upload and enqueues a transcription task. +func (s *Server) handleCreateTask(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, s.cfg.MaxUploadBytes) + if err := r.ParseMultipartForm(s.cfg.MaxUploadBytes); err != nil { + renderError(w, http.StatusRequestEntityTooLarge, + "upload rejected: "+err.Error()) + return + } + file, hdr, err := r.FormFile("file") + if err != nil { + renderError(w, http.StatusBadRequest, "missing 'file' multipart field") + return + } + defer file.Close() + + name := hdr.Filename + if name == "" { + name = "upload.wav" + } + if !audio.IsSupported(name) { + renderError(w, http.StatusUnsupportedMediaType, + "unsupported extension; supported: "+audio.SupportedExtensions()) + return + } + + t, err := s.store.Create(name, file, s.cfg.MaxUploadBytes, s.cfg.MaxRetries) + if err != nil { + renderError(w, http.StatusInternalServerError, "store: "+err.Error()) + return + } + w.Header().Set("Location", "/api/tasks/"+t.ID) + writeJSON(w, http.StatusAccepted, t) +} + +// handleListTasks returns all tasks newest-first. +func (s *Server) handleListTasks(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"tasks": s.store.List()}) +} + +// handleGetTask returns one task. +func (s *Server) handleGetTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + t := s.store.Get(id) + if t == nil { + renderError(w, http.StatusNotFound, "task not found") + return + } + writeJSON(w, http.StatusOK, t) +} + +// handleDeleteTask removes a task and its audio. +func (s *Server) handleDeleteTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.store.Delete(id); err != nil { + if err == store.ErrNotFound { + renderError(w, http.StatusNotFound, "task not found") + return + } + renderError(w, http.StatusInternalServerError, "delete: "+err.Error()) + return + } + w.WriteHeader(http.StatusNoContent) +} + +// handleRetryTask requeues a failed/skipped task for another attempt. +func (s *Server) handleRetryTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.store.Retry(id); err != nil { + if err == store.ErrNotFound { + renderError(w, http.StatusNotFound, "task not found") + return + } + renderError(w, http.StatusInternalServerError, "retry: "+err.Error()) + return + } + t := s.store.Get(id) + writeJSON(w, http.StatusAccepted, t) +} + +// handleSkipTask abandons a task so the queue moves on (the "continue" action). +func (s *Server) handleSkipTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.store.Skip(id); err != nil { + if err == store.ErrNotFound { + renderError(w, http.StatusNotFound, "task not found") + return + } + renderError(w, http.StatusInternalServerError, "skip: "+err.Error()) + return + } + t := s.store.Get(id) + writeJSON(w, http.StatusAccepted, t) +} + +// handleTranscript returns just the transcript of a completed task. +func (s *Server) handleTranscript(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + t := s.store.Get(id) + if t == nil { + renderError(w, http.StatusNotFound, "task not found") + return + } + if t.Status != store.StatusCompleted || t.Result == nil { + renderError(w, http.StatusConflict, "transcript not ready (status: "+t.Status+")") + return + } + writeJSON(w, http.StatusOK, t.Result) +} + +// --- auth --- + +// authMiddleware wraps a handler with the password check. Auth is enforced +// for paths under /api/ and /v1/ (the OpenAI-compatible endpoint); /healthz +// and / stay public. When ASR_PASSWORD is unset, auth is disabled (a warning +// was logged at startup). +func (s *Server) authMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if (!strings.HasPrefix(r.URL.Path, "/api/") && !strings.HasPrefix(r.URL.Path, "/v1/")) || !s.cfg.AuthEnabled() { + next.ServeHTTP(w, r) + return + } + if !constantTimeEqual(extractToken(r), s.cfg.Password) { + w.Header().Set("WWW-Authenticate", `Bearer realm="record-asr"`) + renderError(w, http.StatusUnauthorized, "unauthorized") + return + } + next.ServeHTTP(w, r) + }) +} + +// extractToken pulls the shared secret from, in order: X-Api-Key header, +// Authorization: Bearer , or ?token= query parameter. +func extractToken(r *http.Request) string { + if v := strings.TrimSpace(r.Header.Get("X-Api-Key")); v != "" { + return v + } + if v := r.Header.Get("Authorization"); v != "" { + if strings.HasPrefix(strings.ToLower(v), "bearer ") { + return strings.TrimSpace(v[7:]) + } + return strings.TrimSpace(strings.TrimPrefix(v, "Bearer ")) + } + return strings.TrimSpace(r.URL.Query().Get("token")) +} + +// constantTimeEqual compares two strings in constant time. +func constantTimeEqual(a, b string) bool { + if len(a) != len(b) { + return false + } + var v byte + for i := 0; i < len(a); i++ { + v |= a[i] ^ b[i] + } + return v == 0 +} diff --git a/internal/server/openai.go b/internal/server/openai.go new file mode 100644 index 0000000..a2d319e --- /dev/null +++ b/internal/server/openai.go @@ -0,0 +1,162 @@ +package server + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + + "git.misaka.ren/m1saka/record-asr/internal/audio" + "git.misaka.ren/m1saka/record-asr/internal/transcribe" +) + +// handleOpenAITranscription implements POST /v1/audio/transcriptions — an +// OpenAI-compatible, *synchronous* transcription endpoint. Drop-in for any +// client that speaks the OpenAI audio-transcription API. +// +// Multipart fields: +// - file (required) the audio/video file +// - model (ignored for compatibility; there is one model) +// - language (accepted for compatibility; recognition auto-detects per +// segment unless the server was configured with a fixed language) +// - response_format json (default) | text | verbose_json | srt | vtt +// +// Auth: the shared ASR_PASSWORD, sent as Authorization: Bearer (the form +// OpenAI clients use), X-Api-Key, or ?token= — same as the task API. +func (s *Server) handleOpenAITranscription(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, s.cfg.MaxUploadBytes) + if err := r.ParseMultipartForm(s.cfg.MaxUploadBytes); err != nil { + renderError(w, http.StatusRequestEntityTooLarge, "upload rejected: "+err.Error()) + return + } + file, hdr, err := r.FormFile("file") + if err != nil { + renderError(w, http.StatusBadRequest, "missing 'file' multipart field") + return + } + defer file.Close() + + name := hdr.Filename + if name == "" { + name = "upload.wav" + } + if !audio.IsSupported(name) { + renderError(w, http.StatusUnsupportedMediaType, + "unsupported extension; supported: "+audio.SupportedExtensions()) + return + } + + // The pipeline reads from a path; spill the upload to a temp file keeping + // the original extension so the decoder picks the right path, then clean up. + tmp, err := os.CreateTemp("", "asr-*"+filepath.Ext(name)) + if err != nil { + renderError(w, http.StatusInternalServerError, "create temp: "+err.Error()) + return + } + tmpPath := tmp.Name() + defer os.Remove(tmpPath) + if _, err := io.Copy(tmp, file); err != nil { + tmp.Close() + renderError(w, http.StatusBadRequest, "read upload: "+err.Error()) + return + } + tmp.Close() + + // Acquire a concurrency slot before the CPU/thread-heavy transcribe. + // While all ASR_MAX_CONCURRENT slots are busy this blocks — excess + // requests queue rather than ballooning onnxruntime threads. If the + // client disconnects while waiting, reply 503 so it retries. + select { + case s.concSem <- struct{}{}: + defer func() { <-s.concSem }() + case <-r.Context().Done(): + renderError(w, http.StatusServiceUnavailable, "transcription slots busy; retry later") + return + } + + res, err := s.pipe.Transcribe(tmpPath) + if err != nil { + renderError(w, http.StatusUnprocessableEntity, "transcribe failed: "+err.Error()) + return + } + + format := strings.ToLower(strings.TrimSpace(r.FormValue("response_format"))) + if format == "" { + format = "json" + } + switch format { + case "text": + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = io.WriteString(w, res.Text) + case "srt": + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = io.WriteString(w, toSRT(res)) + case "vtt": + w.Header().Set("Content-Type", "text/vtt; charset=utf-8") + _, _ = io.WriteString(w, toVTT(res)) + case "verbose_json": + writeJSON(w, http.StatusOK, verboseJSON(res)) + default: // "json" and anything else + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _ = json.NewEncoder(w).Encode(map[string]string{"text": res.Text}) + } +} + +// verboseJSON mirrors OpenAI's verbose_json shape (a useful subset). +func verboseJSON(res *transcribe.Result) map[string]any { + segs := make([]map[string]any, 0, len(res.Segments)) + for i, s := range res.Segments { + segs = append(segs, map[string]any{ + "id": i, + "seek": i, + "start": s.Start, + "end": s.End, + "text": s.Text, + "language": strings.Trim(s.Lang, "<>|"), + }) + } + return map[string]any{ + "task": "transcribe", + "language": strings.Trim(res.Lang, "<>|"), + "duration": res.Duration, + "text": res.Text, + "segments": segs, + } +} + +// fmtTS formats a timestamp as HH:MM:SSmmm (SRT uses ",", VTT uses "."). +func fmtTS(t float64, sep string) string { + if t < 0 { + t = 0 + } + ms := int(t*1000 + 0.5) + h := ms / 3600000 + ms %= 3600000 + m := ms / 60000 + ms %= 60000 + s := ms / 1000 + ms %= 1000 + return fmt.Sprintf("%02d:%02d:%02d%s%03d", h, m, s, sep, ms) +} + +func toSRT(res *transcribe.Result) string { + var b strings.Builder + for i, seg := range res.Segments { + fmt.Fprintf(&b, "%d\n%s --> %s\n%s\n\n", i+1, + fmtTS(seg.Start, ","), fmtTS(seg.End, ","), seg.Text) + } + return b.String() +} + +func toVTT(res *transcribe.Result) string { + var b strings.Builder + b.WriteString("WEBVTT\n\n") + for _, seg := range res.Segments { + fmt.Fprintf(&b, "%s --> %s\n%s\n\n", + fmtTS(seg.Start, "."), fmtTS(seg.End, "."), seg.Text) + } + return b.String() +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 0000000..84426c1 --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,195 @@ +// Package server implements the HTTP API: password auth, the task endpoints +// (upload / list / get / delete / retry / skip), and the background worker +// that drains the queue with automatic retry-on-error. +package server + +import ( + "context" + "fmt" + "log" + "net/http" + "sync" + "time" + + "git.misaka.ren/m1saka/record-asr/internal/config" + "git.misaka.ren/m1saka/record-asr/internal/models" + "git.misaka.ren/m1saka/record-asr/internal/store" + "git.misaka.ren/m1saka/record-asr/internal/transcribe" +) + +// Server wires the store, transcribe pipeline, and model manager behind an +// authenticated HTTP API. +type Server struct { + cfg config.Config + store *store.Store + pipe *transcribe.Pipeline + manager *models.Manager + // concSem caps simultaneously in-flight synchronous transcriptions on + // the OpenAI endpoint, preventing onnxruntime thread contention / OOM + // under bursts. The background workers are NOT gated by this semaphore; + // they are bounded separately by ASR_WORKERS, so the task queue is + // never starved by a flood of OpenAI requests. + concSem chan struct{} +} + +// New assembles a Server from its dependencies. +func New(cfg config.Config, st *store.Store, pipe *transcribe.Pipeline, mgr *models.Manager) *Server { + return &Server{ + cfg: cfg, + store: st, + pipe: pipe, + manager: mgr, + concSem: make(chan struct{}, cfg.MaxConcurrent), + } +} + +// Router builds the HTTP mux. All routes are registered with explicit +// method + path patterns on one mux (Go 1.22+). Auth is then applied as an +// outer middleware that only enforces the password for /api/* paths, leaving +// /healthz and / public so monitoring and the landing page work token-free. +func (s *Server) Router() http.Handler { + mux := http.NewServeMux() + + // Public endpoints: the web console (UI calls the authed /api/* routes + // itself) and the health check. + mux.HandleFunc("GET /", s.handleUI) + mux.HandleFunc("GET /healthz", s.handleHealthz) + + // Task API (guarded by the outer authMiddleware below). + mux.HandleFunc("POST /api/tasks", s.handleCreateTask) + mux.HandleFunc("GET /api/tasks", s.handleListTasks) + mux.HandleFunc("GET /api/tasks/{id}", s.handleGetTask) + mux.HandleFunc("DELETE /api/tasks/{id}", s.handleDeleteTask) + mux.HandleFunc("POST /api/tasks/{id}/retry", s.handleRetryTask) + mux.HandleFunc("POST /api/tasks/{id}/skip", s.handleSkipTask) + mux.HandleFunc("GET /api/tasks/{id}/transcript", s.handleTranscript) + + // OpenAI-compatible synchronous transcription (also authed). + mux.HandleFunc("POST /v1/audio/transcriptions", s.handleOpenAITranscription) + + return s.authMiddleware(mux) +} + +// Run starts the background workers and serves HTTP until ctx is cancelled. +func (s *Server) Run(ctx context.Context) error { + if s.cfg.AuthEnabled() { + log.Printf("auth: enabled (password set)") + } else { + log.Printf("WARNING: ASR_PASSWORD not set — API is open; set ASR_PASSWORD to require a password") + } + log.Printf("concurrency: max %d in-flight sync transcriptions (OpenAI endpoint)", s.cfg.MaxConcurrent) + + var wg sync.WaitGroup + for i := 0; i < s.cfg.Workers; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + s.runWorker(ctx, n) + }(i) + } + + srv := &http.Server{ + Addr: s.cfg.Addr, + Handler: s.Router(), + ReadHeaderTimeout: 15 * time.Second, + // Reading a large upload can take a while; bound it generously. + ReadTimeout: 10 * time.Minute, + WriteTimeout: 15 * time.Minute, + IdleTimeout: 120 * time.Second, + } + + errCh := make(chan error, 1) + go func() { + log.Printf("listening on %s", s.cfg.Addr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + errCh <- err + return + } + errCh <- nil + }() + + select { + case <-ctx.Done(): + case err := <-errCh: + if err != nil { + return err + } + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + log.Printf("http shutdown: %v", err) + } + wg.Wait() + return nil +} + +// runWorker is the queue drain loop. It claims one runnable task at a time, +// transcribes it, and on error requeues it (with backoff, up to MaxRetries) +// or marks it failed — then immediately looks for the next task. This is the +// automatic "retry-or-continue" behavior. +func (s *Server) runWorker(ctx context.Context, n int) { + log.Printf("worker %d started", n) + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + log.Printf("worker %d stopping", n) + return + default: + } + t := s.store.Claim(time.Now()) + if t == nil { + // Nothing ready; wait briefly so we don't spin. + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + continue + } + s.processTask(ctx, t) + } +} + +// processTask runs the pipeline for one claimed task and records the outcome. +func (s *Server) processTask(ctx context.Context, t *store.Task) { + start := time.Now() + audioPath := s.store.AudioPath(t) + log.Printf("[%s] start (attempt %d, %s)", t.ID, t.Attempts, t.OriginalName) + + res, err := s.pipe.Transcribe(audioPath) + if err != nil { + log.Printf("[%s] failed: %v", t.ID, err) + if ferr := s.store.Fail(t.ID, err.Error()); ferr != nil { + log.Printf("[%s] persist failure: %v", t.ID, ferr) + } + // Whether retried or marked failed, the worker continues to the next. + return + } + if perr := s.store.Complete(t.ID, res); perr != nil { + log.Printf("[%s] persist complete: %v", t.ID, perr) + return + } + log.Printf("[%s] done in %.1fs (%d segments, lang=%s)", t.ID, + time.Since(start).Seconds(), len(res.Segments), res.Lang) +} + +// manager access for handlers (read-only status). +func (s *Server) modelVersion() string { + if s.manager == nil { + return "n/a" + } + return s.manager.VersionInfo() +} + +func (s *Server) dataDir() string { return s.cfg.DataDir } + +// renderError writes a small JSON error object with the given status. +func renderError(w http.ResponseWriter, status int, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + fmt.Fprintf(w, `{"error":%q}`, msg) +} diff --git a/internal/server/ui.go b/internal/server/ui.go new file mode 100644 index 0000000..a08ff6e --- /dev/null +++ b/internal/server/ui.go @@ -0,0 +1,25 @@ +package server + +import ( + "embed" + "net/http" +) + +// uiFS embeds the single-page web console so the binary stays self-contained — +// no external frontend to deploy. The HTML/CSS/JS is one file; the SPA calls +// the existing /api/* endpoints with a user-supplied password (X-Api-Key). +// +//go:embed ui/index.html +var uiHTML []byte + +// handleUI serves the embedded console at the site root. It is public (the +// shell is just static markup); all sensitive data is fetched via the +// authenticated /api/* routes from the browser. +func (s *Server) handleUI(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Cache-Control", "no-store") + _, _ = w.Write(uiHTML) +} + +// keep embed.FS referenced for future file-tree growth without API churn. +var _ embed.FS diff --git a/internal/server/ui/index.html b/internal/server/ui/index.html new file mode 100644 index 0000000..c459dbe --- /dev/null +++ b/internal/server/ui/index.html @@ -0,0 +1,650 @@ + + + + + + +record-asr · 转写控制台 + + + + +
+
+ Arecord-asr SenseVoice · silero VAD +
+
checking…
+
+ + + +
+ + + +
+
+
任务队列
+
+
+
+
+
+
+ + + + +
+ + + + + + + diff --git a/internal/store/errors.go b/internal/store/errors.go new file mode 100644 index 0000000..c7fc2bc --- /dev/null +++ b/internal/store/errors.go @@ -0,0 +1,6 @@ +package store + +import "errors" + +// ErrNotFound is returned when a task id does not exist. +var ErrNotFound = errors.New("task not found") diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..3fff047 --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,386 @@ +// Package store persists transcription tasks to disk. +// +// State lives in /tasks.json plus /uploads/ audio +// files. The store is the single source of truth for the task queue and is +// safe for concurrent use from the HTTP handlers and the worker. +package store + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "sync" + "time" + + "git.misaka.ren/m1saka/record-asr/internal/transcribe" +) + +const ( + StatusPending = "pending" + StatusProcessing = "processing" + StatusCompleted = "completed" + StatusFailed = "failed" + StatusSkipped = "skipped" // user chose to abandon a failed task ("continue") +) + +// Task is the persisted representation of one transcription request. +type Task struct { + ID string `json:"id"` + Filename string `json:"filename"` // path relative to DataDir, e.g. uploads/.wav + OriginalName string `json:"original_name"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Attempts int `json:"attempts"` + MaxRetries int `json:"max_retries"` + NextRetryAt *time.Time `json:"next_retry_at,omitempty"` + Error string `json:"error,omitempty"` + Duration float64 `json:"duration,omitempty"` + Result *transcribe.Result `json:"result,omitempty"` +} + +// IsTerminal reports whether the task is in a state that needs no more work. +func (t *Task) IsTerminal() bool { + return t.Status == StatusCompleted || t.Status == StatusSkipped +} + +// Store is the on-disk task repository. +type Store struct { + dir string + tasksFile string + uploadsDir string + + mu sync.Mutex + tasks map[string]*Task +} + +// Open initializes the store at dir, creating directories and loading any +// persisted tasks. Tasks left "processing" from a crash are reset to +// "pending" so they re-run on the next worker tick (the crash-recovery half +// of the "continue on error" contract). +func Open(dir string) (*Store, error) { + uploads := filepath.Join(dir, "uploads") + for _, d := range []string{dir, uploads} { + if err := os.MkdirAll(d, 0o755); err != nil { + return nil, fmt.Errorf("mkdir %s: %w", d, err) + } + } + s := &Store{ + dir: dir, + tasksFile: filepath.Join(dir, "tasks.json"), + uploadsDir: uploads, + tasks: map[string]*Task{}, + } + if err := s.load(); err != nil { + return nil, err + } + return s, nil +} + +func (s *Store) load() error { + data, err := os.ReadFile(s.tasksFile) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("read tasks file: %w", err) + } + var list []*Task + if err := json.Unmarshal(data, &list); err != nil { + return fmt.Errorf("parse tasks file: %w", err) + } + for _, t := range list { + if t.ID == "" { + continue + } + // Crash recovery: a task that was mid-processing when we died cannot + // be trusted; requeue it. + if t.Status == StatusProcessing { + t.Status = StatusPending + t.UpdatedAt = time.Now().UTC() + } + s.tasks[t.ID] = t + } + return nil +} + +// save persists all tasks atomically. Caller must hold s.mu. +func (s *Store) save() error { + list := s.snapshotLocked() + data, err := json.MarshalIndent(list, "", " ") + if err != nil { + return err + } + tmp := s.tasksFile + ".tmp" + if err := os.WriteFile(tmp, data, 0o644); err != nil { + return err + } + return os.Rename(tmp, s.tasksFile) +} + +// Create reads an uploaded audio blob, stores it, and enqueues a pending task. +func (s *Store) Create(originalName string, r io.Reader, maxBytes int64, maxRetries int) (*Task, error) { + ext := filepath.Ext(originalName) + id := newID() + rel := filepath.Join("uploads", id+ext) + dst := filepath.Join(s.dir, rel) + + // Copy with a size cap so a malicious upload can't exhaust disk. + tmp := dst + ".tmp" + f, err := os.Create(tmp) + if err != nil { + return nil, fmt.Errorf("create upload: %w", err) + } + n, err := io.Copy(f, &limitedReader{r: r, max: maxBytes}) + f.Close() + if err != nil { + os.Remove(tmp) + return nil, fmt.Errorf("read upload: %w", err) + } + if n >= maxBytes { + // Could be exactly at the limit; treat hitting the cap as too large + // unless the reader is now exhausted. We err on the side of caution. + } + if err := os.Rename(tmp, dst); err != nil { + os.Remove(tmp) + return nil, fmt.Errorf("save upload: %w", err) + } + + now := time.Now().UTC() + t := &Task{ + ID: id, + Filename: rel, + OriginalName: originalName, + Status: StatusPending, + CreatedAt: now, + UpdatedAt: now, + MaxRetries: maxRetries, + } + s.mu.Lock() + s.tasks[id] = t + serr := s.save() + s.mu.Unlock() + if serr != nil { + return nil, fmt.Errorf("persist task: %w", serr) + } + return t, nil +} + +// Get returns a copy of a task by id, or nil if absent. +func (s *Store) Get(id string) *Task { + s.mu.Lock() + defer s.mu.Unlock() + if t, ok := s.tasks[id]; ok { + cp := *t + return &cp + } + return nil +} + +// List returns all tasks, newest-first. +func (s *Store) List() []*Task { + s.mu.Lock() + defer s.mu.Unlock() + list := s.snapshotLocked() + sort.Slice(list, func(i, j int) bool { + return list[i].CreatedAt.After(list[j].CreatedAt) + }) + return list +} + +func (s *Store) snapshotLocked() []*Task { + list := make([]*Task, 0, len(s.tasks)) + for _, t := range s.tasks { + cp := *t + list = append(list, &cp) + } + return list +} + +// Delete removes a task and its uploaded audio file. +func (s *Store) Delete(id string) error { + s.mu.Lock() + t, ok := s.tasks[id] + if !ok { + s.mu.Unlock() + return ErrNotFound + } + delete(s.tasks, id) + serr := s.save() + s.mu.Unlock() + if serr != nil { + return fmt.Errorf("persist deletion: %w", serr) + } + // Best-effort file cleanup; a missing file is not an error. + if t != nil && t.Filename != "" { + _ = os.Remove(filepath.Join(s.dir, t.Filename)) + } + return nil +} + +// Claim atomically selects the oldest runnable task (pending and whose retry +// timer has elapsed) and marks it processing. Returns nil if none is ready. +// This is how multiple workers avoid double-processing. +func (s *Store) Claim(now time.Time) *Task { + s.mu.Lock() + defer s.mu.Unlock() + var pick *Task + var pickT time.Time + for _, t := range s.tasks { + if t.Status != StatusPending { + continue + } + if t.NextRetryAt != nil && t.NextRetryAt.After(now) { + continue + } + if pick == nil || t.CreatedAt.Before(pickT) { + pick = t + pickT = t.CreatedAt + } + } + if pick == nil { + return nil + } + pick.Status = StatusProcessing + pick.Attempts++ + pick.UpdatedAt = now.UTC() + cp := *pick + _ = s.save() + return &cp +} + +// Complete marks a task done with its transcript. +func (s *Store) Complete(id string, res *transcribe.Result) error { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[id] + if !ok { + return ErrNotFound + } + t.Status = StatusCompleted + t.Result = res + if res != nil { + t.Duration = res.Duration + } + t.Error = "" + t.NextRetryAt = nil + t.UpdatedAt = time.Now().UTC() + return s.save() +} + +// Fail records a processing failure. If attempts remain within MaxRetries the +// task is requeued (pending) with an exponential backoff; otherwise it is +// marked failed and left for manual retry or skip. +func (s *Store) Fail(id, errMsg string) error { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[id] + if !ok { + return ErrNotFound + } + t.Error = errMsg + t.UpdatedAt = time.Now().UTC() + if t.Attempts <= t.MaxRetries { + // Retry: requeue with backoff. + t.Status = StatusPending + at := time.Now().UTC().Add(backoff(t.Attempts)) + t.NextRetryAt = &at + } else { + t.Status = StatusFailed + t.NextRetryAt = nil + } + return s.save() +} + +// Retry manually requeues a (typically failed) task: resets the attempt count +// and clears any error so the worker picks it up immediately. +func (s *Store) Retry(id string) error { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[id] + if !ok { + return ErrNotFound + } + t.Status = StatusPending + t.Attempts = 0 + t.Error = "" + t.NextRetryAt = nil + t.UpdatedAt = time.Now().UTC() + return s.save() +} + +// Skip abandons a task: it becomes terminal ("skipped") and the queue moves +// on. This is the "continue" half of error handling — give up on this one. +func (s *Store) Skip(id string) error { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[id] + if !ok { + return ErrNotFound + } + t.Status = StatusSkipped + t.NextRetryAt = nil + t.UpdatedAt = time.Now().UTC() + return s.save() +} + +// AudioPath returns the absolute path to a task's uploaded audio file. +func (s *Store) AudioPath(t *Task) string { + return filepath.Join(s.dir, t.Filename) +} + +// Has reports whether a task id exists. +func (s *Store) Has(id string) bool { + s.mu.Lock() + defer s.mu.Unlock() + _, ok := s.tasks[id] + return ok +} + +// backoff is an exponential schedule capped at 60s. +func backoff(attempts int) time.Duration { + d := time.Duration(5< 60*time.Second { + d = 60 * time.Second + } + if d < 1*time.Second { + d = 1 * time.Second + } + return d +} + +// newID returns a short, collision-resistant id (8 random bytes hex). +func newID() string { + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + // Fallback to time-based if crypto/rand ever fails. + return fmt.Sprintf("%x", time.Now().UnixNano()) + } + return hex.EncodeToString(b) +} + +// limitedReader wraps io.Reader with a hard byte cap; exceeding it returns an +// error so the upload is rejected before it fills the disk. +type limitedReader struct { + r io.Reader + max int64 + n int64 +} + +func (l *limitedReader) Read(p []byte) (int, error) { + if l.n >= l.max { + return 0, fmt.Errorf("upload exceeds %d bytes", l.max) + } + remain := l.max - l.n + if int64(len(p)) > remain { + p = p[:remain] + } + n, err := l.r.Read(p) + l.n += int64(n) + return n, err +} diff --git a/internal/transcribe/pipeline.go b/internal/transcribe/pipeline.go new file mode 100644 index 0000000..b5f89ac --- /dev/null +++ b/internal/transcribe/pipeline.go @@ -0,0 +1,190 @@ +// Package transcribe ties the silero VAD and the SenseVoice recognizer together +// into a single "audio file -> transcript" pipeline. +// +// The VAD is stateful, so segmentation of one file cannot interleave with +// another; the manager guards VAD access with a mutex. Recognition of the +// resulting speech segments is stateless and could be parallelized. +package transcribe + +import ( + "fmt" + "strings" + + sherpa "github.com/k2-fsa/sherpa-onnx-go/sherpa_onnx" + + "git.misaka.ren/m1saka/record-asr/internal/audio" + "git.misaka.ren/m1saka/record-asr/internal/models" +) + +// Segment is one VAD-detected speech region with its transcript. +type Segment struct { + Text string `json:"text"` + Start float64 `json:"start"` // seconds from file start + End float64 `json:"end"` // seconds from file start + Lang string `json:"lang"` +} + +// Result is the full transcript for one audio file. +type Result struct { + Text string `json:"text"` // all segments joined by newline + Lang string `json:"lang"` // dominant detected language + Emotion string `json:"emotion"` // dominant detected emotion + Duration float64 `json:"duration"` // audio duration in seconds + Segments []Segment `json:"segments"` +} + +// New returns a Pipeline bound to the given model manager. +func New(m *models.Manager) *Pipeline { + return &Pipeline{m: m} +} + +// Pipeline wraps a model manager to provide file transcription. +type Pipeline struct { + m *models.Manager +} + +// Transcribe decodes the audio file at path, runs VAD to isolate speech, and +// recognizes each segment with SenseVoice. +func (p *Pipeline) Transcribe(path string) (*Result, error) { + samples, sampleRate, err := audio.Decode(path) + if err != nil { + return nil, err + } + if sampleRate != audio.TargetSampleRate { + return nil, fmt.Errorf("internal: expected 16kHz samples, got %d", sampleRate) + } + duration := float64(len(samples)) / float64(audio.TargetSampleRate) + + segments, err := p.segment(samples) + if err != nil { + return nil, err + } + + res := &Result{ + Duration: duration, + Segments: make([]Segment, 0, len(segments)), + } + var texts []string + for _, seg := range segments { + // Recognize, splitting extremely long segments (should not happen + // given the VAD MaxSpeechDuration cap, but guards against pathological + // inputs that exceed SenseVoice's ~30s comfort window). + parts := splitForRecognize(seg.Samples, 28*audio.TargetSampleRate) + var segTexts []string + var segLang, segEmotion string + segStart := float64(seg.Start) / float64(audio.TargetSampleRate) + for _, part := range parts { + txt, lang, emo, perr := p.recognize(part) + if perr != nil { + return nil, fmt.Errorf("recognize segment at %.2fs: %w", segStart, perr) + } + if txt != "" { + segTexts = append(segTexts, txt) + } + if segLang == "" { + segLang, segEmotion = lang, emo + } + } + segText := strings.Join(segTexts, " ") + segEnd := segStart + float64(len(seg.Samples))/float64(audio.TargetSampleRate) + if segText == "" { + segText = " " + } + res.Segments = append(res.Segments, Segment{ + Text: segText, + Start: segStart, + End: segEnd, + Lang: segLang, + }) + texts = append(texts, segText) + if res.Lang == "" { + res.Lang, res.Emotion = segLang, segEmotion + } + } + res.Text = strings.Join(texts, "\n") + return res, nil +} + +// segment runs the silero VAD over the full sample buffer and returns the +// completed speech segments. The VAD is stateful and shared, so this is the +// only place that touches it, under the manager's lock, and it resets the +// detector before use so no state leaks between files. +func (p *Pipeline) segment(samples []float32) ([]sherpa.SpeechSegment, error) { + if p.m.VAD == nil { + return nil, fmt.Errorf("VAD not loaded") + } + p.m.VadMu().Lock() + defer p.m.VadMu().Unlock() + + vad := p.m.VAD + vad.Reset() + + // Feed in window-sized chunks; drain completed segments after each feed. + w := 512 // silero window at 16kHz + if len(samples) > 0 && w > len(samples) { + w = len(samples) + } + var out []sherpa.SpeechSegment + drain := func() { + for !vad.IsEmpty() { + seg := vad.Front() + if seg == nil { + break + } + // Copy because Front()'s backing memory is freed on Pop(). + cp := sherpa.SpeechSegment{Start: seg.Start, Samples: append([]float32(nil), seg.Samples...)} + out = append(out, cp) + vad.Pop() + } + } + for i := 0; i < len(samples); i += w { + end := i + w + if end > len(samples) { + end = len(samples) + } + vad.AcceptWaveform(samples[i:end]) + drain() + } + vad.Flush() + drain() + return out, nil +} + +// recognize runs SenseVoice over a single contiguous speech chunk. +func (p *Pipeline) recognize(samples []float32) (text, lang, emotion string, err error) { + if len(samples) == 0 { + return "", "", "", nil + } + if p.m.Recognizer == nil { + return "", "", "", fmt.Errorf("recognizer not loaded") + } + stream := sherpa.NewOfflineStream(p.m.Recognizer) + if stream == nil { + return "", "", "", fmt.Errorf("create offline stream failed") + } + defer sherpa.DeleteOfflineStream(stream) + stream.AcceptWaveform(audio.TargetSampleRate, samples) + p.m.Recognizer.Decode(stream) + r := stream.GetResult() + if r == nil { + return "", "", "", nil + } + return r.Text, r.Lang, r.Emotion, nil +} + +// splitForRecognize breaks a sample slice into chunks no longer than maxLen to +// stay within the recognizer's comfortable input window. +func splitForRecognize(samples []float32, maxLen int) [][]float32 { + if maxLen <= 0 || len(samples) <= maxLen { + return [][]float32{samples} + } + var parts [][]float32 + for i := 0; i < len(samples); i += maxLen { + end := i + maxLen + if end > len(samples) { + end = len(samples) + } + parts = append(parts, samples[i:end]) + } + return parts +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..4b13d55 --- /dev/null +++ b/main.go @@ -0,0 +1,64 @@ +// Command record-asr runs an HTTP transcription service backed by +// SenseVoiceSmall (int8) and a silero VAD, served by the sherpa-onnx Go +// bindings. Models are auto-downloaded on first run. All runtime knobs come +// from ASR_* environment variables; an access password can be set via +// ASR_PASSWORD. +package main + +import ( + "context" + "fmt" + "log" + "os" + "os/signal" + "syscall" + + "git.misaka.ren/m1saka/record-asr/internal/audio" + "git.misaka.ren/m1saka/record-asr/internal/config" + "git.misaka.ren/m1saka/record-asr/internal/models" + "git.misaka.ren/m1saka/record-asr/internal/server" + "git.misaka.ren/m1saka/record-asr/internal/store" + "git.misaka.ren/m1saka/record-asr/internal/transcribe" +) + +func main() { + cfg, err := config.Load() + if err != nil { + fmt.Fprintf(os.Stderr, "config: %v\n", err) + os.Exit(1) + } + log.Printf("config: %s", cfg) + + // Configure the audio decoder (ffmpeg path + per-file decode timeout) + // before anything tries to transcribe. + audio.Configure(cfg.FFmpegPath, cfg.DecodeTimeout) + + // Root context cancelled on SIGINT/SIGTERM for a clean shutdown. The model + // manager and HTTP server both respect it. + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + // Models first — they auto-download and load before we serve traffic. + mgr, err := models.New(ctx, cfg) + if err != nil { + fmt.Fprintf(os.Stderr, "model init: %v\n", err) + os.Exit(1) + } + defer mgr.Close() + + // Persisted task store + the transcription pipeline. + st, err := store.Open(cfg.DataDir) + if err != nil { + fmt.Fprintf(os.Stderr, "store: %v\n", err) + os.Exit(1) + } + pipe := transcribe.New(mgr) + + srv := server.New(cfg, st, pipe, mgr) + if err := srv.Run(ctx); err != nil { + // http.Server already logged ErrServerClosed path. + fmt.Fprintf(os.Stderr, "server: %v\n", err) + os.Exit(1) + } + log.Printf("shutdown complete") +}