Update server logic

app/api/notes/notes_audio.py

@@ -2,7 +2,7 @@ from fastapi import APIRouter, BackgroundTasks, HTTPException
 from pydantic import BaseModel
 from typing import Dict, List, Optional
 from app.services.note_store import create_note
-from app.jobs.enrichment_job import enrich_note
+from app.jobs.async_enrichment_job import enrich_note
 from app.models.enums import NoteType, NoteStatus
 from app.utils.time import now_ts
 
@@ -18,10 +18,14 @@ class CreateAudioNoteRequest(BaseModel):
 
     
 @router.post("/audio")
-async def create_audio_note(req: CreateAudioNoteRequest, bg: BackgroundTasks):
+async def create_audio_note(req: CreateAudioNoteRequest, bg: BackgroundTasks): 
     now = now_ts()
     has_enrichment = bool(req.generate)
 
+    allowed = {"normalize", "keywords", "summary", "mindmap"}
+    if not set(req.generate).issubset(allowed):
+        raise HTTPException(400, "Invalid generate task, generate list must not be empty")
+   
     note = {
         "note_id": req.note_id,
         "type": NoteType.audio,
@@ -32,6 +36,15 @@ async def create_audio_note(req: CreateAudioNoteRequest, bg: BackgroundTasks):
         "updated_at": now,
     }
 
+    if has_enrichment:
+        note.update({
+            "normalized_text": None,
+            "title": None,
+            "keywords": None,
+            "summary": None,
+            "mindmap": None,
+        })
+
     create_note(note)
 
     if has_enrichment:

app/api/notes/notes_events.py

@@ -0,0 +1,85 @@
+import asyncio
+import json
+import logging
+from fastapi import APIRouter
+from sse_starlette.sse import EventSourceResponse
+
+from app.services.note_store import get_note
+from app.models.enums import NoteStatus
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+@router.get("/notes/{note_id}/events")
+async def note_events(note_id: str):
+    """
+    Server-Sent Events endpoint:
+    - Push note.status whenever it changes
+    - Close stream when status == ready or error
+    """
+
+    async def event_generator():
+        last_status = None
+
+        try:
+            while True:
+                note = get_note(note_id)
+
+                if not note:
+                    yield {
+                        "event": "error",
+                        "data": json.dumps({
+                            "note_id": note_id,
+                            "status": "not_found",
+                        }),
+                    }
+                    break
+
+                status = note.get("status")
+
+                # Push only when status changes
+                if status != last_status:
+                    yield {
+                        "event": "status",
+                        "data": json.dumps({
+                            "note_id": note_id,
+                            "status": status,
+                        }),
+                    }
+                    last_status = status
+
+                # Terminal states → close SSE
+                if status in (
+                    NoteStatus.ready,
+                    NoteStatus.error,
+                ):
+                    break
+
+                # Avoid hammering Firestore
+                await asyncio.sleep(1.0)
+
+        except asyncio.CancelledError:
+            # Client disconnected (normal)
+            logger.info(
+                "[notes_events] SSE client disconnected note_id=%s",
+                note_id,
+            )
+        except Exception as e:
+            logger.exception(
+                "[notes_events] SSE error note_id=%s: %s",
+                note_id,
+                e,
+            )
+            yield {
+                "event": "error",
+                "data": json.dumps({
+                    "note_id": note_id,
+                    "status": "internal_error",
+                }),
+            }
+
+    return EventSourceResponse(
+        event_generator(),
+        ping=15,   # keep-alive, tránh proxy/nginx kill connection
+    )

app/api/notes/notes_regenerate.py

@@ -4,12 +4,11 @@ from fastapi import APIRouter, BackgroundTasks, HTTPException
 from pydantic import BaseModel
 from typing import List
 from app.services.note_store import get_note, update_note
-from app.jobs.enrichment_job import enrich_note
+from app.jobs.async_enrichment_job import enrich_note
 
 
 router = APIRouter(prefix="/notes", tags=["notes"])
 
-
 class RegenerateRequest(BaseModel):
     generate: List[str]
 
@@ -19,10 +18,19 @@ def regenerate_note(note_id: str, req: RegenerateRequest, bg: BackgroundTasks):
     note = get_note(note_id)
     if not note:
         raise HTTPException(404, "Note not found")
+    
+    allowed = {"normalize", "keywords", "summary", "mindmap"}
+    if not set(req.generate).issubset(allowed):
+        raise HTTPException(400, "Invalid generate task, generate list must not be empty")
 
     # mark processing immediately
     update_note(note_id, {
         "status": NoteStatus.processing,
+        "normalized_text": None,
+        "title": None,
+        "keywords": None,
+        "summary": None,
+        "mindmap": None,
         "updated_at": now_ts(),
     })
 

app/api/notes/notes_text.py

@@ -2,7 +2,7 @@ from fastapi import APIRouter, BackgroundTasks
 from pydantic import BaseModel
 from typing import List, Optional
 from app.services.note_store import create_note
-from app.jobs.enrichment_job import enrich_note
+from app.jobs.async_enrichment_job import enrich_note
 from app.models.enums import NoteType, NoteStatus
 from app.utils.id import new_id
 from app.utils.time import now_ts

app/jobs/async_enrichment_job.py

