fix(voice): restore failed transcription recovery

2026-07-22 15:35:48 +08:00
parent c54f4214df
commit 146e78e335
15 changed files with 603 additions and 64 deletions
--- a/server/src/recording_routes.rs
+++ b/server/src/recording_routes.rs
@@ -3,7 +3,7 @@ use std::{collections::HashMap, sync::Arc};
 use axum::{
    Json,
    extract::{Multipart, Path, Query, State},
-    http::HeaderMap,
+    http::{HeaderMap, StatusCode},
 };
 use chrono::{DateTime, NaiveDate, Utc};
 use serde::{Deserialize, Serialize};
@@ -15,6 +15,7 @@ use uuid::Uuid;
 use crate::{
    AppState,
    error::{ApiError, CommonApiErrors, ErrorBody},
+    transcription_worker::transcription_error_class,
 };

 type SharedState = Arc<AppState>;
@@ -34,6 +35,7 @@ pub fn router() -> OpenApiRouter<SharedState> {
        .routes(routes!(finish_lesson_session))
        .routes(routes!(mark_lesson_applied))
        .routes(routes!(retry_audio_segment))
+        .routes(routes!(discard_audio_segment))
        .routes(routes!(generate_feedback_content))
 }

@@ -94,6 +96,26 @@ struct SegmentRow {
    error_message: Option<String>,
 }

