Spaces:
Sleeping
Sleeping
File size: 4,809 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 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """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
|