@@ -0,0 +1,74 @@
+import logging
+from app.services.enrichment.mindmap import generate_mindmap
+from app.services.enrichment.normalize import normalize_text
+from app.services.enrichment.summary import generate_summary
+from app.services.enrichment.title_keywords import extract_title_and_keywords
+from app.services.note_store import get_note, update_note
+from app.services.enrichment.batch_pipeline import run_pipeline
+from app.models.enums import NoteStatus
+from app.utils.time import now_ts
+
+logger = logging.getLogger(__name__)
+
+
+async def enrich_note(note_id: str, tasks: list[str]):
+    note = get_note(note_id)
+    if not note:
+        return
+
+    update_note(note_id, {
+        "status": NoteStatus.processing,
+        "updated_at": now_ts(),
+    })
+
+    try:
+        # 1️⃣ Normalize
+        if "normalize" in tasks:
+            text = await normalize_text(note.get("raw_text", ""))
+            update_note(note_id, {
+                "normalized_text": text,
+                "status": NoteStatus.normalize_done,
+                "updated_at": now_ts(),
+            })
+        else:
+            text = note.get("raw_text", "")
+
+        # 2️⃣ Title + Keywords
+        if "keywords" in tasks:
+            title, keywords = await extract_title_and_keywords(text)
+            update_note(note_id, {
+                "title": title,
+                "keywords": keywords,
+                "status": NoteStatus.keywords_done,
+                "updated_at": now_ts(),
+            })
+
+        # 3️⃣ Summary
+        if "summary" in tasks:
+            summary = await generate_summary(text)
+            update_note(note_id, {
+                "summary": summary,
+                "status": NoteStatus.summary_done,
+                "updated_at": now_ts(),
+            })
+
+        # 4️⃣ Mindmap
+        if "mindmap" in tasks:
+            mindmap = await generate_mindmap(text)
+            update_note(note_id, {
+                "mindmap": mindmap,
+                "status": NoteStatus.mindmap_done,
+                "updated_at": now_ts(),
+            })
+
+        update_note(note_id, {
+            "status": NoteStatus.ready,
+            "updated_at": now_ts(),
+        })
+
+    except Exception:
+        logger.exception("Enrichment failed note_id=%s", note_id)
+        update_note(note_id, {
+            "status": NoteStatus.error,
+            "updated_at": now_ts(),
+        })

app/jobs/enrichment_job.py

@@ -1,54 +0,0 @@
-import logging
-from app.services.note_store import get_note, update_note
-from app.services.enrichment.pipeline import run_pipeline
-from app.models.enums import NoteStatus
-from app.utils.time import now_ts
-
-logger = logging.getLogger(__name__)
-
-
-async def enrich_note(note_id: str, tasks: list[str]):
-    note = get_note(note_id)
-    if not note:
-        return
-
-    # Mark as processing
-    update_note(
-        note_id,
-        {
-            "status": NoteStatus.processing,
-            "updated_at": now_ts(),
-        },
-    )
-
-    try:
-        # Run NLP pipeline (mutates a copy of note)
-        enriched = await run_pipeline(note, tasks)
-
-        updates = {}
-
-        # Only persist known enrichment fields
-        for field in (
-            "title",
-            "normalized_text",
-            "keywords",
-            "summary",
-            "mindmap",
-        ):
-            if field in enriched:
-                updates[field] = enriched[field]
-
-        updates["status"] = NoteStatus.ready
-        updates["updated_at"] = now_ts()
-
-        update_note(note_id, updates)
-
-    except Exception:
-        logger.exception("Enrichment failed note_id=%s", note_id)
-        update_note(
-            note_id,
-            {
-                "status": NoteStatus.error,
-                "updated_at": now_ts(),
-            },
-        )

app/main.py

@@ -1,5 +1,5 @@
 from fastapi import FastAPI
-from app.api.notes import notes_text, notes_audio, notes_get, notes_update, notes_regenerate
+from app.api.notes import notes_text, notes_audio, notes_get, notes_update, notes_regenerate, notes_events
 from app.api.folders import folders_create, folders_get, folders_update, folders_delete
 
 app = FastAPI(title="Note Service API")
@@ -10,6 +10,7 @@ app.include_router(notes_audio.router)
 app.include_router(notes_get.router)
 app.include_router(notes_update.router)
 app.include_router(notes_regenerate.router)
+app.include_router(notes_events.router)
 
 # Folders
 app.include_router(folders_create.router)

app/models/enums.py

@@ -8,4 +8,8 @@ class NoteStatus(str, Enum):
     created = "created"
     processing = "processing"
     ready = "ready"
-    error = "error"
+    error = "error"
+    normalize_done = "normalize_done"
+    keywords_done = "keywords_done"
+    summary_done = "summary_done"
+    mindmap_done = "mindmap_done"

app/services/enrichment/{pipeline.py → batch_pipeline.py}

@@ -2,7 +2,7 @@ from app.services.enrichment.normalize import normalize_text
 from app.services.enrichment.title_keywords import extract_title_and_keywords
 from app.services.enrichment.summary import generate_summary
 from app.services.enrichment.mindmap import generate_mindmap
-
+# dùng cho regenerate / batch
 async def run_pipeline(note: dict, tasks: list[str]):
     raw_text = note.get("raw_text") or ""
     if not raw_text.strip():

requirements.txt

@@ -3,3 +3,4 @@ uvicorn
 google-genai
 firebase-admin
 pydantic
+sse-starlette