Spaces:
Sleeping
Sleeping
| """Annotation persistence -> in-memory dict + disk at ~/.studybuddy/annotations/{document_id}.json. | |
| Mirrors the session-persistence pattern in routers/session.py. | |
| """ | |
| import json | |
| import logging | |
| import os | |
| import re | |
| import time | |
| from typing import Dict, List, Optional | |
| from pydantic import ValidationError | |
| from app.schemas.annotation import StudentAnnotation | |
| _ANNOT_DIR = os.path.expanduser("~/.studybuddy/annotations") | |
| logger = logging.getLogger(__name__) | |
| class AnnotationService: | |
| def __init__(self) -> None: | |
| self._store: Dict[str, StudentAnnotation] = {} | |
| os.makedirs(_ANNOT_DIR, exist_ok=True) | |
| def _path(self, document_id: str) -> str: | |
| if not re.fullmatch(r"[0-9a-f]{64}", document_id): | |
| raise ValueError("document_id must be a SHA-256 hex digest") | |
| root = os.path.realpath(_ANNOT_DIR) | |
| path = os.path.realpath(os.path.join(root, f"{document_id}.json")) | |
| if os.path.commonpath([root, path]) != root: | |
| raise ValueError("annotation path escaped storage root") | |
| return path | |
| def _hydrate(self, document_id: str) -> None: | |
| """Load annotations for a document into memory if not already loaded.""" | |
| p = self._path(document_id) | |
| if not os.path.exists(p): | |
| return | |
| with open(p, encoding="utf-8") as f: | |
| items = json.load(f) | |
| changed = False | |
| for item in items: | |
| if not isinstance(item, dict): | |
| logger.warning("Skipping malformed annotation row for document %s", document_id) | |
| changed = True | |
| continue | |
| data = dict(item) | |
| if not data.get("document_id"): | |
| data["document_id"] = document_id | |
| changed = True | |
| if not data.get("project_id"): | |
| data["project_id"] = data.get("session_id") or "legacy" | |
| changed = True | |
| try: | |
| a = StudentAnnotation(**data) | |
| except ValidationError as exc: | |
| logger.warning( | |
| "Skipping invalid annotation %s for document %s: %s", | |
| data.get("annotation_id", "<unknown>"), | |
| document_id, | |
| exc, | |
| ) | |
| changed = True | |
| continue | |
| self._store.setdefault(a.annotation_id, a) | |
| if changed: | |
| self._flush(document_id) | |
| def _flush(self, document_id: str) -> None: | |
| rows = [a.model_dump() for a in self._store.values() if a.document_id == document_id] | |
| with open(self._path(document_id), "w", encoding="utf-8") as f: | |
| json.dump(rows, f, indent=2) | |
| def get_for_document(self, document_id: str) -> List[StudentAnnotation]: | |
| self._hydrate(document_id) | |
| return [a for a in self._store.values() if a.document_id == document_id] | |
| def get_for_project_document( | |
| self, | |
| project_id: str, | |
| document_id: str, | |
| ) -> List[StudentAnnotation]: | |
| return [ | |
| annotation | |
| for annotation in self.get_for_document(document_id) | |
| if annotation.project_id == project_id | |
| ] | |
| def get(self, annotation_id: str) -> Optional[StudentAnnotation]: | |
| return self._store.get(annotation_id) | |
| def create(self, annotation: StudentAnnotation) -> StudentAnnotation: | |
| self._store[annotation.annotation_id] = annotation | |
| self._flush(annotation.document_id) | |
| return annotation | |
| def patch_note(self, annotation_id: str, note_text: str) -> Optional[StudentAnnotation]: | |
| a = self._store.get(annotation_id) | |
| if not a: | |
| return None | |
| a.note_text = note_text | |
| a.updated_at = time.time() | |
| self._flush(a.document_id) | |
| return a | |
| def delete(self, annotation_id: str) -> bool: | |
| a = self._store.pop(annotation_id, None) | |
| if a: | |
| self._flush(a.document_id) | |
| return a is not None | |
| def delete_for_document(self, document_id: str) -> None: | |
| """Drop every annotation for a document -> in-memory and on disk.""" | |
| for aid in [a.annotation_id for a in self._store.values() if a.document_id == document_id]: | |
| self._store.pop(aid, None) | |
| p = self._path(document_id) | |
| if os.path.exists(p): | |
| os.remove(p) | |
| # Single process-wide instance -> every caller must go through get_annotation_service() | |
| # rather than constructing AnnotationService() directly, or in-memory state forks | |
| # (a write through one instance won't be visible to another until the next disk read). | |
| _instance: AnnotationService | None = None | |
| def get_annotation_service() -> AnnotationService: | |
| global _instance | |
| if _instance is None: | |
| _instance = AnnotationService() | |
| return _instance | |