Add summary and mindmap logic

app/api/transcribe.py

@@ -11,12 +11,21 @@ from app.core.audio_utils import save_upload_file, get_audio_info, ensure_wav_16
 from app.core.asr_engine import load_model, transcribe_file, transcribe_file_chunks
 from app.config import settings
 from app.services.text_normalizer import normalize_text
+from app.services.nlp_postprocess import normalize_and_extract
+from app.services.summary_service import generate_summary
+from app.services.mindmap_service import generate_mindmap
 from app.services.note_client import NoteServiceClient
 from rq import Queue
 from app.infra.redis_client import redis_client
 from app.jobs.transcribe_job import transcribe_job
 from app.schemas.transcribe import TranscribeResponse
-from app.infra.metrics import REQUEST_COUNT, REQUEST_LATENCY, ASR_DURATION, NORMALIZE_DURATION, ERROR_COUNT
+from app.infra.metrics import (
+    REQUEST_COUNT,
+    REQUEST_LATENCY,
+    ASR_DURATION,
+    NORMALIZE_DURATION,
+    ERROR_COUNT,
+)
 
 router = APIRouter()
 
@@ -32,10 +41,17 @@ async def _startup():
 
 def _ensure_file_limits(path: str):
     if os.path.getsize(path) > settings.MAX_UPLOAD_BYTES:
-        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File size exceeds limit")
+        raise HTTPException(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            detail="File size exceeds limit",
+        )
     info = get_audio_info(path)
     if info and info.get("duration", 0) > settings.MAX_DURATION_SECS:
-        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="Audio duration exceeds limit")
+        raise HTTPException(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            detail="Audio duration exceeds limit",
+        )
+    
 
 @router.post("/transcribe", response_model=TranscribeResponse)
 async def transcribe(file: UploadFile = File(...)):
@@ -43,9 +59,11 @@ async def transcribe(file: UploadFile = File(...)):
     tmp_wav = None
     note_service = NoteServiceClient()
     note_id = str(uuid.uuid4())
+
     start_time = time.perf_counter()
     endpoint = "/transcribe"
     status_label = "success"
+
     with REQUEST_LATENCY.labels(endpoint).time():
         try:
             # write upload to tmp (blocking) -> run in thread
@@ -61,6 +79,7 @@ async def transcribe(file: UploadFile = File(...)):
             info = get_audio_info(tmp_wav) or {}
             duration_sec = info.get("duration", 0)
             ASYNC_THRESHOLD = 120  # 2 phút, có thể chỉnh
+            # ---------- ASYNC JOB ----------
             if duration_sec > ASYNC_THRESHOLD:
                 # Enqueue background job bằng RQ
                 q = Queue("asr", connection=redis_client)
@@ -78,7 +97,7 @@ async def transcribe(file: UploadFile = File(...)):
                     "status": "queued",
                     "duration": duration_sec
                 })
-
+            # ---------- SYNC PIPELINE ----------
             # Nếu audio ngắn, xử lý sync như cũ
             model = ASR_MODEL or await asyncio.to_thread(load_model, 30)
             with ASR_DURATION.labels(endpoint).time():
@@ -87,7 +106,13 @@ async def transcribe(file: UploadFile = File(...)):
 
             # normalize via Gemini (already async safe in your service)
             with NORMALIZE_DURATION.labels(endpoint).time():
-                normalized_text = await normalize_text(text)
+                # normalized_text = await normalize_text(text)
+                nlp = await normalize_and_extract(text)
+                normalized_text = nlp["normalized_text"]
+                keywords = nlp["keywords"]
+
+                summary = await generate_summary(normalized_text)
+                mindmap = await generate_mindmap(normalized_text)
 
             info2 = get_audio_info(tmp_wav) or {}
             # persist to Note Service (async HTTP)
@@ -95,30 +120,39 @@ async def transcribe(file: UploadFile = File(...)):
                 note_id=note_id,
                 raw_text=text,
                 normalized_text=normalized_text,
