Spaces:
Sleeping
Sleeping
File size: 4,225 Bytes
a581a5b |
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 137 138 139 140 141 142 143 144 145 146 |
# api/store.py
from __future__ import annotations
import json
import os
from typing import Dict, Any, List, Optional
from pathlib import Path
from api.models import CourseDirectoryItem, Workspace, PersonContact, WorkspaceCourseRef, WorkspaceMember
DATA_DIR = Path(os.getenv("CLARE_DATA_DIR", "./data"))
COURSE_FILE = DATA_DIR / "course_directory.json"
WORKSPACE_FILE = DATA_DIR / "workspaces.json"
def _ensure_dir():
DATA_DIR.mkdir(parents=True, exist_ok=True)
def _read_json(path: Path) -> Optional[Any]:
if not path.exists():
return None
with path.open("r", encoding="utf-8") as f:
return json.load(f)
def _write_json(path: Path, obj: Any) -> None:
_ensure_dir()
with path.open("w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
def seed_if_missing() -> None:
"""
Provide sane defaults so HF can run out-of-the-box.
Replace these with your real course list/workspaces later.
"""
_ensure_dir()
if not COURSE_FILE.exists():
courses = [
CourseDirectoryItem(
id="course_ai_001",
name="Introduction to AI",
instructor=PersonContact(name="Dr. Smith", email="smith@uni.edu"),
teachingAssistant=PersonContact(name="Alice", email="alice@uni.edu"),
).model_dump()
]
_write_json(COURSE_FILE, {"items": courses})
if not WORKSPACE_FILE.exists():
workspaces = [
Workspace(
id="ws_group_ai_g3",
type="group",
category="course",
name="Group Alpha",
groupNo=3,
courseId="course_ai_001",
courseInfo=WorkspaceCourseRef(id="course_ai_001", name="Introduction to AI"),
members=[
WorkspaceMember(id="u1", name="Sarah", email="sarah@uni.edu", role="owner"),
WorkspaceMember(id="u2", name="Bob", email="bob@uni.edu", role="member"),
],
).model_dump()
]
_write_json(WORKSPACE_FILE, {"items": workspaces})
def load_courses() -> List[CourseDirectoryItem]:
seed_if_missing()
raw = _read_json(COURSE_FILE) or {}
items = raw.get("items", [])
out: List[CourseDirectoryItem] = []
for x in items:
out.append(CourseDirectoryItem(**x))
return out
def save_courses(courses: List[CourseDirectoryItem]) -> None:
_write_json(COURSE_FILE, {"items": [c.model_dump() for c in courses]})
def load_workspaces() -> List[Workspace]:
seed_if_missing()
raw = _read_json(WORKSPACE_FILE) or {}
items = raw.get("items", [])
out: List[Workspace] = []
for x in items:
out.append(Workspace(**x))
return out
def save_workspaces(workspaces: List[Workspace]) -> None:
_write_json(WORKSPACE_FILE, {"items": [w.model_dump() for w in workspaces]})
def enrich_workspace_course_info(workspaces: List[Workspace], courses: List[CourseDirectoryItem]) -> List[Workspace]:
"""
Ensure workspace.courseInfo has instructor/TA to satisfy CourseInfoSection fallback.
Frontend prefers availableCourses, but this makes fallback robust.
"""
by_id = {c.id.strip().lower(): c for c in courses}
by_name = {c.name.strip().lower(): c for c in courses}
enriched: List[Workspace] = []
for w in workspaces:
ws_course = w.courseInfo
hit: Optional[CourseDirectoryItem] = None
if w.courseId:
hit = by_id.get(w.courseId.strip().lower())
if not hit and ws_course and ws_course.id:
hit = by_id.get(ws_course.id.strip().lower())
if not hit and ws_course and ws_course.name:
hit = by_name.get(ws_course.name.strip().lower())
if hit:
w.courseId = w.courseId or hit.id
w.courseInfo = WorkspaceCourseRef(
id=hit.id,
name=hit.name,
instructor=hit.instructor,
teachingAssistant=hit.teachingAssistant,
)
else:
# keep whatever exists; CourseInfoSection will show fallback if nothing matches
w.courseInfo = ws_course
enriched.append(w)
return enriched
def rename_workspace(workspace_id: str, new_name: str) -> Workspace:
workspaces = load_workspaces()
found = None
for w in workspaces:
if w.id == workspace_id:
w.name = new_name
found = w
break
if not found:
raise KeyError(f"workspace not found: {workspace_id}")
save_workspaces(workspaces)
return found
|