Spaces:
Sleeping
Sleeping
File size: 3,121 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | import re
from fastapi import APIRouter, BackgroundTasks, HTTPException
from pydantic import BaseModel
from app.schemas.annotation import StudentAnnotation
from app.services.annotation_service import get_annotation_service
from app.services.project_service import ProjectService
router = APIRouter(prefix="/annotations", tags=["annotations"])
_svc = get_annotation_service()
def _require_project_document(project_id: str, document_id: str) -> None:
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", project_id):
raise HTTPException(404, "Project document not found")
if not re.fullmatch(r"[0-9a-f]{64}", document_id):
raise HTTPException(404, "Project document not found")
project = ProjectService().load(project_id)
if project is None or not any(item.file_id == document_id for item in project.files):
raise HTTPException(404, "Project document not found")
class PatchNoteRequest(BaseModel):
note_text: str
def _refresh_annotation_evidence(project_id: str, document_id: str) -> None:
try:
from app.rag.evidence_ingestion import EvidenceIngestionService
EvidenceIngestionService().refresh_annotations(project_id, document_id)
except KeyError:
pass
@router.get("/stt-status")
def check_stt_status():
from app.services.voice_transcription_service import VoiceTranscriptionService
stt = VoiceTranscriptionService.get()
return {
"available": stt.is_available,
"is_loading": stt.is_model_loading()
}
@router.get("/{document_id}")
def list_annotations(document_id: str, project_id: str):
_require_project_document(project_id, document_id)
return [
annotation.model_dump()
for annotation in _svc.get_for_project_document(project_id, document_id)
]
@router.post("")
def create_annotation(annotation: StudentAnnotation, background_tasks: BackgroundTasks):
_require_project_document(annotation.project_id, annotation.document_id)
created = _svc.create(annotation)
background_tasks.add_task(
_refresh_annotation_evidence,
created.project_id,
created.document_id,
)
return created.model_dump()
@router.patch("/{annotation_id}")
def patch_annotation(annotation_id: str, req: PatchNoteRequest, background_tasks: BackgroundTasks):
updated = _svc.patch_note(annotation_id, req.note_text)
if not updated:
raise HTTPException(404, f"Annotation {annotation_id} not found")
background_tasks.add_task(
_refresh_annotation_evidence,
updated.project_id,
updated.document_id,
)
return updated.model_dump()
@router.delete("/{annotation_id}")
def delete_annotation(annotation_id: str, background_tasks: BackgroundTasks):
existing = _svc.get(annotation_id)
ok = _svc.delete(annotation_id)
if not ok:
raise HTTPException(404, f"Annotation {annotation_id} not found")
if existing:
background_tasks.add_task(
_refresh_annotation_evidence,
existing.project_id,
existing.document_id,
)
return {"status": "deleted"}
|