+                keywords=keywords,
+                summary=summary,
+                mindmap=mindmap,
                 duration=info2.get("duration"),
                 sample_rate=info2.get("samplerate"),
                 chunks=chunks,
                 asr_model="PhoWhisper-base",
-                normalization_model="gemini-1.5"
+                normalization_model="gemini-1.5",
             )
 
             duration = time.perf_counter() - start_time
             logging.info(f"/transcribe success note_id={note_id} duration={duration:.2f}s audio_dur={info2.get('duration')}")
             REQUEST_COUNT.labels(endpoint, status_label).inc()
-            return JSONResponse(status_code=200, content={
-                "note_id": note_id,
-                "status": "transcribed",
-                "duration": info2.get("duration")
-            })
+            return JSONResponse(
+                status_code=200,
+                content={
+                    "note_id": note_id,
+                    "status": "transcribed",
+                    "duration": info2.get("duration"),
+                },
+            )
+        
         except HTTPException:
             status_label = "http_error"
             ERROR_COUNT.labels(endpoint, status_label).inc()
             raise
+
         except Exception as e:
             status_label = "error"
             ERROR_COUNT.labels(endpoint, status_label).inc()
             logging.exception(f"/transcribe failed note_id={note_id}")
             raise HTTPException(status_code=500, detail=f"Transcription failed: {e}")
+        
         finally:
             # cleanup
             for p in [tmp_in, tmp_wav]:
@@ -128,11 +162,11 @@ async def transcribe(file: UploadFile = File(...)):
                 except Exception:
                     pass
 
-
 @router.post("/transcribe-url", response_model=TranscribeResponse)
 async def transcribe_url(payload: dict):
     audio_url = payload.get("audio_url")
     user_id = payload.get("user_id")
+
     if not audio_url:
         raise HTTPException(status_code=400, detail="audio_url required")
     if not user_id:
@@ -140,52 +174,193 @@ async def transcribe_url(payload: dict):
 
     tmp_in = make_temp_path(suffix=Path(audio_url).suffix or ".tmp")
     tmp_wav = None
-    note_service = NoteServiceClient()
     note_id = str(uuid.uuid4())
+    note_service = NoteServiceClient()
 
+    endpoint = "/transcribe-url"
     start_time = time.perf_counter()
