notes: 新建 notes.famzheng.me — 录音 → ASR → LLM 会议纪要

- 后端 axum + sqlite (recordings 表)：上传 multipart 流式落 PVC；spawn worker pending → transcribing (调 mochi 那边 ASR endpoint, fireredasr2 token, Whisper-style multipart) → summarizing (调 gemma-4-31b-it OpenAI 兼容接口) → done
- 鉴权 middleware：Authorization: token <PASSPHRASE>；audio 流播放 ?token= query 兜底；passphrase 走 k8s Secret 不写死
- 前端 Vue3：首次访问弹 passphrase modal；sidebar 录音列表（带状态 chip）+ content 选中显示音频 + 转写 + markdown 纪要；5s polling 进度
- k8s manifest: ns cube-notes / PVC 30Gi / Ingress notes.famzheng.me / bodylimit 600M；Secret notes-creds = {passphrase, asr_token, llm_token}
- portal apps.ts 加 notes entry

apps/notes/src/main.rs

@@ -0,0 +1,619 @@
+//! notes.famzheng.me — 录音 → ASR → LLM 会议纪要。
+//!
+//! 鉴权：所有 /api/* 必须带 `Authorization: token <PASSPHRASE>` header
+//!       （audio 流式播放支持 ?token=<PASSPHRASE> query 兜底，因为 <audio>
+//!        标签没法塞自定义 header）。
+//! 配置：全部通过环境变量注入（PASSPHRASE / ASR_* / LLM_*）；k8s Secret 挂进来。
+
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+
+use axum::{
+    body::Body,
+    extract::{DefaultBodyLimit, Multipart, Path, Request, State},
+    http::{header, StatusCode},
+    middleware::{from_fn_with_state, Next},
+    response::{IntoResponse, Json as JsonResp, Response},
+    routing::{get, post},
+    Router,
+};
+use rusqlite::{params, Connection, OptionalExtension};
+use serde::Serialize;
+use serde_json::{json, Value};
+use tokio::io::AsyncWriteExt;
+use tower::ServiceExt;
+
+const SINGLE_FILE_BYTES: usize = 500 * 1024 * 1024; // 500 MiB / 单录音
+const REQUEST_BYTES: usize = 600 * 1024 * 1024;
+
+#[derive(Clone)]
+struct AppState {
+    db: Arc<Mutex<Connection>>,
+    blobs_dir: PathBuf,
+    passphrase: String,
+    asr_url: String,
+    asr_token: String,
+    llm_gateway: String,
+    llm_token: String,
+    llm_model: String,
+    http: reqwest::Client,
+}
+
+#[tokio::main]
+async fn main() -> std::io::Result<()> {
+    cube_core::init_tracing();
+
+    let db_path = std::env::var("DB_PATH").unwrap_or_else(|_| "/data/app.db".into());
+    let blobs_dir =
+        PathBuf::from(std::env::var("BLOBS_DIR").unwrap_or_else(|_| "/data/blobs".into()));
+    let dist = std::env::var("NOTES_DIST_DIR").unwrap_or_else(|_| "/dist".into());
+
+    let passphrase = std::env::var("PASSPHRASE").unwrap_or_default();
+    if passphrase.is_empty() {
+        tracing::warn!("PASSPHRASE not set — all /api/* will return 401");
+    }
+    let asr_url = std::env::var("ASR_URL")
+        .unwrap_or_else(|_| "http://18.159.112.195:8848/v1/audio/transcriptions".into());
+    let asr_token = std::env::var("ASR_TOKEN").unwrap_or_default();
+    let llm_gateway =
+        std::env::var("LLM_GATEWAY").unwrap_or_else(|_| "http://3.135.65.204:8848/v1".into());
+    let llm_token = std::env::var("LLM_TOKEN").unwrap_or_default();
+    let llm_model = std::env::var("LLM_MODEL").unwrap_or_else(|_| "gemma-4-31b-it".into());
+
+    std::fs::create_dir_all(&blobs_dir).expect("mkdir blobs_dir");
+
+    let conn = Connection::open(&db_path).expect("open sqlite");
+    conn.execute_batch(
+        "PRAGMA journal_mode=WAL;
+         CREATE TABLE IF NOT EXISTS recordings (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            title TEXT NOT NULL,
+            filename TEXT NOT NULL,
+            mime TEXT NOT NULL,
+            size_bytes INTEGER NOT NULL DEFAULT 0,
+            status TEXT NOT NULL DEFAULT 'pending',
+            transcript TEXT,
+            summary TEXT,
+            error TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+         );",
+    )
+    .expect("init schema");
+    tracing::info!(%db_path, blobs = %blobs_dir.display(), "notes ready");
+
+    let http = reqwest::Client::builder()
+        .build()
+        .expect("build reqwest client");
+
+    let state = AppState {
+        db: Arc::new(Mutex::new(conn)),
+        blobs_dir,
+        passphrase,
+        asr_url,
+        asr_token,
+        llm_gateway,
+        llm_token,
+        llm_model,
+        http,
+    };
+
+    // 鉴权 middleware 包到 /api 上
+    let protected_api = Router::new()
+        .route("/recordings", get(list_recordings).post(upload_recording).layer(
+            DefaultBodyLimit::max(REQUEST_BYTES),
+        ))
+        .route("/recordings/:id", get(get_recording).delete(delete_recording))
+        .route("/recordings/:id/audio", get(stream_audio))
+        .route("/recordings/:id/retry", post(retry_recording))
+        .with_state(state.clone())
+        .layer(from_fn_with_state(state.clone(), auth_middleware));
+
+    let api = Router::new()
+        .route("/health", get(|| async { "ok" }))
+        .merge(protected_api);
+
+    let app = cube_core::base(dist).nest("/api", api);
+    cube_core::serve(app, 8080).await
+}
+
+// ---------- 鉴权 middleware ----------
+
+async fn auth_middleware(
+    State(s): State<AppState>,
+    req: Request<Body>,
+    next: Next,
+) -> Response {
+    if s.passphrase.is_empty() {
+        return (StatusCode::UNAUTHORIZED, "server not configured").into_response();
+    }
+    // 优先看 Authorization header
+    let header_ok = req
+        .headers()
+        .get(header::AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .map(|v| {
+            v.strip_prefix("token ")
+                .or_else(|| v.strip_prefix("Token "))
+                .or_else(|| v.strip_prefix("Bearer "))
+                .map(|t| t.trim() == s.passphrase)
+                .unwrap_or(false)
+        })
+        .unwrap_or(false);
+    // 再看 ?token= query（给 <audio src> 兜底）
+    let query_ok = req.uri().query().and_then(|q| {
+        for kv in q.split('&') {
+            if let Some(v) = kv.strip_prefix("token=") {
+                let decoded = percent_decode(v);
+                if decoded == s.passphrase {
+                    return Some(true);
+                }
+            }
+        }
+        None
+    }).unwrap_or(false);
+    if header_ok || query_ok {
+        next.run(req).await
+    } else {
+        (StatusCode::UNAUTHORIZED, "unauthorized").into_response()
+    }
+}
+
+fn percent_decode(s: &str) -> String {
+    let bytes = s.as_bytes();
+    let mut out = Vec::with_capacity(bytes.len());
+    let mut i = 0;
+    while i < bytes.len() {
+        if bytes[i] == b'%' && i + 2 < bytes.len() {
+            if let (Some(h), Some(l)) = (hex(bytes[i + 1]), hex(bytes[i + 2])) {
+                out.push((h << 4) | l);
+                i += 3;
+                continue;
+            }
+        }
+        out.push(if bytes[i] == b'+' { b' ' } else { bytes[i] });
+        i += 1;
+    }
+    String::from_utf8(out).unwrap_or_default()
+}
+fn hex(b: u8) -> Option<u8> {
+    match b {
+        b'0'..=b'9' => Some(b - b'0'),
+        b'a'..=b'f' => Some(b - b'a' + 10),
+        b'A'..=b'F' => Some(b - b'A' + 10),
+        _ => None,
+    }
+}
+
+// ---------- types ----------
+
+#[derive(Serialize)]
+struct RecordingSummary {
+    id: i64,
+    title: String,
+    filename: String,
+    mime: String,
+    size_bytes: i64,
+    status: String,
+    created_at: String,
+    has_transcript: bool,
+    has_summary: bool,
+}
+
+#[derive(Serialize)]
+struct RecordingDetail {
+    id: i64,
+    title: String,
+    filename: String,
+    mime: String,
+    size_bytes: i64,
+    status: String,
+    transcript: Option<String>,
+    summary: Option<String>,
+    error: Option<String>,
+    created_at: String,
+}
+
+// ---------- handlers ----------
+
+async fn list_recordings(
+    State(s): State<AppState>,
+) -> Result<JsonResp<Value>, AppError> {
+    let conn = s.db.lock().unwrap();
+    let mut stmt = conn.prepare(
+        "SELECT id, title, filename, mime, size_bytes, status, created_at,
+                CASE WHEN transcript IS NOT NULL AND length(transcript) > 0 THEN 1 ELSE 0 END,
+                CASE WHEN summary IS NOT NULL AND length(summary) > 0 THEN 1 ELSE 0 END
+         FROM recordings ORDER BY created_at DESC, id DESC",
+    )?;
+    let rows = stmt
+        .query_map([], |r| {
+            let ht: i64 = r.get(7)?;
+            let hs: i64 = r.get(8)?;
+            Ok(RecordingSummary {
+                id: r.get(0)?,
+                title: r.get(1)?,
+                filename: r.get(2)?,
+                mime: r.get(3)?,
+                size_bytes: r.get(4)?,
+                status: r.get(5)?,
+                created_at: r.get(6)?,
+                has_transcript: ht != 0,
+                has_summary: hs != 0,
+            })
+        })?
+        .collect::<Result<Vec<_>, _>>()?;
+    Ok(JsonResp(json!(rows)))
+}
+
+async fn get_recording(
+    State(s): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<JsonResp<RecordingDetail>, AppError> {
+    let conn = s.db.lock().unwrap();
+    type Row = (
+        String, String, String, i64, String,
+        Option<String>, Option<String>, Option<String>, String,
+    );
+    let row: Option<Row> = conn
+        .query_row(
+            "SELECT title, filename, mime, size_bytes, status,
+                    transcript, summary, error, created_at
+             FROM recordings WHERE id = ?1",
+            params![id],
+            |r| {
+                Ok((
+                    r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?,
+                    r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?,
+                ))
+            },
+        )
+        .optional()?;
+    let (title, filename, mime, size_bytes, status, transcript, summary, error, created_at) =
+        row.ok_or(AppError::NotFound)?;
+    Ok(JsonResp(RecordingDetail {
+        id, title, filename, mime, size_bytes, status,
+        transcript, summary, error, created_at,
+    }))
+}
+
+async fn upload_recording(
+    State(s): State<AppState>,
+    mut form: Multipart,
+) -> Result<JsonResp<Value>, AppError> {
+    let mut title: Option<String> = None;
+    let mut filename: Option<String> = None;
+    let mut mime: Option<String> = None;
+    let mut tmp_path: Option<PathBuf> = None;
+    let mut size: usize = 0;
+
+    while let Some(mut field) = form
+        .next_field()
+        .await
+        .map_err(|e| AppError::bad_request(format!("multipart: {e}")))?
+    {
+        let name = field.name().unwrap_or("").to_string();
+        match name.as_str() {
+            "title" => {
+                let s = field
+                    .text()
+                    .await
+                    .map_err(|e| AppError::bad_request(format!("title: {e}")))?;
+                title = Some(s.trim().to_string());
+            }
+            "audio" | "file" => {
+                let fn_ = field
+                    .file_name()
+                    .map(|s| s.to_string())
+                    .unwrap_or_else(|| "recording".to_string());
+                let m = field
+                    .content_type()
+                    .map(|s| s.to_string())
+                    .unwrap_or_else(|| "application/octet-stream".to_string());
+                if !m.starts_with("audio/") && !m.starts_with("video/") && m != "application/octet-stream" {
+                    return Err(AppError::bad_request(format!("unsupported mime '{m}'")));
+                }
+                filename = Some(fn_);
+                mime = Some(m);
+                // 流式落 tmp
+                let tmp = s.blobs_dir.join(format!("upload-{}.tmp", std::process::id()));
+                let mut f = tokio::fs::File::create(&tmp).await.map_err(AppError::Io)?;
+                while let Some(chunk) = field
+                    .chunk()
+                    .await
+                    .map_err(|e| AppError::bad_request(format!("upload: {e}")))?
+                {
+                    size += chunk.len();
+                    if size > SINGLE_FILE_BYTES {
+                        let _ = tokio::fs::remove_file(&tmp).await;
+                        return Err(AppError::bad_request(format!(
+                            "file too large (>{SINGLE_FILE_BYTES} bytes)"
+                        )));
+                    }
+                    f.write_all(&chunk).await.map_err(AppError::Io)?;
+                }
+                f.sync_all().await.map_err(AppError::Io)?;
+                tmp_path = Some(tmp);
+            }
+            _ => {}
+        }
+    }
+
+    let filename = filename.ok_or_else(|| AppError::bad_request("missing audio file"))?;
+    let mime = mime.unwrap_or_else(|| "audio/mpeg".to_string());
+    let tmp_path = tmp_path.ok_or_else(|| AppError::bad_request("no file uploaded"))?;
+    let title = title
+        .filter(|x| !x.is_empty())
+        .unwrap_or_else(|| filename.clone());
+
+    let id = {
+        let conn = s.db.lock().unwrap();
+        conn.execute(
+            "INSERT INTO recordings (title, filename, mime, size_bytes, status)
+             VALUES (?1, ?2, ?3, ?4, 'pending')",
+            params![title, filename, mime, size as i64],
+        )?;
+        conn.last_insert_rowid()
+    };
+
+    let final_path = s.blobs_dir.join(id.to_string());
+    if let Err(e) = tokio::fs::rename(&tmp_path, &final_path).await {
+        let _ = tokio::fs::remove_file(&tmp_path).await;
+        let conn = s.db.lock().unwrap();
+        let _ = conn.execute("DELETE FROM recordings WHERE id = ?1", params![id]);
+        return Err(AppError::Io(e));
+    }
+
+    // 后台处理
+    let state_clone = s.clone();
+    tokio::spawn(async move {
+        process_recording(state_clone, id).await;
+    });
+
+    Ok(JsonResp(json!({ "id": id, "status": "pending" })))
+}
+
+async fn process_recording(s: AppState, id: i64) {
+    set_status(&s, id, "transcribing", None, None);
+    let path = s.blobs_dir.join(id.to_string());
+    let filename: String = {
+        let conn = s.db.lock().unwrap();
+        conn.query_row(
+            "SELECT filename FROM recordings WHERE id = ?1",
+            params![id],
+            |r| r.get(0),
+        )
+        .unwrap_or_else(|_| "audio".to_string())
+    };
+
+    // ASR：multipart POST，OpenAI Whisper 风格
+    let transcript = match call_asr(&s, &path, &filename).await {
+        Ok(t) => t,
+        Err(e) => {
+            tracing::error!(%id, error = %e, "ASR failed");
+            set_status(&s, id, "failed", None, Some(&format!("ASR: {e}")));
+            return;
+        }
+    };
+    // 写 transcript 但还没 summary
+    {
+        let conn = s.db.lock().unwrap();
+        let _ = conn.execute(
+            "UPDATE recordings SET transcript = ?1, status = 'summarizing' WHERE id = ?2",
+            params![&transcript, id],
+        );
+    }
+
+    // LLM：生成会议纪要
+    let summary = match call_llm_summary(&s, &transcript).await {
+        Ok(t) => t,
+        Err(e) => {
+            tracing::error!(%id, error = %e, "LLM failed");
+            set_status(&s, id, "failed", None, Some(&format!("LLM: {e}")));
+            return;
+        }
+    };
+    {
+        let conn = s.db.lock().unwrap();
+        let _ = conn.execute(
+            "UPDATE recordings SET summary = ?1, status = 'done', error = NULL WHERE id = ?2",
+            params![&summary, id],
+        );
+    }
+    tracing::info!(%id, "done");
+}
+
+fn set_status(s: &AppState, id: i64, status: &str, transcript: Option<&str>, error: Option<&str>) {
+    let conn = s.db.lock().unwrap();
+    let _ = conn.execute(
+        "UPDATE recordings SET status = ?1, error = ?2,
+            transcript = COALESCE(?3, transcript)
+         WHERE id = ?4",
+        params![status, error, transcript, id],
+    );
+}
+
+async fn call_asr(
+    s: &AppState,
+    path: &std::path::Path,
+    filename: &str,
+) -> Result<String, String> {
+    let bytes = tokio::fs::read(path).await.map_err(|e| e.to_string())?;
+    let part = reqwest::multipart::Part::bytes(bytes)
+        .file_name(filename.to_string())
+        .mime_str("audio/mpeg")
+        .map_err(|e| e.to_string())?;
+    let form = reqwest::multipart::Form::new()
+        .text("model", "qwen3-asr")
+        .text("response_format", "json")
+        .part("file", part);
+
+    let resp = s
+        .http
+        .post(&s.asr_url)
+        .bearer_auth(&s.asr_token)
+        .multipart(form)
+        .timeout(std::time::Duration::from_secs(600))
+        .send()
+        .await
+        .map_err(|e| format!("connect: {e}"))?;
+    if !resp.status().is_success() {
+        let st = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+        return Err(format!("ASR {st}: {body}"));
+    }
+    let v: Value = resp.json().await.map_err(|e| format!("decode: {e}"))?;
+    let text = v
+        .get("text")
+        .and_then(|x| x.as_str())
+        .map(|s| s.to_string())
+        .ok_or_else(|| format!("ASR response no 'text': {v}"))?;
+    Ok(text)
+}
+
+async fn call_llm_summary(s: &AppState, transcript: &str) -> Result<String, String> {
+    let trimmed = if transcript.chars().count() > 12000 {
+        let mut out = String::new();
+        for (i, c) in transcript.chars().enumerate() {
+            if i >= 12000 { break; }
+            out.push(c);
+        }
+        out + "\n\n[... 后文截断]"
+    } else {
+        transcript.to_string()
+    };
+    let payload = json!({
+        "model": s.llm_model,
+        "messages": [
+            { "role": "system", "content":
+                "你是一个会议纪要助手。根据语音转写整理一份结构化纪要（markdown）：\n\
+                 1. **概要**：1-2 句话总结\n\
+                 2. **关键讨论点**：bullet 列出\n\
+                 3. **决定 / 结论**\n\
+                 4. **行动项 (action items)**：谁、做什么、何时\n\
+                 5. **待跟进 / 未决问题**\n\
+                 转写可能有 ASR 错字，结合上下文合理修正；遇到模糊处标 [？]。" },
+            { "role": "user", "content": trimmed },
+        ],
+        "temperature": 0.3,
+    });
+    let url = format!("{}/chat/completions", s.llm_gateway.trim_end_matches('/'));
+    let resp = s
+        .http
+        .post(&url)
+        .bearer_auth(&s.llm_token)
+        .json(&payload)
+        .timeout(std::time::Duration::from_secs(300))
+        .send()
+        .await
+        .map_err(|e| format!("connect: {e}"))?;
+    if !resp.status().is_success() {
+        let st = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+        return Err(format!("LLM {st}: {body}"));
+    }
+    let v: Value = resp.json().await.map_err(|e| format!("decode: {e}"))?;
+    let text = v
+        .get("choices").and_then(|c| c.get(0))
+        .and_then(|c| c.get("message"))
+        .and_then(|m| m.get("content"))
+        .and_then(|c| c.as_str())
+        .map(|s| s.to_string())
+        .ok_or_else(|| format!("LLM no content: {v}"))?;
+    Ok(text)
+}
+
+async fn delete_recording(
+    State(s): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<JsonResp<Value>, AppError> {
+    let n = {
+        let conn = s.db.lock().unwrap();
+        conn.execute("DELETE FROM recordings WHERE id = ?1", params![id])?
+    };
+    if n == 0 {
+        return Err(AppError::NotFound);
+    }
+    let _ = tokio::fs::remove_file(s.blobs_dir.join(id.to_string())).await;
+    Ok(JsonResp(json!({ "ok": true })))
+}
+
+async fn retry_recording(
+    State(s): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<JsonResp<Value>, AppError> {
+    {
+        let conn = s.db.lock().unwrap();
+        let exists: bool = conn
+            .query_row("SELECT 1 FROM recordings WHERE id = ?1", params![id], |_| Ok(true))
+            .optional()?
+            .unwrap_or(false);
+        if !exists { return Err(AppError::NotFound); }
+        conn.execute(
+            "UPDATE recordings SET status = 'pending', error = NULL WHERE id = ?1",
+            params![id],
+        )?;
+    }
+    let sc = s.clone();
+    tokio::spawn(async move { process_recording(sc, id).await; });
+    Ok(JsonResp(json!({ "ok": true, "status": "pending" })))
+}
+
+async fn stream_audio(
+    State(s): State<AppState>,
+    Path(id): Path<i64>,
+    req: Request<Body>,
+) -> Result<Response, AppError> {
+    let row: Option<(String, String)> = {
+        let conn = s.db.lock().unwrap();
+        conn.query_row(
+            "SELECT mime, filename FROM recordings WHERE id = ?1",
+            params![id],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .optional()?
+    };
+    let (mime, _filename) = row.ok_or(AppError::NotFound)?;
+    let path = s.blobs_dir.join(id.to_string());
+    let mime_hv: header::HeaderValue = mime
+        .parse()
+        .unwrap_or_else(|_| header::HeaderValue::from_static("application/octet-stream"));
+    let svc = tower_http::services::ServeFile::new(&path);
+    let mut resp = svc
+        .oneshot(req)
+        .await
+        .map_err(|e| AppError::Io(std::io::Error::other(e.to_string())))?
+        .into_response();
+    resp.headers_mut().insert(header::CONTENT_TYPE, mime_hv);
+    Ok(resp)
+}
+
+// ---------- error type ----------
+
+enum AppError {
+    BadRequest(String),
+    NotFound,
+    Db(rusqlite::Error),
+    Io(std::io::Error),
+}
+impl AppError {
+    fn bad_request(m: impl Into<String>) -> Self { Self::BadRequest(m.into()) }
+}
+impl From<rusqlite::Error> for AppError {
+    fn from(e: rusqlite::Error) -> Self { Self::Db(e) }
+}
+impl IntoResponse for AppError {
+    fn into_response(self) -> Response {
+        match self {
+            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
+            Self::NotFound => (StatusCode::NOT_FOUND, "not found").into_response(),
+            Self::Db(e) => {
+                tracing::error!(error = %e, "db");
+                (StatusCode::INTERNAL_SERVER_ERROR, "db error").into_response()
+            }
+            Self::Io(e) => {
+                tracing::error!(error = %e, "io");
+                (StatusCode::INTERNAL_SERVER_ERROR, "io error").into_response()
+            }
+        }
+    }
+}