diff --git a/.env.example b/.env.example index 92f9413..baeba14 100644 --- a/.env.example +++ b/.env.example @@ -7,3 +7,25 @@ DATABASE_MAX_CONNECTIONS=5 # A single development origin. Leave unset to disable CORS middleware. # CORS_ALLOWED_ORIGIN=https://servicewechat.com + +# Private server directory used to retain uploaded audio for transcription retry. +AUDIO_STORAGE_DIR=./data/audio + +# Local FunASR is the default Chinese transcription provider. +ASR_PROVIDER=local +LOCAL_ASR_URL=http://127.0.0.1:10095 +ASR_REQUEST_TIMEOUT_SECONDS=1800 +ASR_WORKER_CONCURRENCY=1 +ASR_JOB_LEASE_SECONDS=3600 + +# Optional Tencent Cloud fallback. Set ASR_PROVIDER=tencent and configure all values. +# TENCENT_CLOUD_ASR_APP_ID=1234567890 +# TENCENT_CLOUD_ASR_SECRET_ID=your-secret-id +# TENCENT_CLOUD_ASR_SECRET_KEY=your-secret-key +TENCENT_CLOUD_ASR_ENGINE_TYPE=16k_zh + +# Optional OpenAI-compatible chat completions endpoint for polished feedback summaries. +# Without it, the API uses a local extractive summary and remains functional. +# FEEDBACK_SUMMARY_API_URL=https://api.example.com/v1/chat/completions +# FEEDBACK_SUMMARY_API_KEY=your-api-key +# FEEDBACK_SUMMARY_MODEL=your-model diff --git a/.gitignore b/.gitignore index ad36eaf..c10ee8c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Rust build output /target/ +/data/ # Local configuration and secrets .env diff --git a/API.md b/API.md index c10caa6..0289534 100644 --- a/API.md +++ b/API.md @@ -26,6 +26,39 @@ cargo run --bin migrate cargo run ``` +## 语音转录和反馈汇总 + +长短录音统一通过小程序原生录音器上传到本服务。服务先将音频保存在 `AUDIO_STORAGE_DIR` 并立即响应，再由持久化后台队列调用本地 FunASR；模型暂时不可用或 API 重启都不会丢失已上传音频。 + +```text +AUDIO_STORAGE_DIR=./data/audio +ASR_PROVIDER=local +LOCAL_ASR_URL=http://127.0.0.1:10095 +ASR_REQUEST_TIMEOUT_SECONDS=1800 +ASR_WORKER_CONCURRENCY=1 +ASR_JOB_LEASE_SECONDS=3600 +``` + +在项目根目录执行 `docker compose -f compose.asr.yml up --build -d` 启动中文模型。`GET /health` 中 `speech_available=true` 表示模型已下载并可以接收任务。 + +腾讯云仅作为可选提供方保留。需要切换时配置： + +```text +ASR_PROVIDER=tencent +TENCENT_CLOUD_ASR_APP_ID=1234567890 +TENCENT_CLOUD_ASR_SECRET_ID=your-secret-id +TENCENT_CLOUD_ASR_SECRET_KEY=your-secret-key +TENCENT_CLOUD_ASR_ENGINE_TYPE=16k_zh +``` + +反馈汇总支持可选的 OpenAI 兼容 Chat Completions 接口。未配置时使用本地抽取式汇总，录音转录和保存流程仍可工作： + +```text +FEEDBACK_SUMMARY_API_URL=https://api.example.com/v1/chat/completions +FEEDBACK_SUMMARY_API_KEY=your-api-key +FEEDBACK_SUMMARY_MODEL=your-model +``` + ## 临时身份边界 目前所有业务接口都要求 `X-User-Id` 请求头，值为 UUID。例如： @@ -40,7 +73,7 @@ X-User-Id: 5a4da7f0-d70c-465f-bcab-124c504aa9f0 | 方法 | 路径 | 说明 | | --- | --- | --- | -| `GET` | `/health` | 服务状态及数据库是否已配置。 | +| `GET` | `/health` | 服务、数据库、语音提供方及模型可用状态。 | | `GET` | `/api/v1/profiles` | 列出学生档案；可选 `query`、`grade`、`subject` 参数。 | | `POST` | `/api/v1/profiles` | 创建学生档案。 | | `GET` | `/api/v1/profiles/{profile_id}` | 读取单个学生档案。 | @@ -51,6 +84,15 @@ X-User-Id: 5a4da7f0-d70c-465f-bcab-124c504aa9f0 | `GET` | `/api/v1/feedback-records` | 列出反馈记录；可选 `profile_id` 参数。 | | `POST` | `/api/v1/feedback-records` | 创建反馈记录。 | | `DELETE` | `/api/v1/feedback-records/{record_id}` | 删除反馈记录。 | +| `GET` | `/api/v1/feedback-sessions/active` | 读取指定学生进行中的语音反馈批次。 | +| `POST` | `/api/v1/feedback-sessions` | 创建或恢复语音反馈批次。 | +| `PUT` | `/api/v1/feedback-sessions/{session_id}` | 自动保存反馈正文和日期。 | +| `POST` | `/api/v1/feedback-sessions/{session_id}/lessons` | 开始下一节课堂录音。 | +| `POST` | `/api/v1/feedback-lessons/{lesson_id}/segments` | 上传一个自动切分的片段并加入后台转录队列。 | +| `POST` | `/api/v1/feedback-lessons/{lesson_id}/finish` | 结束课节并汇总片段状态。 | +| `POST` | `/api/v1/feedback-lessons/{lesson_id}/mark-applied` | 标记短录音文字已直接加入反馈。 | +| `POST` | `/api/v1/audio-segments/{segment_id}/retry` | 重试失败片段的转录。 | +| `POST` | `/api/v1/feedback-sessions/{session_id}/generate` | 汇总多节转录并生成反馈正文。 | ### 创建学生档案 @@ -83,6 +125,7 @@ X-User-Id: 5a4da7f0-d70c-465f-bcab-124c504aa9f0 { "profile_id": "5fa5af2c-cfc9-4bb8-b0b3-8715f080b167", "feedback_date": "2026-07-14", - "content": "今天完成了色彩搭配练习，构图有明显进步。" + "content": "今天完成了色彩搭配练习，构图有明显进步。", + "feedback_session_id": null } ``` diff --git a/API_GUIDE.md b/API_GUIDE.md index b1dbca9..3e1af0f 100644 --- a/API_GUIDE.md +++ b/API_GUIDE.md @@ -2,6 +2,12 @@ ## 1. 启动服务 +先在项目根目录启动中文转录服务： + +```bash +docker compose -f compose.asr.yml up --build -d +``` + 在 `server/.env` 中配置 `DATABASE_URL` 后执行： ```bash @@ -28,12 +34,12 @@ X-User-Id: 5a4da7f0-d70c-465f-bcab-124c504aa9f0 请求体接口也已提供可发送的默认 JSON。建议按以下顺序测试： -1. 执行 `GET /health`，确认 `database_configured` 为 `true`。 +1. 执行 `GET /health`，确认 `database_configured` 和 `speech_available` 均为 `true`。首次模型下载期间后者为 `false`。 2. 执行 `GET /api/v1/profile-defaults`，即使用户未保存过预设也会返回系统默认值。 3. 执行 `PUT /api/v1/profile-defaults`，可直接发送预填的美术课预设。 4. 执行 `POST /api/v1/profiles`，可直接发送预填档案；保存响应中的 `id`。 5. 将真实档案 `id` 填入 `{profile_id}`，测试单条查询、更新或删除。 -6. 执行 `POST /api/v1/feedback-records`。默认 `profile_id` 为 `null`，无需档案即可直接创建。 +6. 执行 `POST /api/v1/feedback-records`。默认 `profile_id` 和 `feedback_session_id` 为 `null`，无需档案或录音即可直接创建。 7. 如需关联档案，将第 4 步得到的 `id` 写入 `profile_id`。创建后保存反馈响应中的 `id`。 8. 将真实反馈 `id` 填入 `{record_id}`，测试删除接口。 @@ -74,7 +80,8 @@ curl -X POST http://127.0.0.1:8080/api/v1/feedback-records \ -d '{ "profile_id": null, "feedback_date": "2026-07-15", - "content": "今天完成了色彩搭配练习，构图有明显进步。" + "content": "今天完成了色彩搭配练习，构图有明显进步。", + "feedback_session_id": null }' ``` @@ -114,6 +121,8 @@ curl 'http://127.0.0.1:8080/api/v1/feedback-records?profile_id=<真实档案ID>' } ``` +录音上传接口在音频入库后立即返回 `processing`。Rust 后台工作线程会调用 FunASR，并将片段更新为 `ready` 或 `failed`；小程序会自动刷新状态，短录音转写完成后仍会直接加入反馈正文。 + ## 6. OpenAPI 维护方式 `/openapi.json` 在服务运行时由 `utoipa` 根据 Rust 路由注解生成，Scalar 使用同一份规范。新增或修改接口时，需要同时更新处理函数上的 `#[utoipa::path]`、请求/响应 schema 注释和示例；规范完整性测试会检查操作数量、接口说明和默认身份参数。 diff --git a/Cargo.lock b/Cargo.lock index 6b5d333..b79f207 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,6 +67,7 @@ dependencies = [ "matchit", "memchr", "mime", + "multer", "percent-encoding", "pin-project-lite", "serde_core", @@ -175,6 +176,23 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.45" @@ -219,6 +237,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc" version = "3.4.0" @@ -308,6 +335,15 @@ dependencies = [ "serde", ] +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -466,8 +502,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -477,8 +515,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi", + "rand_core 0.10.1", + "wasm-bindgen", ] [[package]] @@ -609,6 +650,23 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots 1.0.8", ] [[package]] @@ -617,13 +675,21 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", + "futures-channel", + "futures-util", "http", "http-body", "hyper", + "ipnet", + "libc", + "percent-encoding", "pin-project-lite", + "socket2", "tokio", "tower-service", + "tracing", ] [[package]] @@ -764,6 +830,12 @@ dependencies = [ "serde_core", ] +[[package]] +name = "ipnet" +version = "2.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" + [[package]] name = "itoa" version = "1.0.18" @@ -845,6 +917,12 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "matchers" version = "0.2.0" @@ -882,6 +960,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime_guess" +version = "2.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" +dependencies = [ + "mime", + "unicase", +] + [[package]] name = "mio" version = "1.2.2" @@ -893,6 +981,23 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "multer" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http", + "httparse", + "memchr", + "mime", + "spin", + "version_check", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -913,7 +1018,7 @@ dependencies = [ "num-integer", "num-iter", "num-traits", - "rand", + "rand 0.8.7", "smallvec", "zeroize", ] @@ -1069,6 +1174,62 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand 0.10.2", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.46" @@ -1092,7 +1253,18 @@ checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", "rand_chacha", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", ] [[package]] @@ -1102,7 +1274,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", ] [[package]] @@ -1114,6 +1286,21 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1149,6 +1336,46 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64", + "bytes", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "mime_guess", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots 1.0.8", +] + [[package]] name = "ring" version = "0.17.14" @@ -1176,13 +1403,19 @@ dependencies = [ "num-traits", "pkcs1", "pkcs8", - "rand_core", + "rand_core 0.6.4", "signature", "spki", "subtle", "zeroize", ] +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustls" version = "0.23.42" @@ -1203,6 +1436,7 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ + "web-time", "zeroize", ] @@ -1308,7 +1542,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -1319,7 +1553,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -1355,7 +1589,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ "digest", - "rand_core", + "rand_core 0.6.4", ] [[package]] @@ -1521,7 +1755,7 @@ dependencies = [ "memchr", "once_cell", "percent-encoding", - "rand", + "rand 0.8.7", "rsa", "serde", "sha1", @@ -1561,7 +1795,7 @@ dependencies = [ "md-5", "memchr", "once_cell", - "rand", + "rand 0.8.7", "serde", "serde_json", "sha2", @@ -1639,6 +1873,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -1656,16 +1893,21 @@ name = "teaching-feedback-api" version = "0.1.0" dependencies = [ "axum", + "base64", "chrono", "dotenvy", + "hmac", + "reqwest", "serde", "serde_json", + "sha1", "sqlx", "thiserror", "tokio", "tower-http", "tracing", "tracing-subscriber", + "url", "utoipa", "utoipa-axum", "utoipa-scalar", @@ -1753,6 +1995,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -1788,12 +2040,15 @@ checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ "bitflags", "bytes", + "futures-util", "http", "http-body", "pin-project-lite", + "tower", "tower-layer", "tower-service", "tracing", + "url", ] [[package]] @@ -1870,12 +2125,24 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "typenum" version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "unicase" +version = "2.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539" + [[package]] name = "unicode-bidi" version = "0.3.18" @@ -2006,6 +2273,15 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -2031,6 +2307,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.126" @@ -2063,6 +2349,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "0.26.11" diff --git a/Cargo.toml b/Cargo.toml index b540799..e976154 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,17 +6,22 @@ rust-version = "1.85" default-run = "teaching-feedback-api" [dependencies] -axum = { version = "0.8", features = ["macros"] } +axum = { version = "0.8", features = ["macros", "multipart"] } +base64 = "0.22" chrono = { version = "0.4", features = ["serde"] } dotenvy = "0.15" +hmac = "0.12" +reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +sha1 = "0.10" sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "postgres", "chrono", "uuid", "migrate", "macros"] } thiserror = "2" -tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "signal"] } +tokio = { version = "1", features = ["fs", "macros", "rt-multi-thread", "net", "signal"] } tower-http = { version = "0.6", features = ["cors", "trace"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +url = "2" utoipa = { version = "5.5", features = ["chrono", "uuid"] } utoipa-axum = "0.2" utoipa-scalar = { version = "0.3", features = ["axum"] } diff --git a/migrations/0002_voice_feedback_sessions.sql b/migrations/0002_voice_feedback_sessions.sql new file mode 100644 index 0000000..d0bd60f --- /dev/null +++ b/migrations/0002_voice_feedback_sessions.sql @@ -0,0 +1,57 @@ +CREATE TABLE feedback_sessions ( + id UUID PRIMARY KEY, + owner_id UUID NOT NULL, + profile_id UUID REFERENCES student_profiles(id) ON DELETE CASCADE, + feedback_date DATE NOT NULL, + content TEXT NOT NULL DEFAULT '', + status VARCHAR(16) NOT NULL DEFAULT 'active' + CHECK (status IN ('active', 'finalized')), + finalized_record_id UUID REFERENCES feedback_records(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX feedback_sessions_owner_status_idx + ON feedback_sessions (owner_id, status, updated_at DESC); + +CREATE UNIQUE INDEX feedback_sessions_one_active_idx + ON feedback_sessions (owner_id, COALESCE(profile_id, '00000000-0000-0000-0000-000000000000'::UUID)) + WHERE status = 'active'; + +CREATE TABLE lesson_sessions ( + id UUID PRIMARY KEY, + feedback_session_id UUID NOT NULL REFERENCES feedback_sessions(id) ON DELETE CASCADE, + sequence INTEGER NOT NULL CHECK (sequence > 0), + status VARCHAR(16) NOT NULL DEFAULT 'recording' + CHECK (status IN ('recording', 'processing', 'ready', 'failed')), + duration_ms INTEGER NOT NULL DEFAULT 0 CHECK (duration_ms >= 0), + applied_directly BOOLEAN NOT NULL DEFAULT FALSE, + included_in_generation BOOLEAN NOT NULL DEFAULT FALSE, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + ended_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (feedback_session_id, sequence) +); + +CREATE INDEX lesson_sessions_feedback_session_idx + ON lesson_sessions (feedback_session_id, sequence); + +CREATE TABLE audio_segments ( + id UUID PRIMARY KEY, + lesson_session_id UUID NOT NULL REFERENCES lesson_sessions(id) ON DELETE CASCADE, + sequence INTEGER NOT NULL CHECK (sequence > 0), + duration_ms INTEGER NOT NULL CHECK (duration_ms > 0), + audio_format VARCHAR(8) NOT NULL, + audio_path TEXT NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'processing' + CHECK (status IN ('processing', 'ready', 'failed')), + transcript TEXT, + error_message TEXT, + provider_request_id VARCHAR(100), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (lesson_session_id, sequence) +); + +CREATE INDEX audio_segments_lesson_session_idx + ON audio_segments (lesson_session_id, sequence); diff --git a/migrations/0003_async_transcription_jobs.sql b/migrations/0003_async_transcription_jobs.sql new file mode 100644 index 0000000..1f05673 --- /dev/null +++ b/migrations/0003_async_transcription_jobs.sql @@ -0,0 +1,8 @@ +ALTER TABLE audio_segments + ADD COLUMN transcription_attempts INTEGER NOT NULL DEFAULT 0 + CHECK (transcription_attempts >= 0), + ADD COLUMN processing_started_at TIMESTAMPTZ; + +CREATE INDEX audio_segments_processing_queue_idx + ON audio_segments (created_at, id) + WHERE status = 'processing'; diff --git a/src/config.rs b/src/config.rs index 7e34ed0..6cf8a67 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,4 +1,33 @@ -use std::{env, net::IpAddr}; +use std::{env, net::IpAddr, path::PathBuf}; + +#[derive(Debug, Clone)] +pub enum SpeechProviderConfig { + Local { api_url: String }, + Tencent(TencentSpeechConfig), +} + +#[derive(Debug, Clone)] +pub struct TencentSpeechConfig { + pub app_id: u64, + pub secret_id: String, + pub secret_key: String, + pub engine_type: String, +} + +#[derive(Debug, Clone)] +pub struct SpeechConfig { + pub provider: SpeechProviderConfig, + pub request_timeout_seconds: u64, + pub worker_concurrency: usize, + pub job_lease_seconds: i64, +} + +#[derive(Debug, Clone)] +pub struct SummaryConfig { + pub api_url: String, + pub api_key: Option, + pub model: String, +} #[derive(Debug, Clone)] pub struct Config { @@ -7,6 +36,9 @@ pub struct Config { pub database_url: Option, pub database_max_connections: u32, pub cors_allowed_origin: Option, + pub audio_storage_dir: PathBuf, + pub speech: Option, + pub summary: Option, } impl Config { @@ -24,6 +56,9 @@ impl Config { .parse() .map_err(|_| "DATABASE_MAX_CONNECTIONS must be a valid u32 value".to_owned())?; + let speech = speech_config()?; + let summary = summary_config()?; + Ok(Self { host, port, @@ -34,6 +69,99 @@ impl Config { cors_allowed_origin: env::var("CORS_ALLOWED_ORIGIN") .ok() .filter(|value| !value.trim().is_empty()), + audio_storage_dir: env::var("AUDIO_STORAGE_DIR") + .map(PathBuf::from) + .unwrap_or_else(|_| PathBuf::from("./data/audio")), + speech, + summary, }) } } + +fn speech_config() -> Result, String> { + let provider = optional_env("ASR_PROVIDER").unwrap_or_else(|| "local".to_owned()); + if matches!(provider.as_str(), "disabled" | "none") { + return Ok(None); + } + + let provider = match provider.as_str() { + "local" | "funasr" => SpeechProviderConfig::Local { + api_url: optional_env("LOCAL_ASR_URL") + .unwrap_or_else(|| "http://127.0.0.1:10095".to_owned()) + .trim_end_matches('/') + .to_owned(), + }, + "tencent" => { + let app_id = required_env("TENCENT_CLOUD_ASR_APP_ID")? + .parse() + .map_err(|_| "TENCENT_CLOUD_ASR_APP_ID must be an integer".to_owned())?; + SpeechProviderConfig::Tencent(TencentSpeechConfig { + app_id, + secret_id: required_env("TENCENT_CLOUD_ASR_SECRET_ID")?, + secret_key: required_env("TENCENT_CLOUD_ASR_SECRET_KEY")?, + engine_type: optional_env("TENCENT_CLOUD_ASR_ENGINE_TYPE") + .unwrap_or_else(|| "16k_zh".to_owned()), + }) + } + _ => { + return Err( + "ASR_PROVIDER must be local, funasr, tencent, disabled, or none".to_owned(), + ); + } + }; + + let request_timeout_seconds = parse_env("ASR_REQUEST_TIMEOUT_SECONDS", 1_800_u64)?; + let worker_concurrency = parse_env("ASR_WORKER_CONCURRENCY", 1_usize)?; + let job_lease_seconds = parse_env("ASR_JOB_LEASE_SECONDS", 3_600_i64)?; + if request_timeout_seconds == 0 { + return Err("ASR_REQUEST_TIMEOUT_SECONDS must be greater than 0".to_owned()); + } + if worker_concurrency == 0 { + return Err("ASR_WORKER_CONCURRENCY must be greater than 0".to_owned()); + } + if job_lease_seconds <= 0 { + return Err("ASR_JOB_LEASE_SECONDS must be greater than 0".to_owned()); + } + + Ok(Some(SpeechConfig { + provider, + request_timeout_seconds, + worker_concurrency, + job_lease_seconds, + })) +} + +fn summary_config() -> Result, String> { + let Some(api_url) = optional_env("FEEDBACK_SUMMARY_API_URL") else { + return Ok(None); + }; + let model = optional_env("FEEDBACK_SUMMARY_MODEL").ok_or_else(|| { + "FEEDBACK_SUMMARY_MODEL is required when summary API is configured".to_owned() + })?; + Ok(Some(SummaryConfig { + api_url: api_url.trim_end_matches('/').to_owned(), + api_key: optional_env("FEEDBACK_SUMMARY_API_KEY"), + model, + })) +} + +fn optional_env(name: &str) -> Option { + env::var(name) + .ok() + .map(|value| value.trim().to_owned()) + .filter(|value| !value.is_empty()) +} + +fn required_env(name: &str) -> Result { + optional_env(name).ok_or_else(|| format!("{name} is required for the selected ASR provider")) +} + +fn parse_env(name: &str, default: T) -> Result +where + T: std::str::FromStr, +{ + optional_env(name).map_or_else( + || Ok(default), + |value| value.parse().map_err(|_| format!("{name} is invalid")), + ) +} diff --git a/src/error.rs b/src/error.rs index 2ec8e8d..3d821ac 100644 --- a/src/error.rs +++ b/src/error.rs @@ -16,6 +16,8 @@ pub enum ApiError { NotFound, #[error("database has not been configured")] DatabaseUnavailable, + #[error("{0}")] + ServiceUnavailable(String), #[error("internal server error")] Internal, } @@ -75,6 +77,7 @@ impl IntoResponse for ApiError { StatusCode::SERVICE_UNAVAILABLE, "database has not been configured".to_owned(), ), + Self::ServiceUnavailable(message) => (StatusCode::SERVICE_UNAVAILABLE, message), Self::Internal => ( StatusCode::INTERNAL_SERVER_ERROR, "internal server error".to_owned(), diff --git a/src/main.rs b/src/main.rs index 6149602..2d76a9d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,14 @@ mod config; mod error; +mod recording_routes; mod routes; +mod speech; +mod summary; +mod transcription_worker; -use std::sync::Arc; +use std::{path::PathBuf, sync::Arc}; -use axum::{Json, Router, http::HeaderValue, routing::get}; +use axum::{Json, Router, extract::DefaultBodyLimit, http::HeaderValue, routing::get}; use config::Config; use sqlx::{PgPool, postgres::PgPoolOptions}; use tower_http::{cors::CorsLayer, trace::TraceLayer}; @@ -14,6 +18,9 @@ use utoipa_scalar::{Scalar, Servable}; #[derive(Clone)] pub struct AppState { pub pool: Option, + pub audio_storage_dir: PathBuf, + pub speech: speech::SpeechService, + pub summary: summary::SummaryService, } #[tokio::main] @@ -25,7 +32,19 @@ async fn main() -> Result<(), Box> { let config = Config::from_env().map_err(std::io::Error::other)?; let pool = connect_database(&config).await?; - let app = build_app(AppState { pool }, config.cors_allowed_origin.as_deref())?; + let speech = speech::SpeechService::new(config.speech.clone()); + let state = AppState { + pool, + audio_storage_dir: config.audio_storage_dir.clone(), + speech: speech.clone(), + summary: summary::SummaryService::new(config.summary.clone()), + }; + if let Some(pool) = state.pool.clone() + && speech.configured() + { + transcription_worker::spawn(pool, speech); + } + let app = build_app(state, config.cors_allowed_origin.as_deref())?; let address = std::net::SocketAddr::from((config.host, config.port)); let listener = tokio::net::TcpListener::bind(address).await?; @@ -49,6 +68,7 @@ pub fn build_app(state: AppState, cors_allowed_origin: Option<&str>) -> Result; + +const MAX_AUDIO_UPLOAD_BYTES: usize = 15 * 1024 * 1024; +const MAX_SEGMENT_DURATION_MS: i32 = 600_000; +const SHORT_RECORDING_DURATION_MS: i32 = 60_000; +const SHORT_TRANSCRIPT_CHARS: usize = 600; + +pub fn router() -> OpenApiRouter { + OpenApiRouter::new() + .routes(routes!(get_active_feedback_session)) + .routes(routes!(ensure_feedback_session)) + .routes(routes!(update_feedback_session)) + .routes(routes!(create_lesson_session)) + .routes(routes!(upload_audio_segment)) + .routes(routes!(finish_lesson_session)) + .routes(routes!(mark_lesson_applied)) + .routes(routes!(retry_audio_segment)) + .routes(routes!(generate_feedback_content)) +} + +#[derive(Debug, Deserialize)] +struct ActiveSessionQuery { + profile_id: Option, +} + +#[derive(Debug, Deserialize, ToSchema)] +#[schema(example = json!({ + "profile_id": "11111111-1111-4111-8111-111111111111", + "feedback_date": "2026-07-16", + "content": "" +}))] +struct FeedbackSessionInput { + profile_id: Option, + feedback_date: NaiveDate, + #[serde(default)] + content: String, +} + +#[derive(Debug, Deserialize, ToSchema)] +struct FeedbackSessionUpdateInput { + feedback_date: NaiveDate, + content: String, +} + +#[derive(Debug, FromRow)] +struct FeedbackSessionRow { + id: Uuid, + profile_id: Option, + feedback_date: NaiveDate, + content: String, + status: String, + created_at: DateTime, + updated_at: DateTime, +} + +#[derive(Debug, FromRow)] +struct LessonRow { + id: Uuid, + sequence: i32, + status: String, + duration_ms: i32, + applied_directly: bool, + included_in_generation: bool, + started_at: DateTime, + ended_at: Option>, +} + +#[derive(Debug, FromRow)] +struct SegmentRow { + id: Uuid, + lesson_session_id: Uuid, + sequence: i32, + duration_ms: i32, + status: String, + error_message: Option, +} + +#[derive(Debug, Serialize, ToSchema)] +struct FeedbackSessionResponse { + id: Uuid, + profile_id: Option, + feedback_date: NaiveDate, + content: String, + status: String, + lesson_count: usize, + total_duration_ms: i64, + processing_segment_count: usize, + failed_segment_count: usize, + needs_generation: bool, + lessons: Vec, + created_at: DateTime, + updated_at: DateTime, +} + +#[derive(Debug, Serialize, ToSchema)] +struct LessonSummaryResponse { + id: Uuid, + sequence: i32, + status: String, + duration_ms: i32, + applied_directly: bool, + included_in_generation: bool, + segments: Vec, + started_at: DateTime, + ended_at: Option>, +} + +#[derive(Debug, Serialize, ToSchema)] +struct AudioSegmentResponse { + id: Uuid, + sequence: i32, + duration_ms: i32, + status: String, + error_message: Option, +} + +#[derive(Debug, Serialize, ToSchema)] +struct LessonCreatedResponse { + id: Uuid, + sequence: i32, + status: String, + started_at: DateTime, +} + +#[derive(Debug, Serialize, ToSchema)] +struct FinishedLessonResponse { + id: Uuid, + status: String, + duration_ms: i32, + transcript: String, + can_apply_directly: bool, +} + +#[derive(Debug, Deserialize, ToSchema)] +struct GenerateFeedbackInput { + #[schema(example = "学生本周构图更加稳定。")] + existing_content: String, +} + +#[derive(Debug, Deserialize, ToSchema)] +struct MarkLessonAppliedInput { + content: String, +} + +#[derive(Debug, Serialize, ToSchema)] +struct GenerateFeedbackResponse { + content: String, + method: String, +} + +#[derive(Debug, ToSchema)] +#[allow(dead_code)] +struct AudioSegmentUpload { + sequence: i32, + duration_ms: i32, + audio_format: String, + audio: String, +} + +#[utoipa::path( + get, + path = "/api/v1/feedback-sessions/active", + tag = "语音反馈", + summary = "读取进行中的反馈批次", + description = "读取当前用户指定学生下尚未保存的反馈批次；profile_id 留空时读取不关联学生的批次。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("profile_id" = Option, Query, description = "学生档案 ID；留空表示不关联学生") + ), + responses( + (status = 200, description = "进行中的反馈批次；不存在时返回 null", body = Option), + CommonApiErrors + ) +)] +async fn get_active_feedback_session( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result>, ApiError> { + let owner_id = current_user(&headers)?; + let session = sqlx::query_as::<_, FeedbackSessionRow>( + "SELECT id, profile_id, feedback_date, content, status, created_at, updated_at \ + FROM feedback_sessions \ + WHERE owner_id = $1 AND profile_id IS NOT DISTINCT FROM $2 AND status = 'active' \ + ORDER BY updated_at DESC LIMIT 1", + ) + .bind(owner_id) + .bind(query.profile_id) + .fetch_optional(pool(&state)?) + .await?; + + match session { + Some(session) => Ok(Json(Some( + load_session_response(pool(&state)?, session).await?, + ))), + None => Ok(Json(None)), + } +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-sessions", + tag = "语音反馈", + summary = "创建或恢复反馈批次", + description = "为学生创建一个反馈批次；已有未保存批次时直接恢复，并同步最新反馈日期。", + params(("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0"))), + request_body(content = FeedbackSessionInput, content_type = "application/json"), + responses( + (status = 200, description = "创建或恢复后的反馈批次", body = FeedbackSessionResponse), + (status = 404, description = "学生档案不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn ensure_feedback_session( + State(state): State, + headers: HeaderMap, + Json(input): Json, +) -> Result, ApiError> { + let owner_id = current_user(&headers)?; + let pool = pool(&state)?; + validate_draft_content(&input.content)?; + if let Some(profile_id) = input.profile_id { + assert_profile_owner(pool, owner_id, profile_id).await?; + } + + let existing = sqlx::query_as::<_, FeedbackSessionRow>( + "UPDATE feedback_sessions SET feedback_date = $3, content = $4, updated_at = now() \ + WHERE id = ( \ + SELECT id FROM feedback_sessions \ + WHERE owner_id = $1 AND profile_id IS NOT DISTINCT FROM $2 AND status = 'active' \ + ORDER BY updated_at DESC LIMIT 1 \ + ) \ + RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", + ) + .bind(owner_id) + .bind(input.profile_id) + .bind(input.feedback_date) + .bind(&input.content) + .fetch_optional(pool) + .await?; + let session = + match existing { + Some(session) => session, + None => sqlx::query_as::<_, FeedbackSessionRow>( + "INSERT INTO feedback_sessions (id, owner_id, profile_id, feedback_date, content) \ + VALUES ($1, $2, $3, $4, $5) \ + RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", + ) + .bind(Uuid::new_v4()) + .bind(owner_id) + .bind(input.profile_id) + .bind(input.feedback_date) + .bind(&input.content) + .fetch_one(pool) + .await?, + }; + + Ok(Json(load_session_response(pool, session).await?)) +} + +#[utoipa::path( + put, + path = "/api/v1/feedback-sessions/{session_id}", + tag = "语音反馈", + summary = "自动保存反馈草稿", + description = "在用户编辑反馈正文或日期时自动保存进行中的反馈批次。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("session_id" = Uuid, Path, description = "反馈批次 ID") + ), + request_body(content = FeedbackSessionUpdateInput, content_type = "application/json"), + responses( + (status = 200, description = "更新后的反馈批次", body = FeedbackSessionResponse), + (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn update_feedback_session( + State(state): State, + headers: HeaderMap, + Path(session_id): Path, + Json(input): Json, +) -> Result, ApiError> { + let owner_id = current_user(&headers)?; + validate_draft_content(&input.content)?; + let session = sqlx::query_as::<_, FeedbackSessionRow>( + "UPDATE feedback_sessions SET feedback_date = $3, content = $4, updated_at = now() \ + WHERE id = $1 AND owner_id = $2 AND status = 'active' \ + RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", + ) + .bind(session_id) + .bind(owner_id) + .bind(input.feedback_date) + .bind(&input.content) + .fetch_optional(pool(&state)?) + .await? + .ok_or(ApiError::NotFound)?; + Ok(Json(load_session_response(pool(&state)?, session).await?)) +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-sessions/{session_id}/lessons", + tag = "语音反馈", + summary = "开始一节课堂录音", + description = "在反馈批次中自动创建下一节课堂录音。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("session_id" = Uuid, Path, description = "反馈批次 ID") + ), + responses( + (status = 201, description = "课堂录音已创建", body = LessonCreatedResponse), + (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn create_lesson_session( + State(state): State, + headers: HeaderMap, + Path(session_id): Path, +) -> Result<(axum::http::StatusCode, Json), ApiError> { + let owner_id = current_user(&headers)?; + assert_session_owner(pool(&state)?, owner_id, session_id).await?; + sqlx::query( + "DELETE FROM lesson_sessions \ + WHERE feedback_session_id = $1 AND status IN ('recording', 'processing')", + ) + .bind(session_id) + .execute(pool(&state)?) + .await?; + let id = Uuid::new_v4(); + let lesson = sqlx::query_as::<_, (Uuid, i32, String, DateTime)>( + "INSERT INTO lesson_sessions (id, feedback_session_id, sequence) \ + SELECT $1, $2, COALESCE(MAX(sequence), 0) + 1 \ + FROM lesson_sessions WHERE feedback_session_id = $2 \ + RETURNING id, sequence, status, started_at", + ) + .bind(id) + .bind(session_id) + .fetch_one(pool(&state)?) + .await?; + touch_session(pool(&state)?, session_id).await?; + Ok(( + axum::http::StatusCode::CREATED, + Json(LessonCreatedResponse { + id: lesson.0, + sequence: lesson.1, + status: lesson.2, + started_at: lesson.3, + }), + )) +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-lessons/{lesson_id}/segments", + tag = "语音反馈", + summary = "上传并转录录音片段", + description = "上传单个录音片段并加入后端转录队列。接口保存音频后立即返回，页面通过反馈批次状态无感获取转录结果。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("lesson_id" = Uuid, Path, description = "课堂录音 ID") + ), + request_body(content = AudioSegmentUpload, content_type = "multipart/form-data"), + responses( + (status = 201, description = "片段已保存并进入转录队列", body = AudioSegmentResponse), + (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn upload_audio_segment( + State(state): State, + headers: HeaderMap, + Path(lesson_id): Path, + mut multipart: Multipart, +) -> Result<(axum::http::StatusCode, Json), ApiError> { + let owner_id = current_user(&headers)?; + let session_id = assert_lesson_owner(pool(&state)?, owner_id, lesson_id).await?; + let mut sequence = None; + let mut duration_ms = None; + let mut audio_format = None; + let mut audio = None; + + while let Some(field) = multipart + .next_field() + .await + .map_err(|error| ApiError::BadRequest(format!("invalid multipart body: {error}")))? + { + let name = field.name().unwrap_or_default().to_owned(); + match name.as_str() { + "sequence" => sequence = Some(parse_multipart_i32(field, "sequence").await?), + "duration_ms" => duration_ms = Some(parse_multipart_i32(field, "duration_ms").await?), + "audio_format" => { + audio_format = Some( + field + .text() + .await + .map_err(|_| ApiError::BadRequest("audio_format is invalid".to_owned()))?, + ) + } + "audio" => { + let bytes = field + .bytes() + .await + .map_err(|_| ApiError::BadRequest("audio file is invalid".to_owned()))?; + if bytes.len() > MAX_AUDIO_UPLOAD_BYTES { + return Err(ApiError::BadRequest("audio file exceeds 15 MB".to_owned())); + } + audio = Some(bytes.to_vec()); + } + _ => {} + } + } + + let sequence = sequence + .filter(|value| *value > 0) + .ok_or_else(|| ApiError::BadRequest("sequence must be a positive integer".to_owned()))?; + let duration_ms = duration_ms + .filter(|value| (1..=MAX_SEGMENT_DURATION_MS).contains(value)) + .ok_or_else(|| { + ApiError::BadRequest("duration_ms must be between 1 and 600000".to_owned()) + })?; + let audio_format = audio_format + .map(|value| value.trim().to_ascii_lowercase()) + .filter(|value| matches!(value.as_str(), "mp3" | "aac" | "wav" | "m4a" | "amr")) + .ok_or_else(|| ApiError::BadRequest("unsupported audio_format".to_owned()))?; + let audio = audio + .filter(|bytes| !bytes.is_empty()) + .ok_or_else(|| ApiError::BadRequest("audio file cannot be empty".to_owned()))?; + + let file_id = Uuid::new_v4(); + let directory = state + .audio_storage_dir + .join(owner_id.to_string()) + .join(lesson_id.to_string()); + tokio::fs::create_dir_all(&directory) + .await + .map_err(internal_io_error)?; + let audio_path = directory.join(format!("{file_id}.{audio_format}")); + tokio::fs::write(&audio_path, &audio) + .await + .map_err(internal_io_error)?; + + let segment_id = sqlx::query_scalar::<_, Uuid>( + "INSERT INTO audio_segments \ + (id, lesson_session_id, sequence, duration_ms, audio_format, audio_path, status) \ + VALUES ($1, $2, $3, $4, $5, $6, 'processing') \ + ON CONFLICT (lesson_session_id, sequence) DO UPDATE SET \ + duration_ms = EXCLUDED.duration_ms, audio_format = EXCLUDED.audio_format, \ + audio_path = EXCLUDED.audio_path, status = 'processing', transcript = NULL, \ + error_message = NULL, provider_request_id = NULL, processing_started_at = NULL, \ + updated_at = now() \ + RETURNING id", + ) + .bind(file_id) + .bind(lesson_id) + .bind(sequence) + .bind(duration_ms) + .bind(&audio_format) + .bind(audio_path.to_string_lossy().as_ref()) + .fetch_one(pool(&state)?) + .await?; + sqlx::query( + "UPDATE lesson_sessions SET status = 'processing', updated_at = now() WHERE id = $1", + ) + .bind(lesson_id) + .execute(pool(&state)?) + .await?; + + touch_session(pool(&state)?, session_id).await?; + Ok(( + axum::http::StatusCode::CREATED, + Json(AudioSegmentResponse { + id: segment_id, + sequence, + duration_ms, + status: "processing".to_owned(), + error_message: None, + }), + )) +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-lessons/{lesson_id}/finish", + tag = "语音反馈", + summary = "结束一节课堂录音", + description = "汇总该节课的片段状态和转录文本；短录音会标记为可直接加入反馈。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("lesson_id" = Uuid, Path, description = "课堂录音 ID") + ), + responses( + (status = 200, description = "课堂录音汇总结果", body = FinishedLessonResponse), + (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn finish_lesson_session( + State(state): State, + headers: HeaderMap, + Path(lesson_id): Path, +) -> Result, ApiError> { + let owner_id = current_user(&headers)?; + let session_id = assert_lesson_owner(pool(&state)?, owner_id, lesson_id).await?; + let segments = sqlx::query_as::<_, (String, i32, Option)>( + "SELECT status, duration_ms, transcript FROM audio_segments \ + WHERE lesson_session_id = $1 ORDER BY sequence", + ) + .bind(lesson_id) + .fetch_all(pool(&state)?) + .await?; + let duration_ms = segments.iter().map(|segment| segment.1).sum::(); + let transcript = segments + .iter() + .filter_map(|segment| segment.2.as_deref()) + .collect::>() + .join("\n"); + let status = if segments.is_empty() || segments.iter().any(|segment| segment.0 == "failed") { + "failed" + } else if segments.iter().any(|segment| segment.0 == "processing") { + "processing" + } else { + "ready" + }; + sqlx::query( + "UPDATE lesson_sessions SET status = $2, duration_ms = $3, ended_at = now(), updated_at = now() \ + WHERE id = $1", + ) + .bind(lesson_id) + .bind(status) + .bind(duration_ms) + .execute(pool(&state)?) + .await?; + touch_session(pool(&state)?, session_id).await?; + + Ok(Json(FinishedLessonResponse { + id: lesson_id, + status: status.to_owned(), + duration_ms, + can_apply_directly: status == "ready" + && duration_ms <= SHORT_RECORDING_DURATION_MS + && transcript.chars().count() <= SHORT_TRANSCRIPT_CHARS, + transcript, + })) +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-lessons/{lesson_id}/mark-applied", + tag = "语音反馈", + summary = "标记短录音已加入反馈", + description = "短录音文字直接加入反馈内容后，标记该节课无需再次汇总。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("lesson_id" = Uuid, Path, description = "课堂录音 ID") + ), + request_body(content = MarkLessonAppliedInput, content_type = "application/json"), + responses( + (status = 204, description = "已标记"), + (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn mark_lesson_applied( + State(state): State, + headers: HeaderMap, + Path(lesson_id): Path, + Json(input): Json, +) -> Result { + let owner_id = current_user(&headers)?; + validate_draft_content(&input.content)?; + let pool = pool(&state)?; + let session_id = assert_lesson_owner(pool, owner_id, lesson_id).await?; + let mut transaction = pool.begin().await?; + let result = sqlx::query( + "UPDATE lesson_sessions SET applied_directly = TRUE, updated_at = now() \ + WHERE id = $1 AND status = 'ready'", + ) + .bind(lesson_id) + .execute(&mut *transaction) + .await?; + if result.rows_affected() == 0 { + return Err(ApiError::BadRequest("lesson is not ready".to_owned())); + } + sqlx::query("UPDATE feedback_sessions SET content = $2, updated_at = now() WHERE id = $1") + .bind(session_id) + .bind(&input.content) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(axum::http::StatusCode::NO_CONTENT) +} + +#[utoipa::path( + post, + path = "/api/v1/audio-segments/{segment_id}/retry", + tag = "语音反馈", + summary = "重试失败的录音转录", + description = "将后端保留的失败音频重新加入转录队列，接口立即返回。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("segment_id" = Uuid, Path, description = "录音片段 ID") + ), + responses( + (status = 200, description = "片段已重新进入转录队列", body = AudioSegmentResponse), + (status = 404, description = "录音片段不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn retry_audio_segment( + State(state): State, + headers: HeaderMap, + Path(segment_id): Path, +) -> Result, ApiError> { + let owner_id = current_user(&headers)?; + let segment = sqlx::query_as::<_, (i32, i32, String, Uuid)>( + "SELECT s.sequence, s.duration_ms, s.audio_path, l.id \ + FROM audio_segments s \ + JOIN lesson_sessions l ON l.id = s.lesson_session_id \ + JOIN feedback_sessions f ON f.id = l.feedback_session_id \ + WHERE s.id = $1 AND f.owner_id = $2 AND f.status = 'active'", + ) + .bind(segment_id) + .bind(owner_id) + .fetch_optional(pool(&state)?) + .await? + .ok_or(ApiError::NotFound)?; + tokio::fs::metadata(&segment.2) + .await + .map_err(internal_io_error)?; + sqlx::query( + "UPDATE audio_segments SET status = 'processing', error_message = NULL, \ + processing_started_at = NULL, updated_at = now() WHERE id = $1", + ) + .bind(segment_id) + .execute(pool(&state)?) + .await?; + sqlx::query( + "UPDATE lesson_sessions SET status = 'processing', updated_at = now() WHERE id = $1", + ) + .bind(segment.3) + .execute(pool(&state)?) + .await?; + let session_id = sqlx::query_scalar::<_, Uuid>( + "SELECT feedback_session_id FROM lesson_sessions WHERE id = $1", + ) + .bind(segment.3) + .fetch_one(pool(&state)?) + .await?; + touch_session(pool(&state)?, session_id).await?; + Ok(Json(AudioSegmentResponse { + id: segment_id, + sequence: segment.0, + duration_ms: segment.1, + status: "processing".to_owned(), + error_message: None, + })) +} + +#[utoipa::path( + post, + path = "/api/v1/feedback-sessions/{session_id}/generate", + tag = "语音反馈", + summary = "生成汇总反馈", + description = "将尚未处理的长录音或多节课堂转录与教师已有内容合并。未配置大模型时使用本地抽取式汇总。", + params( + ("X-User-Id" = Uuid, Header, description = "开发阶段用户 ID", example = json!("5a4da7f0-d70c-465f-bcab-124c504aa9f0")), + ("session_id" = Uuid, Path, description = "反馈批次 ID") + ), + request_body(content = GenerateFeedbackInput, content_type = "application/json"), + responses( + (status = 200, description = "汇总后的反馈正文", body = GenerateFeedbackResponse), + (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), + CommonApiErrors + ) +)] +async fn generate_feedback_content( + State(state): State, + headers: HeaderMap, + Path(session_id): Path, + Json(input): Json, +) -> Result, ApiError> { + let owner_id = current_user(&headers)?; + validate_draft_content(&input.existing_content)?; + assert_session_owner(pool(&state)?, owner_id, session_id).await?; + let pending_status = sqlx::query_as::<_, (bool, bool)>( + "SELECT \ + EXISTS(SELECT 1 FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ + WHERE l.feedback_session_id = $1 AND s.status = 'processing'), \ + EXISTS(SELECT 1 FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ + WHERE l.feedback_session_id = $1 AND s.status = 'failed')", + ) + .bind(session_id) + .fetch_one(pool(&state)?) + .await?; + if pending_status.0 { + return Err(ApiError::BadRequest( + "录音仍在后台转录中，请稍后再生成反馈".to_owned(), + )); + } + if pending_status.1 { + return Err(ApiError::BadRequest( + "部分录音转录失败，请重试后再生成反馈".to_owned(), + )); + } + let lessons = sqlx::query_as::<_, (Uuid, String)>( + "SELECT l.id, string_agg(s.transcript, E'\\n' ORDER BY s.sequence) \ + FROM lesson_sessions l \ + JOIN audio_segments s ON s.lesson_session_id = l.id \ + WHERE l.feedback_session_id = $1 AND l.status = 'ready' \ + AND l.applied_directly = FALSE AND l.included_in_generation = FALSE \ + AND s.status = 'ready' AND s.transcript IS NOT NULL \ + GROUP BY l.id, l.sequence ORDER BY l.sequence", + ) + .bind(session_id) + .fetch_all(pool(&state)?) + .await?; + if lessons.is_empty() { + return Err(ApiError::BadRequest("没有待汇总的录音内容".to_owned())); + } + let transcripts = lessons + .iter() + .map(|lesson| lesson.1.clone()) + .collect::>(); + let summary = state + .summary + .summarize(&input.existing_content, &transcripts) + .await + .map_err(ApiError::ServiceUnavailable)?; + let lesson_ids = lessons.iter().map(|lesson| lesson.0).collect::>(); + let pool = pool(&state)?; + let mut transaction = pool.begin().await?; + sqlx::query( + "UPDATE lesson_sessions SET included_in_generation = TRUE, updated_at = now() \ + WHERE id = ANY($1)", + ) + .bind(&lesson_ids) + .execute(&mut *transaction) + .await?; + sqlx::query("UPDATE feedback_sessions SET content = $2, updated_at = now() WHERE id = $1") + .bind(session_id) + .bind(&summary.content) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + + Ok(Json(GenerateFeedbackResponse { + content: summary.content, + method: summary.method.to_owned(), + })) +} + +async fn load_session_response( + pool: &PgPool, + session: FeedbackSessionRow, +) -> Result { + let lessons = sqlx::query_as::<_, LessonRow>( + "SELECT id, sequence, status, duration_ms, applied_directly, included_in_generation, \ + started_at, ended_at FROM lesson_sessions \ + WHERE feedback_session_id = $1 ORDER BY sequence", + ) + .bind(session.id) + .fetch_all(pool) + .await?; + let segments = sqlx::query_as::<_, SegmentRow>( + "SELECT s.id, s.lesson_session_id, s.sequence, s.duration_ms, s.status, s.error_message \ + FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ + WHERE l.feedback_session_id = $1 ORDER BY l.sequence, s.sequence", + ) + .bind(session.id) + .fetch_all(pool) + .await?; + let processing_segment_count = segments + .iter() + .filter(|segment| segment.status == "processing") + .count() + + lessons + .iter() + .filter(|lesson| matches!(lesson.status.as_str(), "recording" | "processing")) + .count(); + let failed_segment_count = segments + .iter() + .filter(|segment| segment.status == "failed") + .count(); + let total_duration_ms = lessons + .iter() + .map(|lesson| i64::from(lesson.duration_ms)) + .sum(); + let needs_generation = lessons.iter().any(|lesson| { + lesson.status == "ready" && !lesson.applied_directly && !lesson.included_in_generation + }); + let mut segments_by_lesson: HashMap> = HashMap::new(); + for segment in segments { + segments_by_lesson + .entry(segment.lesson_session_id) + .or_default() + .push(AudioSegmentResponse { + id: segment.id, + sequence: segment.sequence, + duration_ms: segment.duration_ms, + status: segment.status, + error_message: segment.error_message, + }); + } + let lesson_count = lessons.len(); + let lesson_responses = lessons + .into_iter() + .map(|lesson| LessonSummaryResponse { + id: lesson.id, + sequence: lesson.sequence, + status: lesson.status, + duration_ms: lesson.duration_ms, + applied_directly: lesson.applied_directly, + included_in_generation: lesson.included_in_generation, + segments: segments_by_lesson.remove(&lesson.id).unwrap_or_default(), + started_at: lesson.started_at, + ended_at: lesson.ended_at, + }) + .collect(); + + Ok(FeedbackSessionResponse { + id: session.id, + profile_id: session.profile_id, + feedback_date: session.feedback_date, + content: session.content, + status: session.status, + lesson_count, + total_duration_ms, + processing_segment_count, + failed_segment_count, + needs_generation, + lessons: lesson_responses, + created_at: session.created_at, + updated_at: session.updated_at, + }) +} + +async fn parse_multipart_i32( + field: axum::extract::multipart::Field<'_>, + name: &str, +) -> Result { + field + .text() + .await + .map_err(|_| ApiError::BadRequest(format!("{name} is invalid")))? + .parse() + .map_err(|_| ApiError::BadRequest(format!("{name} must be an integer"))) +} + +fn current_user(headers: &HeaderMap) -> Result { + let value = headers + .get("x-user-id") + .and_then(|header| header.to_str().ok()) + .ok_or(ApiError::Unauthorized)?; + Uuid::parse_str(value).map_err(|_| ApiError::BadRequest("X-User-Id must be a UUID".to_owned())) +} + +fn pool(state: &AppState) -> Result<&PgPool, ApiError> { + state.pool.as_ref().ok_or(ApiError::DatabaseUnavailable) +} + +async fn assert_profile_owner( + pool: &PgPool, + owner_id: Uuid, + profile_id: Uuid, +) -> Result<(), ApiError> { + let exists = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM student_profiles WHERE id = $1 AND owner_id = $2)", + ) + .bind(profile_id) + .bind(owner_id) + .fetch_one(pool) + .await?; + if exists { + Ok(()) + } else { + Err(ApiError::NotFound) + } +} + +async fn assert_session_owner( + pool: &PgPool, + owner_id: Uuid, + session_id: Uuid, +) -> Result<(), ApiError> { + let exists = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM feedback_sessions WHERE id = $1 AND owner_id = $2 AND status = 'active')", + ) + .bind(session_id) + .bind(owner_id) + .fetch_one(pool) + .await?; + if exists { + Ok(()) + } else { + Err(ApiError::NotFound) + } +} + +async fn assert_lesson_owner( + pool: &PgPool, + owner_id: Uuid, + lesson_id: Uuid, +) -> Result { + sqlx::query_scalar::<_, Uuid>( + "SELECT f.id FROM lesson_sessions l \ + JOIN feedback_sessions f ON f.id = l.feedback_session_id \ + WHERE l.id = $1 AND f.owner_id = $2 AND f.status = 'active'", + ) + .bind(lesson_id) + .bind(owner_id) + .fetch_optional(pool) + .await? + .ok_or(ApiError::NotFound) +} + +async fn touch_session(pool: &PgPool, session_id: Uuid) -> Result<(), ApiError> { + sqlx::query("UPDATE feedback_sessions SET updated_at = now() WHERE id = $1") + .bind(session_id) + .execute(pool) + .await?; + Ok(()) +} + +fn internal_io_error(error: std::io::Error) -> ApiError { + tracing::error!(?error, "audio file operation failed"); + ApiError::Internal +} + +fn validate_draft_content(content: &str) -> Result<(), ApiError> { + if content.chars().count() > 2000 { + return Err(ApiError::BadRequest( + "feedback content cannot exceed 2000 characters".to_owned(), + )); + } + Ok(()) +} diff --git a/src/routes.rs b/src/routes.rs index 4fec9c5..4ab7de8 100644 --- a/src/routes.rs +++ b/src/routes.rs @@ -18,6 +18,7 @@ use uuid::Uuid; use crate::{ AppState, error::{ApiError, CommonApiErrors, ErrorBody}, + recording_routes, }; type SharedState = Arc; @@ -34,6 +35,7 @@ pub fn router() -> (Router, OpenApi) { .routes(routes!(get_defaults, update_defaults)) .routes(routes!(list_feedback_records, create_feedback_record)) .routes(routes!(delete_feedback_record)) + .merge(recording_routes::router()) .split_for_parts(); let mut info = Info::new("教学反馈助手 API", env!("CARGO_PKG_VERSION")); @@ -49,6 +51,7 @@ pub fn router() -> (Router, OpenApi) { api_tag("学生档案", "创建、查询、更新和删除当前用户的学生档案。"), api_tag("档案预设", "读取或保存当前用户新建档案时使用的默认值。"), api_tag("反馈记录", "创建、查询和删除当前用户的教学反馈记录。"), + api_tag("语音反馈", "录制多节课堂、转录音频并生成反馈草稿。"), ]); (router, openapi) @@ -62,12 +65,27 @@ fn api_tag(name: &str, description: &str) -> Tag { /// 健康检查响应。 #[derive(Serialize, ToSchema)] -#[schema(example = json!({"status": "ok", "database_configured": true}))] +#[schema(example = json!({ + "status": "ok", + "database_configured": true, + "speech_configured": true, + "speech_available": true, + "speech_provider": "local-funasr", + "summary_configured": false +}))] struct HealthResponse { /// HTTP 服务状态；正常时固定为 `ok`。 status: &'static str, /// 是否已经通过 `DATABASE_URL` 配置数据库连接。 database_configured: bool, + /// 是否已经配置长音频转录服务。 + speech_configured: bool, + /// 当前转录服务是否可连接；本地模型启动和下载完成后为 `true`。 + speech_available: bool, + /// 当前语音提供方；未启用转录时为 `null`。 + speech_provider: Option<&'static str>, + /// 是否已经配置生成式反馈汇总服务；未配置时使用本地抽取式汇总。 + summary_configured: bool, } /// 检查服务状态。 @@ -85,6 +103,10 @@ async fn health(State(state): State) -> Json { Json(HealthResponse { status: "ok", database_configured: state.pool.is_some(), + speech_configured: state.speech.configured(), + speech_available: state.speech.available().await, + speech_provider: state.speech.provider_name(), + summary_configured: state.summary.configured(), }) } @@ -561,7 +583,8 @@ struct FeedbackRecord { #[schema(example = json!({ "profile_id": null, "feedback_date": "2026-07-15", - "content": "今天完成了色彩搭配练习，构图有明显进步。" + "content": "今天完成了色彩搭配练习，构图有明显进步。", + "feedback_session_id": null }))] struct FeedbackRecordInput { /// 可选的学生档案 ID；如填写，必须属于当前用户。Scalar 默认留空以便直接测试。 @@ -570,6 +593,8 @@ struct FeedbackRecordInput { feedback_date: NaiveDate, /// 教学反馈正文，不可为空。 content: String, + /// 可选的语音反馈批次 ID；保存后对应批次会被标记为已完成。 + feedback_session_id: Option, } #[derive(Debug, Deserialize)] @@ -634,7 +659,8 @@ async fn list_feedback_records( example = json!({ "profile_id": null, "feedback_date": "2026-07-15", - "content": "今天完成了色彩搭配练习，构图有明显进步。" + "content": "今天完成了色彩搭配练习，构图有明显进步。", + "feedback_session_id": null }) ), responses( @@ -648,13 +674,33 @@ async fn create_feedback_record( headers: HeaderMap, Json(input): Json, ) -> Result<(axum::http::StatusCode, Json), ApiError> { - if input.content.trim().is_empty() { - return Err(ApiError::BadRequest("content cannot be empty".to_owned())); + if input.content.trim().is_empty() || input.content.chars().count() > 2000 { + return Err(ApiError::BadRequest( + "content must contain 1 to 2000 characters".to_owned(), + )); } let owner_id = current_user(&headers)?; + let pool = pool(&state)?; if let Some(profile_id) = input.profile_id { - assert_profile_owner(pool(&state)?, owner_id, profile_id).await?; + assert_profile_owner(pool, owner_id, profile_id).await?; } + if let Some(session_id) = input.feedback_session_id { + let session_profile_id = sqlx::query_scalar::<_, Option>( + "SELECT profile_id FROM feedback_sessions \ + WHERE id = $1 AND owner_id = $2 AND status = 'active'", + ) + .bind(session_id) + .bind(owner_id) + .fetch_optional(pool) + .await? + .ok_or(ApiError::NotFound)?; + if session_profile_id != input.profile_id { + return Err(ApiError::BadRequest( + "feedback session profile does not match feedback record".to_owned(), + )); + } + } + let mut transaction = pool.begin().await?; let record = sqlx::query_as::<_, FeedbackRecord>( "INSERT INTO feedback_records (id, owner_id, profile_id, feedback_date, content) \ VALUES ($1, $2, $3, $4, $5) \ @@ -665,8 +711,19 @@ async fn create_feedback_record( .bind(input.profile_id) .bind(input.feedback_date) .bind(input.content.trim()) - .fetch_one(pool(&state)?) + .fetch_one(&mut *transaction) .await?; + if let Some(session_id) = input.feedback_session_id { + sqlx::query( + "UPDATE feedback_sessions SET status = 'finalized', finalized_record_id = $2, updated_at = now() \ + WHERE id = $1", + ) + .bind(session_id) + .bind(record.id) + .execute(&mut *transaction) + .await?; + } + transaction.commit().await?; Ok((axum::http::StatusCode::CREATED, Json(record))) } @@ -827,7 +884,7 @@ mod tests { .filter(|operation| operation.get("responses").is_some()) .collect::>(); - assert_eq!(operations.len(), 11); + assert_eq!(operations.len(), 20); for operation in &operations { assert_non_empty(operation, "summary"); assert_non_empty(operation, "description"); diff --git a/src/speech.rs b/src/speech.rs new file mode 100644 index 0000000..aa3843d --- /dev/null +++ b/src/speech.rs @@ -0,0 +1,350 @@ +use std::{collections::BTreeMap, time::Duration}; + +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use hmac::{Hmac, Mac}; +use reqwest::{ + Client, + header::{AUTHORIZATION, CONTENT_TYPE}, + multipart::{Form, Part}, +}; +use serde::Deserialize; +use sha1::Sha1; + +use crate::config::{SpeechConfig, SpeechProviderConfig, TencentSpeechConfig}; + +const ASR_HOST: &str = "asr.cloud.tencent.com"; + +#[derive(Clone)] +pub struct SpeechService { + config: Option, + client: Client, +} + +#[derive(Debug)] +pub struct Transcription { + pub text: String, + pub duration_ms: i32, + pub request_id: String, +} + +#[derive(Deserialize)] +struct LocalResponse { + text: String, + #[serde(default)] + duration_ms: i32, + #[serde(default)] + request_id: String, +} + +#[derive(Deserialize)] +struct FlashResponse { + code: i32, + message: String, + request_id: Option, + audio_duration: Option, + #[serde(default)] + flash_result: Vec, +} + +#[derive(Deserialize)] +struct FlashResult { + text: String, +} + +impl SpeechService { + pub fn new(config: Option) -> Self { + let timeout = config.as_ref().map_or(Duration::from_secs(30), |value| { + Duration::from_secs(value.request_timeout_seconds) + }); + let client = Client::builder() + .timeout(timeout) + .build() + .expect("reqwest client configuration should be valid"); + Self { config, client } + } + + pub fn configured(&self) -> bool { + self.config.is_some() + } + + pub fn provider_name(&self) -> Option<&'static str> { + self.config.as_ref().map(|config| match config.provider { + SpeechProviderConfig::Local { .. } => "local-funasr", + SpeechProviderConfig::Tencent(_) => "tencent", + }) + } + + pub fn worker_concurrency(&self) -> usize { + self.config + .as_ref() + .map_or(0, |config| config.worker_concurrency) + } + + pub fn job_lease_seconds(&self) -> i64 { + self.config + .as_ref() + .map_or(3_600, |config| config.job_lease_seconds) + } + + pub async fn available(&self) -> bool { + let Some(config) = &self.config else { + return false; + }; + match &config.provider { + SpeechProviderConfig::Local { api_url } => { + let request = self.client.get(format!("{api_url}/health")).send(); + matches!( + tokio::time::timeout(Duration::from_secs(2), request).await, + Ok(Ok(response)) if response.status().is_success() + ) + } + SpeechProviderConfig::Tencent(_) => true, + } + } + + pub async fn transcribe( + &self, + audio: Vec, + audio_format: &str, + duration_ms: i32, + ) -> Result { + let config = self + .config + .as_ref() + .ok_or_else(|| "语音转录服务尚未配置".to_owned())?; + match &config.provider { + SpeechProviderConfig::Local { api_url } => { + self.transcribe_local(api_url, audio, audio_format, duration_ms) + .await + } + SpeechProviderConfig::Tencent(config) => { + self.transcribe_tencent(config, audio, audio_format).await + } + } + } + + async fn transcribe_local( + &self, + api_url: &str, + audio: Vec, + audio_format: &str, + duration_ms: i32, + ) -> Result { + let part = Part::bytes(audio) + .file_name(format!("segment.{audio_format}")) + .mime_str(audio_content_type(audio_format)) + .map_err(|error| format!("录音格式无效：{error}"))?; + let form = Form::new() + .part("file", part) + .text("language", "zh") + .text("duration_ms", duration_ms.to_string()); + let response = self + .client + .post(local_transcription_url(api_url)) + .multipart(form) + .send() + .await + .map_err(|error| format!("本地语音转录服务连接失败：{error}"))?; + let status = response.status(); + let body = response + .text() + .await + .map_err(|error| format!("读取本地语音转录响应失败：{error}"))?; + if !status.is_success() { + return Err(format!( + "本地语音转录服务返回 HTTP {status}：{}", + compact_error(&body) + )); + } + let result: LocalResponse = serde_json::from_str(&body) + .map_err(|error| format!("本地语音转录响应格式无效：{error}"))?; + transcription(result.text, result.duration_ms, result.request_id) + } + + async fn transcribe_tencent( + &self, + config: &TencentSpeechConfig, + audio: Vec, + audio_format: &str, + ) -> Result { + let timestamp = chrono::Utc::now().timestamp(); + let (url, signature) = signed_request(config, audio_format, timestamp)?; + let response = self + .client + .post(url) + .header(AUTHORIZATION, signature) + .header(CONTENT_TYPE, "application/octet-stream") + .body(audio) + .send() + .await + .map_err(|error| format!("语音转录服务连接失败：{error}"))?; + let status = response.status(); + let body = response + .text() + .await + .map_err(|error| format!("读取语音转录响应失败：{error}"))?; + if !status.is_success() { + return Err(format!("语音转录服务返回 HTTP {status}")); + } + + let result: FlashResponse = serde_json::from_str(&body) + .map_err(|error| format!("语音转录响应格式无效：{error}"))?; + if result.code != 0 { + return Err(format!( + "语音转录失败（{}）：{}", + result.code, result.message + )); + } + let text = result + .flash_result + .into_iter() + .map(|channel| channel.text) + .collect::>() + .join("\n"); + transcription( + text, + result.audio_duration.unwrap_or_default(), + result.request_id.unwrap_or_default(), + ) + } +} + +fn transcription( + text: String, + duration_ms: i32, + request_id: String, +) -> Result { + let text = text.trim().to_owned(); + if text.is_empty() { + return Err("未识别到清晰语音".to_owned()); + } + Ok(Transcription { + text, + duration_ms, + request_id, + }) +} + +fn local_transcription_url(api_url: &str) -> String { + format!("{}/v1/transcriptions", api_url.trim_end_matches('/')) +} + +fn audio_content_type(audio_format: &str) -> &'static str { + match audio_format { + "aac" => "audio/aac", + "wav" => "audio/wav", + "m4a" => "audio/mp4", + "amr" => "audio/amr", + _ => "audio/mpeg", + } +} + +fn compact_error(body: &str) -> String { + let value = body.split_whitespace().collect::>().join(" "); + value.chars().take(240).collect() +} + +fn signed_request( + config: &TencentSpeechConfig, + audio_format: &str, + timestamp: i64, +) -> Result<(String, String), String> { + let mut params = BTreeMap::new(); + params.insert("convert_num_mode", "1".to_owned()); + params.insert("engine_type", config.engine_type.clone()); + params.insert("filter_dirty", "0".to_owned()); + params.insert("filter_modal", "1".to_owned()); + params.insert("filter_punc", "0".to_owned()); + params.insert("first_channel_only", "1".to_owned()); + params.insert("secretid", config.secret_id.clone()); + params.insert("speaker_diarization", "0".to_owned()); + params.insert("timestamp", timestamp.to_string()); + params.insert("voice_format", audio_format.to_owned()); + params.insert("word_info", "0".to_owned()); + + let query = url::form_urlencoded::Serializer::new(String::new()) + .extend_pairs(params.iter()) + .finish(); + let path = format!("/asr/flash/v1/{}", config.app_id); + let source = format!("POST{ASR_HOST}{path}?{query}"); + let mut mac = Hmac::::new_from_slice(config.secret_key.as_bytes()) + .map_err(|_| "语音转录密钥格式无效".to_owned())?; + mac.update(source.as_bytes()); + let signature = BASE64.encode(mac.finalize().into_bytes()); + + Ok((format!("https://{ASR_HOST}{path}?{query}"), signature)) +} + +#[cfg(test)] +mod tests { + use axum::{ + Json, Router, + routing::{get, post}, + }; + use serde_json::json; + + use crate::config::{SpeechConfig, SpeechProviderConfig, TencentSpeechConfig}; + + use super::{SpeechService, local_transcription_url, signed_request}; + + #[test] + fn signed_request_sorts_parameters_and_is_stable() { + let config = TencentSpeechConfig { + app_id: 123, + secret_id: "secret-id".to_owned(), + secret_key: "secret-key".to_owned(), + engine_type: "16k_zh".to_owned(), + }; + let first = signed_request(&config, "mp3", 1_700_000_000).unwrap(); + let second = signed_request(&config, "mp3", 1_700_000_000).unwrap(); + + assert_eq!(first, second); + assert!(first.0.contains("convert_num_mode=1&engine_type=16k_zh")); + assert!(first.0.ends_with("voice_format=mp3&word_info=0")); + assert!(!first.1.is_empty()); + } + + #[test] + fn local_url_has_one_transcription_suffix() { + assert_eq!( + local_transcription_url("http://127.0.0.1:10095/"), + "http://127.0.0.1:10095/v1/transcriptions" + ); + } + + #[tokio::test] + async fn local_provider_sends_audio_and_parses_response() { + let app = Router::new() + .route("/health", get(|| async { Json(json!({"status": "ok"})) })) + .route( + "/v1/transcriptions", + post(|| async { + Json(json!({ + "text": "课堂练习完成。", + "duration_ms": 12_345, + "request_id": "local-request" + })) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let service = SpeechService::new(Some(SpeechConfig { + provider: SpeechProviderConfig::Local { + api_url: format!("http://{address}"), + }, + request_timeout_seconds: 5, + worker_concurrency: 1, + job_lease_seconds: 60, + })); + + assert!(service.available().await); + let result = service + .transcribe(vec![1, 2, 3], "mp3", 10_000) + .await + .unwrap(); + assert_eq!(result.text, "课堂练习完成。"); + assert_eq!(result.duration_ms, 12_345); + assert_eq!(result.request_id, "local-request"); + } +} diff --git a/src/summary.rs b/src/summary.rs new file mode 100644 index 0000000..df5dbd5 --- /dev/null +++ b/src/summary.rs @@ -0,0 +1,193 @@ +use std::collections::HashSet; + +use reqwest::header::{AUTHORIZATION, CONTENT_TYPE}; +use serde::{Deserialize, Serialize}; + +use crate::config::SummaryConfig; + +const MAX_FEEDBACK_CHARS: usize = 2000; + +#[derive(Clone)] +pub struct SummaryService { + config: Option, + client: reqwest::Client, +} + +#[derive(Debug)] +pub struct SummaryResult { + pub content: String, + pub method: &'static str, +} + +#[derive(Serialize)] +struct ChatRequest<'a> { + model: &'a str, + messages: Vec>, + temperature: f32, +} + +#[derive(Serialize)] +struct ChatMessage<'a> { + role: &'a str, + content: &'a str, +} + +#[derive(Deserialize)] +struct ChatResponse { + choices: Vec, +} + +#[derive(Deserialize)] +struct ChatChoice { + message: ChatChoiceMessage, +} + +#[derive(Deserialize)] +struct ChatChoiceMessage { + content: String, +} + +impl SummaryService { + pub fn new(config: Option) -> Self { + Self { + config, + client: reqwest::Client::new(), + } + } + + pub fn configured(&self) -> bool { + self.config.is_some() + } + + pub async fn summarize( + &self, + existing_content: &str, + lesson_transcripts: &[String], + ) -> Result { + let Some(config) = &self.config else { + return Ok(SummaryResult { + content: extractive_summary(existing_content, lesson_transcripts), + method: "extractive", + }); + }; + + let transcript_text = lesson_transcripts + .iter() + .enumerate() + .map(|(index, transcript)| format!("第{}节课：\n{}", index + 1, transcript)) + .collect::>() + .join("\n\n"); + let user_prompt = format!( + "教师已填写内容：\n{}\n\n新增课堂转录：\n{}", + existing_content.trim(), + transcript_text + ); + let system_prompt = "你是教学反馈整理助手。请把教师已有内容与课堂转录合并成一份面向家长的中文反馈，保留具体事实，按课堂表现、进步、问题和后续建议组织。不要虚构，不要提及录音或转录，不超过2000个汉字，只输出反馈正文。"; + let request = ChatRequest { + model: &config.model, + messages: vec![ + ChatMessage { + role: "system", + content: system_prompt, + }, + ChatMessage { + role: "user", + content: &user_prompt, + }, + ], + temperature: 0.2, + }; + let mut builder = self + .client + .post(&config.api_url) + .header(CONTENT_TYPE, "application/json") + .json(&request); + if let Some(api_key) = &config.api_key { + builder = builder.header(AUTHORIZATION, format!("Bearer {api_key}")); + } + let response = builder + .send() + .await + .map_err(|error| format!("反馈汇总服务连接失败：{error}"))?; + let status = response.status(); + if !status.is_success() { + return Err(format!("反馈汇总服务返回 HTTP {status}")); + } + let result: ChatResponse = response + .json() + .await + .map_err(|error| format!("反馈汇总响应格式无效：{error}"))?; + let content = result + .choices + .into_iter() + .next() + .map(|choice| choice.message.content.trim().to_owned()) + .filter(|content| !content.is_empty()) + .ok_or_else(|| "反馈汇总服务未返回内容".to_owned())?; + + Ok(SummaryResult { + content: content.chars().take(MAX_FEEDBACK_CHARS).collect(), + method: "llm", + }) + } +} + +fn extractive_summary(existing_content: &str, lesson_transcripts: &[String]) -> String { + let existing = existing_content.trim(); + let mut result = existing + .chars() + .take(MAX_FEEDBACK_CHARS) + .collect::(); + let mut seen = HashSet::new(); + let mut selected = Vec::new(); + + for transcript in lesson_transcripts { + for sentence in transcript.split_inclusive(['。', '！', '？', '\n']) { + let sentence = sentence.trim(); + if sentence.chars().count() < 4 || !seen.insert(sentence.to_owned()) { + continue; + } + selected.push(sentence.to_owned()); + } + } + + if !selected.is_empty() && result.chars().count() < MAX_FEEDBACK_CHARS { + if !result.is_empty() { + result.push_str("\n\n"); + } + result.push_str("课堂记录汇总：\n"); + for sentence in selected { + let prefix = "- "; + let remaining = MAX_FEEDBACK_CHARS.saturating_sub(result.chars().count()); + if remaining <= prefix.chars().count() + 4 { + break; + } + result.push_str(prefix); + result.extend( + sentence + .chars() + .take(remaining - prefix.chars().count() - 1), + ); + result.push('\n'); + } + } + + result.trim().to_owned() +} + +#[cfg(test)] +mod tests { + use super::extractive_summary; + + #[test] + fn extractive_summary_preserves_manual_content_and_deduplicates() { + let result = extractive_summary( + "手动备注。", + &["完成了构图练习。完成了构图练习。色彩更大胆。".to_owned()], + ); + + assert!(result.starts_with("手动备注。")); + assert_eq!(result.matches("完成了构图练习。").count(), 1); + assert!(result.contains("色彩更大胆。")); + } +} diff --git a/src/transcription_worker.rs b/src/transcription_worker.rs new file mode 100644 index 0000000..36b325f --- /dev/null +++ b/src/transcription_worker.rs @@ -0,0 +1,202 @@ +use std::time::Duration; + +use sqlx::{FromRow, PgPool}; +use uuid::Uuid; + +use crate::speech::{SpeechService, Transcription}; + +const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(750); +const UNAVAILABLE_POLL_INTERVAL: Duration = Duration::from_secs(5); + +#[derive(Debug, FromRow)] +struct TranscriptionJob { + id: Uuid, + lesson_session_id: Uuid, + duration_ms: i32, + audio_format: String, + audio_path: String, +} + +pub fn spawn(pool: PgPool, speech: SpeechService) { + for worker_id in 1..=speech.worker_concurrency() { + let pool = pool.clone(); + let speech = speech.clone(); + tokio::spawn(async move { + run(worker_id, pool, speech).await; + }); + } +} + +async fn run(worker_id: usize, pool: PgPool, speech: SpeechService) { + let mut service_available = true; + loop { + match has_claimable_job(&pool, speech.job_lease_seconds()).await { + Ok(false) => { + tokio::time::sleep(IDLE_POLL_INTERVAL).await; + continue; + } + Ok(true) => {} + Err(error) => { + tracing::error!(worker_id, error = %error, "failed to inspect transcription queue"); + tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; + continue; + } + } + let available = speech.available().await; + if !available { + if service_available { + tracing::warn!(worker_id, provider = ?speech.provider_name(), "ASR service is unavailable; queued audio will wait"); + } + service_available = false; + tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; + continue; + } + if !service_available { + tracing::info!(worker_id, provider = ?speech.provider_name(), "ASR service is available again"); + } + service_available = true; + + match claim_next_job(&pool, speech.job_lease_seconds()).await { + Ok(Some(job)) => { + tracing::info!(worker_id, segment_id = %job.id, "transcribing audio segment"); + if let Err(error) = process_job(&pool, &speech, &job).await { + tracing::error!(worker_id, segment_id = %job.id, error = %error, "transcription job update failed"); + tokio::time::sleep(IDLE_POLL_INTERVAL).await; + } + } + Ok(None) => tokio::time::sleep(IDLE_POLL_INTERVAL).await, + Err(error) => { + tracing::error!(worker_id, error = %error, "failed to claim transcription job"); + tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; + } + } + } +} + +async fn has_claimable_job(pool: &PgPool, lease_seconds: i64) -> Result { + sqlx::query_scalar( + "SELECT EXISTS(\ + SELECT 1 FROM audio_segments \ + WHERE status = 'processing' \ + AND (processing_started_at IS NULL \ + OR processing_started_at < now() - ($1 * interval '1 second'))\ + )", + ) + .bind(lease_seconds) + .fetch_one(pool) + .await +} + +async fn claim_next_job( + pool: &PgPool, + lease_seconds: i64, +) -> Result, sqlx::Error> { + sqlx::query_as::<_, TranscriptionJob>( + "WITH next_job AS (\ + SELECT id FROM audio_segments \ + WHERE status = 'processing' \ + AND (processing_started_at IS NULL \ + OR processing_started_at < now() - ($1 * interval '1 second')) \ + ORDER BY created_at, id \ + FOR UPDATE SKIP LOCKED LIMIT 1\ + ) \ + UPDATE audio_segments AS segment \ + SET processing_started_at = now(), \ + transcription_attempts = transcription_attempts + 1, \ + updated_at = now() \ + FROM next_job \ + WHERE segment.id = next_job.id \ + RETURNING segment.id, segment.lesson_session_id, segment.duration_ms, \ + segment.audio_format, segment.audio_path", + ) + .bind(lease_seconds) + .fetch_optional(pool) + .await +} + +async fn process_job( + pool: &PgPool, + speech: &SpeechService, + job: &TranscriptionJob, +) -> Result<(), sqlx::Error> { + let result = match tokio::fs::read(&job.audio_path).await { + Ok(audio) => { + speech + .transcribe(audio, &job.audio_format, job.duration_ms) + .await + } + Err(error) => Err(format!("读取录音文件失败：{error}")), + }; + + match result { + Ok(transcription) => complete_job(pool, job, transcription).await?, + Err(message) => fail_job(pool, job, &message).await?, + } + refresh_lesson(pool, job.lesson_session_id).await?; + Ok(()) +} + +async fn complete_job( + pool: &PgPool, + job: &TranscriptionJob, + transcription: Transcription, +) -> Result<(), sqlx::Error> { + let duration_ms = if transcription.duration_ms > 0 { + transcription.duration_ms + } else { + job.duration_ms + }; + sqlx::query( + "UPDATE audio_segments \ + SET status = 'ready', duration_ms = $2, transcript = $3, error_message = NULL, \ + provider_request_id = $4, processing_started_at = NULL, updated_at = now() \ + WHERE id = $1", + ) + .bind(job.id) + .bind(duration_ms) + .bind(&transcription.text) + .bind(&transcription.request_id) + .execute(pool) + .await?; + tracing::info!(segment_id = %job.id, "audio segment transcription is ready"); + Ok(()) +} + +async fn fail_job(pool: &PgPool, job: &TranscriptionJob, message: &str) -> Result<(), sqlx::Error> { + tracing::warn!(segment_id = %job.id, error = %message, "audio transcription failed"); + sqlx::query( + "UPDATE audio_segments \ + SET status = 'failed', error_message = $2, processing_started_at = NULL, updated_at = now() \ + WHERE id = $1", + ) + .bind(job.id) + .bind(message) + .execute(pool) + .await?; + Ok(()) +} + +async fn refresh_lesson(pool: &PgPool, lesson_id: Uuid) -> Result<(), sqlx::Error> { + sqlx::query( + "UPDATE lesson_sessions AS lesson \ + SET status = CASE \ + WHEN EXISTS (SELECT 1 FROM audio_segments WHERE lesson_session_id = $1 AND status = 'failed') THEN 'failed' \ + WHEN EXISTS (SELECT 1 FROM audio_segments WHERE lesson_session_id = $1 AND status = 'processing') THEN 'processing' \ + ELSE 'ready' \ + END, \ + duration_ms = COALESCE((SELECT SUM(duration_ms) FROM audio_segments WHERE lesson_session_id = $1), 0), \ + updated_at = now() \ + WHERE lesson.id = $1 AND lesson.ended_at IS NOT NULL", + ) + .bind(lesson_id) + .execute(pool) + .await?; + sqlx::query( + "UPDATE feedback_sessions SET updated_at = now() \ + WHERE id = (SELECT feedback_session_id FROM lesson_sessions WHERE id = $1)", + ) + .bind(lesson_id) + .execute(pool) + .await?; + Ok(()) +}