-    try:
-        # download blocking -> thread
-        await asyncio.to_thread(download_file_from_url, audio_url, tmp_in)
-
-        _ensure_file_limits(tmp_in)
-
-        tmp_wav = make_temp_path(suffix=".wav")
-        await asyncio.to_thread(ensure_wav_16k_mono, tmp_in, tmp_wav)
-
-        model = ASR_MODEL or await asyncio.to_thread(load_model, 30)
-        text = await asyncio.to_thread(transcribe_file, model, tmp_wav, 30.0, 5.0)
-        chunks = await asyncio.to_thread(transcribe_file_chunks, model, tmp_wav, 30.0, 5.0)
-        normalized_text = await normalize_text(text)
-        info2 = get_audio_info(tmp_wav) or {}
-
-        await note_service.save_transcript(
-            note_id=note_id,
-            raw_text=text,
-            normalized_text=normalized_text,
-            duration=info2.get("duration"),
-            sample_rate=info2.get("samplerate"),
-            chunks=chunks,
-            asr_model="PhoWhisper-base",
-            normalization_model="gemini-1.5"
-        )
+    status_label = "success"
+
+    with REQUEST_LATENCY.labels(endpoint).time():
+        try:
+            # 1. Download from Cloudinary (blocking)
+            await asyncio.to_thread(download_file_from_url, audio_url, tmp_in)
+
+            # 2. File & duration limits
+            _ensure_file_limits(tmp_in)
+
+            # 3. Convert to wav 16k mono
+            tmp_wav = make_temp_path(suffix=".wav")
+            await asyncio.to_thread(ensure_wav_16k_mono, tmp_in, tmp_wav)
+
+            # 4. Check duration for sync / async
+            info = get_audio_info(tmp_wav) or {}
+            duration_sec = info.get("duration", 0)
+            ASYNC_THRESHOLD = 120  # seconds
+
+            # ---------- ASYNC JOB ----------
+            if duration_sec > ASYNC_THRESHOLD:
+                q = Queue("asr", connection=redis_client)
+                job = q.enqueue(
+                    transcribe_job,
+                    tmp_wav,
+                    note_id,
+                    job_timeout=1800,
+                )
+
+                logging.info(
+                    f"/transcribe-url queued note_id={note_id} "
+                    f"job_id={job.id} duration={duration_sec:.1f}s"
+                )
+                REQUEST_COUNT.labels(endpoint, "queued").inc()
+
+                return JSONResponse(
+                    status_code=202,
+                    content={
+                        "note_id": note_id,
+                        "job_id": job.id,
+                        "status": "queued",
+                        "duration": duration_sec,
+                    },
+                )
+
+            # ---------- SYNC PIPELINE ----------
+            model = ASR_MODEL or await asyncio.to_thread(load_model, 30)
+
+            with ASR_DURATION.labels(endpoint).time():
+                text = await asyncio.to_thread(
+                    transcribe_file, model, tmp_wav, 30.0, 5.0
+                )
+                chunks = await asyncio.to_thread(
+                    transcribe_file_chunks, model, tmp_wav, 30.0, 5.0
+                )
+
+            with NORMALIZE_DURATION.labels(endpoint).time():
+                nlp = await normalize_and_extract(text)
+                normalized_text = nlp["normalized_text"]
+                keywords = nlp["keywords"]
+
+                summary = await generate_summary(normalized_text)
+                mindmap = await generate_mindmap(normalized_text)
+
+            # 5. Persist to Note Service
+            await note_service.save_transcript(
+                note_id=note_id,
+                raw_text=text,
+                normalized_text=normalized_text,
+                keywords=keywords,
+                summary=summary,
+                mindmap=mindmap,
+                duration=info.get("duration"),
+                sample_rate=info.get("samplerate"),
+                chunks=chunks,
+                asr_model="PhoWhisper-base",
+                normalization_model="gemini-1.5",
+            )
+
+            duration = time.perf_counter() - start_time
+            logging.info(
+                f"/transcribe-url success note_id={note_id} "
+                f"duration={duration:.2f}s audio_dur={info.get('duration')}"
+            )
+
+            REQUEST_COUNT.labels(endpoint, status_label).inc()
+            return JSONResponse(
+                status_code=200,
+                content={
+                    "note_id": note_id,
+                    "status": "transcribed",
+                    "duration": info.get("duration"),
+                },
+            )
+
+        except HTTPException:
+            status_label = "http_error"
+            ERROR_COUNT.labels(endpoint, status_label).inc()
+            raise
+
+        except Exception as e:
+            status_label = "error"
+            ERROR_COUNT.labels(endpoint, status_label).inc()
+            logging.exception(f"/transcribe-url failed note_id={note_id}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+        finally:
+            for p in [tmp_in, tmp_wav]:
+                try:
+                    if p and os.path.exists(p):
+                        os.remove(p)
+                except Exception:
+                    pass
+
+# @router.post("/transcribe-url", response_model=TranscribeResponse)
+# async def transcribe_url(payload: dict):
+#     audio_url = payload.get("audio_url")
+#     user_id = payload.get("user_id")
+#     if not audio_url:
+#         raise HTTPException(status_code=400, detail="audio_url required")
+#     if not user_id:
+#         raise HTTPException(status_code=400, detail="user_id required")
+
+#     tmp_in = make_temp_path(suffix=Path(audio_url).suffix or ".tmp")
+#     tmp_wav = None
+#     note_service = NoteServiceClient()
+#     note_id = str(uuid.uuid4())
+
+#     start_time = time.perf_counter()
+#     try:
+#         # download blocking -> thread
+#         await asyncio.to_thread(download_file_from_url, audio_url, tmp_in)
+
+#         _ensure_file_limits(tmp_in)
+
+#         tmp_wav = make_temp_path(suffix=".wav")
+#         await asyncio.to_thread(ensure_wav_16k_mono, tmp_in, tmp_wav)
+
+#         model = ASR_MODEL or await asyncio.to_thread(load_model, 30)
+#         text = await asyncio.to_thread(transcribe_file, model, tmp_wav, 30.0, 5.0)
+#         chunks = await asyncio.to_thread(transcribe_file_chunks, model, tmp_wav, 30.0, 5.0)
+
+#         # NLP pipeline: normalize, extract keywords, then summary and mindmap
+#         nlp = await normalize_and_extract(text)
+#         normalized_text = nlp.get("normalized_text", text)
+#         keywords = nlp.get("keywords", [])
+
+#         summary = await generate_summary(normalized_text)
+#         mindmap = await generate_mindmap(normalized_text)
+
+#         info2 = get_audio_info(tmp_wav) or {}
+
+#         await note_service.save_transcript(
+#             note_id=note_id,
+#             raw_text=text,
+#             normalized_text=normalized_text,
+#             keywords=keywords,
+#             summary=summary,
+#             mindmap=mindmap,
+#             duration=info2.get("duration"),
+#             sample_rate=info2.get("samplerate"),
+#             chunks=chunks,
+#             asr_model="PhoWhisper-base",
+#             normalization_model="gemini-1.5"
+#         )
 
-        duration = time.perf_counter() - start_time
-        logging.info(f"/transcribe-url success note_id={note_id} duration={duration:.2f}s audio_dur={info2.get('duration')}")
-        return JSONResponse(status_code=200, content={
-            "note_id": note_id,
-            "status": "transcribed",
-            "duration": info2.get("duration")
-        })
-    except HTTPException:
-        raise
-    except Exception as e:
-        logging.exception(f"/transcribe-url failed note_id={note_id}")
-        raise HTTPException(status_code=500, detail=f"Transcription failed: {e}")
-    finally:
-        for p in [tmp_in, tmp_wav]:
-            try:
-                if p and os.path.exists(p):
-                    os.remove(p)
-            except Exception:
-                pass
+#         duration = time.perf_counter() - start_time
+#         logging.info(f"/transcribe-url success note_id={note_id} duration={duration:.2f}s audio_dur={info2.get('duration')}")
+#         return JSONResponse(status_code=200, content={
+#             "note_id": note_id,
+#             "status": "transcribed",
+#             "duration": info2.get("duration")
+#         })
+#     except HTTPException:
+#         raise
+#     except Exception as e:
+#         logging.exception(f"/transcribe-url failed note_id={note_id}")
+#         raise HTTPException(status_code=500, detail=f"Transcription failed: {e}")
+#     finally:
+#         for p in [tmp_in, tmp_wav]:
+#             try:
+#                 if p and os.path.exists(p):
+#                     os.remove(p)
+#             except Exception:
+#                 pass

app/jobs/transcribe_job.py

@@ -1,27 +1,37 @@
 from app.core.asr_engine import load_model, transcribe_file
-from app.services.text_normalizer import normalize_text
 from app.services.note_client import NoteServiceClient
+from app.services.nlp_postprocess import normalize_and_extract
+from app.services.summary_service import generate_summary
+from app.services.mindmap_service import generate_mindmap
 
 # This function will be run by RQ worker
 def transcribe_job(tmp_wav: str, note_id: str):
     model = load_model()
-    text = transcribe_file(model, tmp_wav, 30.0, 5.0)
+    raw_text = transcribe_file(model, tmp_wav, 30.0, 5.0)
+    nlp = asyncio.run(normalize_and_extract(raw_text))
+    normalized = nlp["normalized_text"]
+    keywords = nlp["keywords"]
+
+    summary = asyncio.run(generate_summary(normalized))
+    mindmap = asyncio.run(generate_mindmap(normalized))
+
+    note_service = NoteServiceClient()
+
     # normalize_text có thể là async, nhưng RQ chỉ chạy sync nên cần chạy event loop nếu cần
     import asyncio
-    if asyncio.iscoroutinefunction(normalize_text):
-        normalized = asyncio.run(normalize_text(text))
-    else:
-        normalized = normalize_text(text)
-    note_service = NoteServiceClient()
-    # Gửi transcript sang Note Service
-    note_service.save_transcript(
-        note_id=note_id,
-        raw_text=text,
-        normalized_text=normalized,
-        duration=None,
-        sample_rate=None,
-        chunks=None,
-        asr_model="PhoWhisper-base",
-        normalization_model="gemini-1.5"
+    asyncio.run(
+        note_service.save_transcript(
+            note_id=note_id,
+            raw_text=raw_text,
+            normalized_text=normalized,
+            keywords=keywords,
+            summary=summary,
+            mindmap=mindmap,
+            duration=None,
+            sample_rate=None,
+            chunks=None,
+            asr_model="PhoWhisper-base",
+            normalization_model="gemini-1.5",
+        )
     )
     return True

app/services/mindmap_service.py

@@ -0,0 +1,56 @@
+import asyncio, json
+from app.config.settings import GEMINI_API_KEY
+import google.generativeai as genai
+
+if GEMINI_API_KEY:
+    genai.configure(api_key=GEMINI_API_KEY)
+    _model = genai.GenerativeModel("gemini-pro")
+else:
+    _model = None
+
+
+async def generate_mindmap(text: str) -> dict:
+    if not _model:
+        return {}
+
+    prompt = f"""
+Bạn là chuyên gia tạo Sơ đồ tư duy. Hãy phân tích văn bản sau và tạo cấu trúc JSON Mindmap.
+Yêu cầu:
+1. Xác định Ý chính làm Root.
+2. Phân tách ý phụ thành nhánh con (tối đa 3 cấp).
+3. Nhãn (label) ngắn gọn (< 7 từ).
+4. Màu sắc (colorHex): Root="#6200EE", Con="#F59E2B", "#2ECF9A", "#2F9BFF".
+
+Cấu trúc JSON bắt buộc (Chỉ trả về JSON):
+{{
+    "root": {{
+        "label": "Chủ đề",
+        "colorHex": "#6200EE",
+        "children": [
+            {{
+                "label": "Ý 1",
+                "colorHex": "#F59E2B",
+                "children": []
+            }}
+        ]
+    }}
+}}
+
+Văn bản:
+{text}
+"""
+
+    loop = asyncio.get_event_loop()
+
+    def call():
+        r = _model.generate_content(prompt)
+        return r.text
+
+    raw = await loop.run_in_executor(None, call)
+
+    start = raw.find("{")
+    end = raw.rfind("}")
+    if start != -1 and end != -1:
+        return json.loads(raw[start:end+1])
+
+    return {}

app/services/nlp_postprocess.py

@@ -0,0 +1,71 @@
+from app.infra.redis_client import redis_client
+from app.utils.hashing import sha256
+from app.config.settings import GEMINI_API_KEY
+import google.generativeai as genai
+import asyncio
+import json
+
+CACHE_TTL = 60 * 60 * 24 * 3  # 3 days
+
+if GEMINI_API_KEY:
+    genai.configure(api_key=GEMINI_API_KEY)
+    _model = genai.GenerativeModel("gemini-pro")
+else:
+    _model = None
+
+
+async def normalize_and_extract(raw_text: str) -> dict:
+    """
+    return {
+      "normalized_text": "...",
+      "keywords": [...]
+    }
+    """
+    cache_key = f"nlp:{sha256(raw_text)}"
+    cached = redis_client.get(cache_key)
+    if cached:
+        return json.loads(cached)
+
+    prompt = f"""
+Bạn là một hệ thống Xử lý Hậu kỳ NLP (NLP Post-Processing) Tiếng Việt.
+
+Đầu vào là văn bản thô (raw transcript), có thể thiếu dấu câu và sai chính tả do nhận dạng giọng nói (ví dụ: 'ăn chứa' -> 'ăn chưa').
+
+Nhiệm vụ (Trả về JSON duy nhất):
+1. [ASR Correction & Punctuation]: Sửa lỗi chính tả ASR, thêm dấu câu, viết hoa chuẩn xác.
+
+Văn bản đầu vào: \"\"\"{raw_text}\"\"\"
+
+Cấu trúc JSON bắt buộc:
+{{
+    "normalizedText": "Văn bản đã sửa hoàn chỉnh...",
+    "keywords": ["Từ khóa 1", "Từ khóa 2", "..."]
+}}
+"""
+
+    result = {
+        "normalized_text": raw_text,
+        "keywords": []
+    }
+
+    if _model:
+        loop = asyncio.get_event_loop()
+
+        def call():
+            r = _model.generate_content(prompt)
+            return r.text
+
+        text = await loop.run_in_executor(None, call)
+
+        # clean JSON
+        start = text.find("{")
+        end = text.rfind("}")
+        if start != -1 and end != -1:
+            data = json.loads(text[start:end+1])
+            result = {
+                "normalized_text": data.get("normalizedText", raw_text),
+                "keywords": data.get("keywords", [])
+            }
+
+    redis_client.setex(cache_key, CACHE_TTL, json.dumps(result))
+    return result

app/services/note_client.py

@@ -18,6 +18,7 @@ class NoteServiceClient:
         )
     )
     async def save_transcript(self, note_id: str, raw_text: str, normalized_text: str,
+                              keywords: list, summary: str, mindmap: dict,
                               duration: float, sample_rate: int, chunks: list,
                               asr_model: str = "PhoWhisper-base",
                               normalization_model: str = "gemini-1.5"):