+#[derive(Debug, FromRow)]
+struct RetryableSegmentRow {
+    sequence: i32,
+    duration_ms: i32,
+    audio_path: String,
+    lesson_id: Uuid,
+    feedback_session_id: Uuid,
+    transcription_attempts: i32,
+    error_message: Option<String>,
+}
+
+#[derive(Debug, FromRow)]
+struct GenerationBlockingSegmentRow {
+    id: Uuid,
+    lesson_id: Uuid,
+    status: String,
+    transcription_attempts: i32,
+    error_message: Option<String>,
+}
+
 #[derive(Debug, Serialize, ToSchema)]
 struct FeedbackSessionResponse {
    id: Uuid,
@@ -665,19 +687,23 @@ async fn retry_audio_segment(
        .authenticate(state.pool.as_ref(), &headers)
        .await?
        .user_id;
-    let segment = sqlx::query_as::<_, (i32, i32, String, Uuid)>(
-        "SELECT s.sequence, s.duration_ms, s.audio_path, l.id \
+    let mut transaction = pool(&state)?.begin().await?;
+    let segment = sqlx::query_as::<_, RetryableSegmentRow>(
+        "SELECT s.sequence, s.duration_ms, s.audio_path, l.id AS lesson_id, \
+                f.id AS feedback_session_id, s.transcription_attempts, s.error_message \
         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'",
+         WHERE s.id = $1 AND f.owner_id = $2 AND f.status = 'active' \
+           AND s.status = 'failed' \
+         FOR UPDATE OF s",
    )
    .bind(segment_id)
    .bind(owner_id)
-    .fetch_optional(pool(&state)?)
+    .fetch_optional(&mut *transaction)
    .await?
    .ok_or(ApiError::NotFound)?;
-    tokio::fs::metadata(&segment.2)
+    tokio::fs::metadata(&segment.audio_path)
        .await
        .map_err(internal_io_error)?;
    sqlx::query(
@@ -685,30 +711,144 @@ async fn retry_audio_segment(
         processing_started_at = NULL, updated_at = now() WHERE id = $1",
    )
    .bind(segment_id)
-    .execute(pool(&state)?)
+    .execute(&mut *transaction)
    .await?;
    sqlx::query(
        "UPDATE lesson_sessions SET status = 'processing', updated_at = now() WHERE id = $1",
    )
-    .bind(segment.3)
-    .execute(pool(&state)?)
+    .bind(segment.lesson_id)
+    .execute(&mut *transaction)
    .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?;
+    sqlx::query("UPDATE feedback_sessions SET updated_at = now() WHERE id = $1")
+        .bind(segment.feedback_session_id)
+        .execute(&mut *transaction)
+        .await?;
+    transaction.commit().await?;
+    tracing::info!(
+        event = "transcription_retry_requested",
+        feedback_session_id = %segment.feedback_session_id,
+        lesson_id = %segment.lesson_id,
+        segment_id = %segment_id,
+        attempt = segment.transcription_attempts + 1,
+        previous_attempt = segment.transcription_attempts,
+        error_class = transcription_error_class(segment.error_message.as_deref().unwrap_or("")),
+        status = "processing",
+        "failed audio segment returned to transcription queue"
+    );
    Ok(Json(AudioSegmentResponse {
        id: segment_id,
-        sequence: segment.0,
-        duration_ms: segment.1,
+        sequence: segment.sequence,
+        duration_ms: segment.duration_ms,
        status: "processing".to_owned(),
        error_message: None,
    }))
 }

+#[utoipa::path(
+    delete,
+    path = "/api/v1/audio-segments/{segment_id}",
+    tag = "语音反馈",
+    summary = "放弃失败的录音片段",
+    description = "删除无法使用的失败录音。若课堂不再包含片段，则同时删除空课堂。",
+    params(
+        ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer <access-token>"),
+        ("segment_id" = Uuid, Path, description = "失败录音片段 ID")
+    ),
+    responses(
+        (status = 204, description = "失败片段已放弃"),
+        (status = 404, description = "失败片段不存在或不属于当前用户", body = ErrorBody),
+        CommonApiErrors
+    )
+)]
+async fn discard_audio_segment(
+    State(state): State<SharedState>,
+    headers: HeaderMap,
+    Path(segment_id): Path<Uuid>,
+) -> Result<StatusCode, ApiError> {
+    let owner_id = state
+        .auth
+        .authenticate(state.pool.as_ref(), &headers)
+        .await?
+        .user_id;
+    let mut transaction = pool(&state)?.begin().await?;
+    let segment = sqlx::query_as::<_, RetryableSegmentRow>(
+        "SELECT s.sequence, s.duration_ms, s.audio_path, l.id AS lesson_id, \
+                f.id AS feedback_session_id, s.transcription_attempts, s.error_message \
+         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' \
+           AND s.status = 'failed' \
+         FOR UPDATE OF s",
+    )
+    .bind(segment_id)
+    .bind(owner_id)
+    .fetch_optional(&mut *transaction)
+    .await?
+    .ok_or(ApiError::NotFound)?;
+
+    sqlx::query("DELETE FROM audio_segments WHERE id = $1 AND status = 'failed'")
+        .bind(segment_id)
+        .execute(&mut *transaction)
+        .await?;
+    let remaining_segment_count = sqlx::query_scalar::<_, i64>(
+        "SELECT COUNT(*) FROM audio_segments WHERE lesson_session_id = $1",
+    )
+    .bind(segment.lesson_id)
+    .fetch_one(&mut *transaction)
+    .await?;
+    if remaining_segment_count == 0 {
+        sqlx::query("DELETE FROM lesson_sessions WHERE id = $1")
+            .bind(segment.lesson_id)
+            .execute(&mut *transaction)
+            .await?;
+    } else {
+        sqlx::query(
+            "UPDATE lesson_sessions 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 = (SELECT COALESCE(SUM(duration_ms), 0) FROM audio_segments WHERE lesson_session_id = $1), \
+               updated_at = now() \
+             WHERE id = $1",
+        )
+        .bind(segment.lesson_id)
+        .execute(&mut *transaction)
+        .await?;
+    }
+    sqlx::query("UPDATE feedback_sessions SET updated_at = now() WHERE id = $1")
+        .bind(segment.feedback_session_id)
+        .execute(&mut *transaction)
+        .await?;
+    transaction.commit().await?;
+
+    if let Err(error) = tokio::fs::remove_file(&segment.audio_path).await
+        && error.kind() != std::io::ErrorKind::NotFound
+    {
+        tracing::warn!(
+            event = "audio_storage_delete_failed",
+            feedback_session_id = %segment.feedback_session_id,
+            lesson_id = %segment.lesson_id,
+            segment_id = %segment_id,
+            error_class = "audio_storage_delete_failed",
+            "discarded audio file could not be removed from storage"
+        );
+    }
+    tracing::info!(
+        event = "failed_transcription_discarded",
+        feedback_session_id = %segment.feedback_session_id,
+        lesson_id = %segment.lesson_id,
+        segment_id = %segment_id,
+        attempt = segment.transcription_attempts,
+        error_class = transcription_error_class(segment.error_message.as_deref().unwrap_or("")),
+        status = "discarded",
+        "failed audio segment discarded"
+    );
+    Ok(StatusCode::NO_CONTENT)
+}
+
 #[utoipa::path(
    post,
    path = "/api/v1/feedback-sessions/{session_id}/generate",
@@ -739,22 +879,30 @@ async fn generate_feedback_content(
        .user_id;
    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')",
+    let blocking_segments = sqlx::query_as::<_, GenerationBlockingSegmentRow>(
+        "SELECT s.id, l.id AS lesson_id, s.status, s.transcription_attempts, s.error_message \
+         FROM audio_segments s \
+         JOIN lesson_sessions l ON l.id = s.lesson_session_id \
+         WHERE l.feedback_session_id = $1 AND s.status IN ('processing', 'failed') \
+         ORDER BY l.sequence, s.sequence",
    )
    .bind(session_id)
-    .fetch_one(pool(&state)?)
+    .fetch_all(pool(&state)?)
    .await?;
-    if pending_status.0 {
+    if blocking_segments
+        .iter()
+        .any(|segment| segment.status == "processing")
+    {
+        log_generation_rejection(session_id, &blocking_segments, "processing");
        return Err(ApiError::BadRequest(
            "录音仍在后台转录中，请稍后再生成反馈".to_owned(),
        ));
    }
-    if pending_status.1 {
+    if blocking_segments
+        .iter()
+        .any(|segment| segment.status == "failed")
+    {
+        log_generation_rejection(session_id, &blocking_segments, "failed");
        return Err(ApiError::BadRequest(
            "部分录音转录失败，请重试后再生成反馈".to_owned(),
        ));
@@ -979,6 +1127,34 @@ fn internal_io_error(_: std::io::Error) -> ApiError {
    ApiError::Internal
 }

+fn log_generation_rejection(
+    session_id: Uuid,
+    blocking_segments: &[GenerationBlockingSegmentRow],
+    blocking_status: &str,
+) {
+    for segment in blocking_segments
+        .iter()
+        .filter(|segment| segment.status == blocking_status)
+    {
+        let error_class = if segment.status == "processing" {
+            "transcription_processing"
+        } else {
+            transcription_error_class(segment.error_message.as_deref().unwrap_or(""))
+        };
+        tracing::warn!(
+            event = "feedback_generation_rejected",
+            feedback_session_id = %session_id,
+            lesson_id = %segment.lesson_id,
+            segment_id = %segment.id,
+            attempt = segment.transcription_attempts,
+            status = %segment.status,
+            error_class,
+            rejection_reason = %blocking_status,
+            "feedback generation blocked by audio transcription state"
+        );
+    }
+}
+
 fn validate_draft_content(content: &str) -> Result<(), ApiError> {
    if content.chars().count() > 2000 {
        return Err(ApiError::BadRequest(
--- a/server/src/routes.rs
+++ b/server/src/routes.rs
@@ -924,7 +924,7 @@ mod tests {
            .filter(|operation| operation.get("responses").is_some())
            .collect::<Vec<_>>();

-        assert_eq!(operations.len(), 23);
+        assert_eq!(operations.len(), 24);
        for operation in &operations {
            assert_non_empty(operation, "summary");
            assert_non_empty(operation, "description");
--- a/server/src/transcription_worker.rs
+++ b/server/src/transcription_worker.rs
@@ -335,10 +335,13 @@ async fn fail_job(
    Ok(())
 }

-fn transcription_error_class(message: &str) -> &'static str {
-    if message.contains("连接失败") || message.contains("unavailable") {
+pub(crate) fn transcription_error_class(message: &str) -> &'static str {
+    let normalized = message.to_lowercase();
+    if normalized.contains("no clear speech") || message.contains("未识别到清晰语音") {
+        "no_clear_speech"
+    } else if message.contains("连接失败") || normalized.contains("unavailable") {
        "asr_unavailable"
-    } else if message.contains("HTTP") {
+    } else if normalized.contains("http") {
        "asr_http_error"
    } else if message.contains("读取录音") {
        "audio_read_failed"
@@ -379,3 +382,24 @@ async fn refresh_lesson(pool: &PgPool, lesson_id: Uuid) -> Result<(), sqlx::Erro
    .await?;
    Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::transcription_error_class;
+
+    #[test]
+    fn no_clear_speech_is_classified_before_http_errors() {
+        assert_eq!(
+            transcription_error_class("FunASR 返回 HTTP 422: no clear speech was recognized"),
+            "no_clear_speech"
+        );
+    }
+
+    #[test]
+    fn generic_http_error_keeps_provider_error_class() {
+        assert_eq!(
+            transcription_error_class("FunASR 返回 HTTP 500"),
+            "asr_http_error"
+        );
+    }
+}