@@ -25,6 +26,9 @@ class NoteServiceClient:
         payload = {
             "raw_text": raw_text,
             "normalized_text": normalized_text,
+            "keywords": keywords,
+            "summary": summary,
+            "mindmap": mindmap,
             "duration": duration,
             "sample_rate": sample_rate,
             "chunks": chunks,

app/services/summary_service.py

@@ -0,0 +1,35 @@
+import asyncio
+from app.config.settings import GEMINI_API_KEY
+import google.generativeai as genai
+
+if GEMINI_API_KEY:
+    genai.configure(api_key=GEMINI_API_KEY)
+    _model = genai.GenerativeModel("gemini-pro")
+else:
+    _model = None
+
+
+async def generate_summary(text: str) -> str:
+    if not _model:
+        return ""
+
+    prompt = f"""
+    Bạn là chuyên gia tóm tắt. Hãy tóm tắt văn bản sau thành **một đoạn văn duy nhất**.
+    Yêu cầu:
+    1. Viết khoảng 3-5 câu, tổng hợp đầy đủ chủ đề và các ý chính.
+    2. Viết liền mạch, KHÔNG xuống dòng, KHÔNG dùng gạch đầu dòng hay đánh số.
+    3. Chỉ dựa trên thông tin được cung cấp, tuyệt đối KHÔNG tự thêm thông tin bên ngoài.
+    4. Trả về văn bản thuần (plain text).
+
+    Văn bản:
+    \"\"\"{text}\"\"\"
+    """
+
+    loop = asyncio.get_event_loop()
+
+    def call():
+        r = _model.generate_content(prompt)
+        return r.text.strip()
+
+    result = await loop.run_in_executor(None, call)
+    return result.replace("```", "").strip()